Actually, it is not just a question of a write only data source. The issue is
that in my case (and I imagine this is true for others), the schema is not read
from the database but is understood from the options. This means that I have no
way of understanding the schema without supplying the read options. On the
other hand, when writing, I have the schema from the dataframe.
I know the data source V2 API is considered experimental API and I have no
problem with it, however, this means that the change will require a change in
how the end user works with it (they suddenly need to add schema information
which they did not before), not to mention this being a regression.
As to the pull request, this only handles cases where the save mode is not
append, for the original example (having non existent path but have append will
still fail and according to the documentation of Append, if the path does not
exist it should create it).
I am currently having problem compiling everything so I can’t test it myself
but wouldn’t changing the relation definition in “save”:
val relation = DataSourceV2Relation.create(source, options, None,
Option(df.schema))
and changing create to look like this:
def create(source: DataSourceV2, options: Map[String, String], tableIdent:
Option[TableIdentifier] = None, userSpecifiedSchema: Option[StructType] =
None): DataSourceV2Relation = {
val schema = userSpecifiedSchema.getOrElse(source.createReader(options,
userSpecifiedSchema).readSchema())
val ident = tableIdent.orElse(tableFromOptions(options))
DataSourceV2Relation(
source, schema.toAttributes, options, ident, userSpecifiedSchema)
}
Correct this?
Or even creating a new create which simply gets the schema as non optional?
Thanks,
Assaf
From: Hyukjin Kwon [mailto:[email protected]]
Sent: Thursday, October 11, 2018 10:24 AM
To: Mendelson, Assaf; Wenchen Fan
Cc: dev
Subject: Re: Possible bug in DatasourceV2
[EXTERNAL EMAIL]
Please report any suspicious attachments, links, or requests for sensitive
information.
See https://github.com/apache/spark/pull/22688
+WEnchen, here looks the problem raised. This might have to be considered as a
blocker ...
On Thu, 11 Oct 2018, 2:48 pm assaf.mendelson,
<[email protected]<mailto:[email protected]>> wrote:
Hi,
I created a datasource writer WITHOUT a reader. When I do, I get an
exception: org.apache.spark.sql.AnalysisException: Data source is not
readable: DefaultSource
The reason for this is that when save is called, inside the source match to
WriterSupport we have the following code:
val source = cls.newInstance().asInstanceOf[DataSourceV2]
source match {
case ws: WriteSupport =>
val sessionOptions = DataSourceV2Utils.extractSessionConfigs(
source,
df.sparkSession.sessionState.conf)
val options = sessionOptions ++ extraOptions
--> val relation = DataSourceV2Relation.create(source, options)
if (mode == SaveMode.Append) {
runCommand(df.sparkSession, "save") {
AppendData.byName(relation, df.logicalPlan)
}
} else {
val writer = ws.createWriter(
UUID.randomUUID.toString, df.logicalPlan.output.toStructType,
mode,
new DataSourceOptions(options.asJava))
if (writer.isPresent) {
runCommand(df.sparkSession, "save") {
WriteToDataSourceV2(writer.get, df.logicalPlan)
}
}
}
but DataSourceV2Relation.create actively creates a reader
(source.createReader) to extract the schema:
def create(
source: DataSourceV2,
options: Map[String, String],
tableIdent: Option[TableIdentifier] = None,
userSpecifiedSchema: Option[StructType] = None): DataSourceV2Relation
= {
val reader = source.createReader(options, userSpecifiedSchema)
val ident = tableIdent.orElse(tableFromOptions(options))
DataSourceV2Relation(
source, reader.readSchema().toAttributes, options, ident,
userSpecifiedSchema)
}
This makes me a little confused.
First, the schema is defined by the dataframe itself, not by the data
source, i.e. it should be extracted from df.schema and not by
source.createReader
Second, I see that relation is actually only use if the mode is
SaveMode.append (btw this means if it is needed it should be defined inside
the "if"). I am not sure I understand the portion of the AppendData but why
would reading from the source be included?
Am I missing something here?
Thanks,
Assaf
--
Sent from: http://apache-spark-developers-list.1001551.n3.nabble.com/
---------------------------------------------------------------------
To unsubscribe e-mail:
[email protected]<mailto:[email protected]>
