Skip to content

[SPARK-33659][SS] Document the current behavior for DataStreamWriter.toTable API #30885

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
xuanyuanking wants to merge 3 commits into from
Closed

[SPARK-33659][SS] Document the current behavior for DataStreamWriter.toTable API #30885

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
fc9dd3a
add doc
xuanyuanking Dec 22, 2020
c158775
address comments
xuanyuanking Dec 23, 2020
eff4b9d
delete experimental tag
xuanyuanking Dec 24, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 10 additions & 1 deletion python/pyspark/sql/streaming.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1501,6 +1501,16 @@ def toTable(self, tableName, format=None, outputMode=None, partitionBy=None, que
A new table will be created if the table not exists. The returned
:class:`StreamingQuery` object can be used to interact with the stream.

Note, if the table does not exist, both V1 and V2 tables will respect the
Comment thread
xuanyuanking marked this conversation as resolved.
Outdated
partitioningColumns provided by ``partitionBy``. However, when the table exists, only V1
table will pass the partitioning info to the sink, V2 table will ignore the provided
partitioning info.

Similar to the above issue of partitioning, the new table created by this API lacks
functionality (e.g., customized properties, options, and serde info) on creating V2 tables.
Please create a table manually before the execution to avoid creating a table with
incomplete information.

.. versionadded:: 3.1.0

Parameters
Expand Down Expand Up @@ -1543,7 +1553,6 @@ def toTable(self, tableName, format=None, outputMode=None, partitionBy=None, que
... format='parquet',
... checkpointLocation='/tmp/checkpoint') # doctest: +SKIP
"""
# TODO(SPARK-33659): document the current behavior for DataStreamWriter.toTable API
self.options(**options)
if outputMode is not None:
self.outputMode(outputMode)
Expand Down
12 changes: 12 additions & 0 deletions sql/core/src/main/scala/org/apache/spark/sql/streaming/DataStreamWriter.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -301,12 +301,24 @@ final class DataStreamWriter[T] private[sql](ds: Dataset[T]) {
def start(): StreamingQuery = startInternal(None)

/**
* :: Experimental ::

@HyukjinKwon HyukjinKwon Dec 22, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Experimental -> Evolving?

@xuanyuanking xuanyuanking Dec 23, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Per the comment

spark/common/tags/src/main/java/org/apache/spark/annotation/Experimental.java

Lines 28 to 31 in cc23581

* NOTE: If there exists a Scaladoc comment that immediately precedes this annotation, the first
* line of the comment must be ":: Experimental ::" with no trailing blank line. This is because
* of the known issue that Scaladoc displays only either the annotation or the comment, whichever
* comes first.
:: Experimental :: is the tag for scaladoc.

@HeartSaVioR HeartSaVioR Dec 23, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Both annotations (experimental and evolving) provide different semantics, right? Adding the different tag would give more confusion as it's not clear whether this is experimental vs evolving. You're getting it from Experimental, not Evolving.

I'll reopen this.

@xuanyuanking xuanyuanking Dec 24, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Make sense, let me delete this experimental one, only keep evolving.

*
* Starts the execution of the streaming query, which will continually output results to the given
* table as new data arrives. A new table will be created if the table not exists. The returned

@viirya viirya Dec 22, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Here it documented a new table will be created if not existing, but later it also documents "Please create a table manually before the execution". It looks confusing, I think. Could we rephrase them together and give a more concrete description about table creation?

@viirya viirya Dec 22, 2020 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Maybe we have two small paragraphs for v1 and v2 table separately? E.g.

For v1 table, partitioning columns provided by `partitionBy` will be respected
no matter the table exists or not. A new table will be created if the table not exists.

For v2 table, `partitionBy` will be ignored if the table already exists. `partitionBy`
will be respected only if the v2 table does not exist. Besides, the v2 table created
by this API lacks some functionalities (e.g., customized properties, options, and serde info).
If you need them, please create the v2 table manually before the execution to avoid
creating a table with incomplete information.

@HeartSaVioR HeartSaVioR Dec 23, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

+1 to @viirya suggestion. My request was to describe the impact of options (mostly partitionBy) for matrix of v1 vs v2 and existing vs non-existing.

Separating the case of v1 vs v2 is more important, so the suggestion looks better.

@xuanyuanking xuanyuanking Dec 23, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for the rephrase, done in c158775

* [[StreamingQuery]] object can be used to interact with the stream.
*
* Note, if the table does not exist, both V1 and V2 tables will respect the partitioningColumns
Comment thread
xuanyuanking marked this conversation as resolved.
Outdated
* provided by `partitionBy`. However, when the table exists, only V1 table will pass the
* partitioning info to the sink, V2 table will ignore the provided partitioning info.
*
* Similar to the above issue of partitioning, the new table created by this API lacks
* functionality (e.g., customized properties, options, and serde info) on creating V2 tables.
* Please create a table manually before the execution to avoid creating a table with incomplete
* information.
*
* @since 3.1.0
*/
@Evolving
@throws[TimeoutException]
def toTable(tableName: String): StreamingQuery = {
this.tableName = tableName
Expand Down
6 changes: 3 additions & 3 deletions sql/core/src/test/scala/org/apache/spark/sql/streaming/test/DataStreamTableAPISuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -275,7 +275,7 @@ class DataStreamTableAPISuite extends StreamTest with BeforeAndAfter {
val tableName = "stream_test"
withTable(tableName) {
// The file written by batch will not be seen after the table was written by a streaming
// query. This is because we loads files from the metadata log instead of listing them
// query. This is because we load files from the metadata log instead of listing them
// using HDFS API.
Seq(4, 5, 6).toDF("value").write.format("parquet")
.option("path", dir.getCanonicalPath).saveAsTable(tableName)
Expand All @@ -289,7 +289,7 @@ class DataStreamTableAPISuite extends StreamTest with BeforeAndAfter {
val tableName = "stream_test"
withTable(tableName) {
// The file written by batch will not be seen after the table was written by a streaming
// query. This is because we loads files from the metadata log instead of listing them
// query. This is because we load files from the metadata log instead of listing them
// using HDFS API.
Seq(4, 5, 6).toDF("value").write.format("parquet").saveAsTable(tableName)

Expand All @@ -302,7 +302,7 @@ class DataStreamTableAPISuite extends StreamTest with BeforeAndAfter {
val tableName = "stream_test"
withTable(tableName) {
// The file written by batch will not be seen after the table was written by a streaming
// query. This is because we loads files from the metadata log instead of listing them
// query. This is because we load files from the metadata log instead of listing them
// using HDFS API.
Seq(4, 5, 6).toDF("value").write
.mode("append").format("parquet").save(dir.getCanonicalPath)
Expand Down