[SPARK-28650][SS][DOC] Correct explanation of guarantee for ForeachWriter #25407

HeartSaVioR · 2019-08-11T13:35:10Z

What changes were proposed in this pull request?

This patch modifies the explanation of guarantee for ForeachWriter as it doesn't guarantee same output for (partitionId, epochId). Refer the description of SPARK-28650 for more details.

Spark itself still guarantees same output for same epochId (batch) if the preconditions are met, 1) source is always providing the same input records for same offset request. 2) the query is idempotent in overall (indeterministic calculation like now(), random() can break this).

Assuming breaking preconditions as an exceptional case (the preconditions are implicitly required even before), we still can describe the guarantee with epochId, though it will be harder to leverage the guarantee: 1) ForeachWriter should implement a feature to track whether all the partitions are written successfully for given epochId 2) There's pretty less chance to leverage the fact, as the chance for Spark to successfully write all partitions and fail to checkpoint the batch is small.

Credit to @zsxwing on discovering the broken guarantee.

How was this patch tested?

This is just a documentation change, both on javadoc and guide doc.

…iter

SparkQA · 2019-08-11T16:55:19Z

Test build #108929 has finished for PR 25407 at commit 5ac50cb.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

dongjoon-hyun · 2019-08-11T16:58:37Z

sql/core/src/main/scala/org/apache/spark/sql/ForeachWriter.scala

- *     continuous mode, then this guarantee does not hold and therefore should not be used for
- *     deduplication.
+ * <li>Spark doesn't guarantee same output for (partitionId, epochId) on failure, so deduplication
+ *     cannot be achieved with (partitionId, epochId). Refer SPARK-28650 for more details.


SPARK-28650 has only the following content, could we remove this Refer SPARK-28650 for more details by embedding this information?

But we can break this easily actually when restarting a query but a batch is re-run (e.g., upgrade Spark)
Source returns a different DataFrame that has a different partition number (e.g., we start to not create empty partitions in Kafka Source V2).
A new added optimization rule may change the number of partitions in the new run.
Change the file split size in the new run.
Since we cannot guarantee that the same (partitionId, epochId) has the same data. We should update the document for "ForeachWriter".

I'd inline a few of those examples of when it can occur, and leave it at that.

While we would want to warn for existing ForeachWriter users (this can't be done in javadoc - it should be noticed via release note, streaming doc guide - maybe?, etc), not all end users would like to know the details. (I'd expect there would be more users ignoring the details than trying to understand all the details.) As they could still get the details from visiting the issue SPARK-28650, I'd borrow some example as e.g. from there and leave reference.

dongjoon-hyun · 2019-08-11T16:59:17Z

Could you review and sign-off this please, @zsxwing ?

srowen

Yes, sounds like the docs need an update. Is there any place in the user docs that this kind of gotcha should be documented? Or is it too niche?

srowen · 2019-08-11T17:16:27Z

sql/core/src/main/scala/org/apache/spark/sql/ForeachWriter.scala

- *     continuous mode, then this guarantee does not hold and therefore should not be used for
- *     deduplication.
+ * <li>Spark doesn't guarantee same output for (partitionId, epochId) on failure, so deduplication
+ *     cannot be achieved with (partitionId, epochId). Refer SPARK-28650 for more details.


I'd inline a few of those examples of when it can occur, and leave it at that.

srowen · 2019-08-11T17:17:40Z

sql/core/src/main/scala/org/apache/spark/sql/ForeachWriter.scala

+ * <li>Spark doesn't guarantee same output for (partitionId, epochId) on failure, so deduplication
+ *     cannot be achieved with (partitionId, epochId). Refer SPARK-28650 for more details.
+ *
+ *     You can still apply deduplication on `epochId`, but there's less benefit to leverage this,


Nit: I'd keep the voice consistent. "epochId can still be used for deduplication .." instead of "you can ..."
Can we clarify if there are cases where deduplication on both is still valid?

Nit: I'd keep the voice consistent. "epochId can still be used for deduplication .." instead of "you can ..."

I'll do the update.

Can we clarify if there are cases where deduplication on both is still valid?

As you could see the cases in SPARK-28650, Spark and source can break this guarantee which end users would be not easy to determine. Guarantee can still be broken even end users don't change the query, so I'd rather not enumerating the cases and let end users encounter the odd cases. What do you think about this @zsxwing ?

HeartSaVioR · 2019-08-11T21:46:52Z

Yes, sounds like the docs need an update. Is there any place in the user docs that this kind of gotcha should be documented? Or is it too niche?

I found same explanation is placed in structured streaming guide doc - I'll modify it as well. I'd emphasize this (release note, etc.) as this is changing the guarantee and end users may have to change their implementation of ForeachWriter.

HeartSaVioR · 2019-08-11T22:00:02Z

I guess it's not only applied to Spark 3.0 but applied to all the versions. (may worth to port back) End users may need to be noticed even they don't upgrade their Spark version, as they need to revisit their implementation of ForeachWriter.

SparkQA · 2019-08-12T01:17:34Z

Test build #108941 has finished for PR 25407 at commit 041a3a4.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

docs/structured-streaming-programming-guide.md

zsxwing · 2019-08-12T17:18:11Z

docs/structured-streaming-programming-guide.md

+  See [SPARK-28650](https://issues.apache.org/jira/browse/SPARK-28650) for more details. `epochId` can still be used
+  for deduplication, but there's less benefit to leverage this, as the chance for Spark to successfully write all
+  partitions and fail to checkpoint the batch is small. You also need to care about whether epoch is fully written,
+  via ensuring all partitions for the epochId are written successfully.


ensuring all partitions for the epochId are written successfully.

Hm, looks like ForeachWriter doesn't know the number of partitions, so it cannot implement something like this.

Oh you're right. There's no context around this unless end users do some kind of hacks (aggregating over couple of batches) and it's not worth to do with such hard hacks. I'll just get rid of guarantee for deduplication on epochId.

SparkQA · 2019-08-12T18:33:36Z

Test build #108981 has finished for PR 25407 at commit b6681bd.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

…ealize

HeartSaVioR · 2019-08-13T08:19:31Z

I guess I addressed all review comments. Could we please take another round of reviews?

HyukjinKwon

Looks fine but let me leave it to @zsxwing

SparkQA · 2019-08-13T11:43:45Z

Test build #109030 has finished for PR 25407 at commit 050e178.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

HeartSaVioR · 2019-08-16T06:15:57Z

@zsxwing kindly reminder.

srowen · 2019-08-19T16:04:38Z

@zsxwing if there are no more comments I think we can merge.

zsxwing · 2019-08-19T22:42:48Z

docs/structured-streaming-programming-guide.md

-  Hence, (partition_id, epoch_id) can be used to deduplicate and/or transactionally commit 
-  data and achieve exactly-once guarantees. However, if the streaming query is being executed 
-  in the continuous mode, then this guarantee does not hold and therefore should not be used for deduplication.
+- **Note:** Spark does not guarantee same output for (partitionId, epochId) on failure, so deduplication


nit: remove on failure. If a user stops a query, we may re-run a batch. I would not call this case as a failure. In addition, I think we can suggest users to use foreachBatch here if they needs deduplication.

We should also update the above table from exactly-once to at-least-once.

Great point! Updated.

Btw, while I'm modifying the fault-tolerant of Foreach Sink from Depends on the implementation to Yes (at-least-once) as well, your screenshot seems to point out File Sink. Doesn't it guarantee exactly-once for corresponding Spark query via File Sink's specific metadata? I guess FileStreamSinkLog guarantees unique write per batch. If that's not the case and you've found another broken fault-tolerance for File Sink, I feel it would be nice to have another JIRA (at least another PR) to track them separately, with description of new finding.

Sorry. I took a wrong screenshot.

Oh OK. Never mind.

zsxwing · 2019-08-19T22:43:00Z

sql/core/src/main/scala/org/apache/spark/sql/ForeachWriter.scala

- *     and achieve exactly-once guarantees. However, if the streaming query is being executed in the
- *     continuous mode, then this guarantee does not hold and therefore should not be used for
- *     deduplication.
+ * <li>Spark doesn't guarantee same output for (partitionId, epochId) on failure, so deduplication


SparkQA · 2019-08-20T03:18:27Z

Test build #109371 has finished for PR 25407 at commit 48e908f.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

zsxwing · 2019-08-20T07:56:01Z

LGTM. Merging to master and 2.4.

…iter # What changes were proposed in this pull request? This patch modifies the explanation of guarantee for ForeachWriter as it doesn't guarantee same output for `(partitionId, epochId)`. Refer the description of [SPARK-28650](https://issues.apache.org/jira/browse/SPARK-28650) for more details. Spark itself still guarantees same output for same epochId (batch) if the preconditions are met, 1) source is always providing the same input records for same offset request. 2) the query is idempotent in overall (indeterministic calculation like now(), random() can break this). Assuming breaking preconditions as an exceptional case (the preconditions are implicitly required even before), we still can describe the guarantee with `epochId`, though it will be harder to leverage the guarantee: 1) ForeachWriter should implement a feature to track whether all the partitions are written successfully for given `epochId` 2) There's pretty less chance to leverage the fact, as the chance for Spark to successfully write all partitions and fail to checkpoint the batch is small. Credit to zsxwing on discovering the broken guarantee. ## How was this patch tested? This is just a documentation change, both on javadoc and guide doc. Closes #25407 from HeartSaVioR/SPARK-28650. Authored-by: Jungtaek Lim (HeartSaVioR) <kabhwan@gmail.com> Signed-off-by: Shixiong Zhu <zsxwing@gmail.com> (cherry picked from commit b37c8d5) Signed-off-by: Shixiong Zhu <zsxwing@gmail.com>

HeartSaVioR · 2019-08-20T08:50:38Z

Thanks for the quick review and merge!

dongjoon-hyun · 2019-08-20T16:14:43Z

Thank you, @HeartSaVioR , @zsxwing , @srowen , @HyukjinKwon . It's nice to have this in 2.4.4 documentation.

…iter # What changes were proposed in this pull request? This patch modifies the explanation of guarantee for ForeachWriter as it doesn't guarantee same output for `(partitionId, epochId)`. Refer the description of [SPARK-28650](https://issues.apache.org/jira/browse/SPARK-28650) for more details. Spark itself still guarantees same output for same epochId (batch) if the preconditions are met, 1) source is always providing the same input records for same offset request. 2) the query is idempotent in overall (indeterministic calculation like now(), random() can break this). Assuming breaking preconditions as an exceptional case (the preconditions are implicitly required even before), we still can describe the guarantee with `epochId`, though it will be harder to leverage the guarantee: 1) ForeachWriter should implement a feature to track whether all the partitions are written successfully for given `epochId` 2) There's pretty less chance to leverage the fact, as the chance for Spark to successfully write all partitions and fail to checkpoint the batch is small. Credit to zsxwing on discovering the broken guarantee. ## How was this patch tested? This is just a documentation change, both on javadoc and guide doc. Closes apache#25407 from HeartSaVioR/SPARK-28650. Authored-by: Jungtaek Lim (HeartSaVioR) <kabhwan@gmail.com> Signed-off-by: Shixiong Zhu <zsxwing@gmail.com> (cherry picked from commit b37c8d5) Signed-off-by: Shixiong Zhu <zsxwing@gmail.com>

[SPARK-28650][SS][DOC] Correct explanation of guarantee for ForeachWr…

5ac50cb

…iter

dongjoon-hyun added STRUCTURED STREAMING DOCUMENTATION labels Aug 11, 2019

dongjoon-hyun reviewed Aug 11, 2019

View reviewed changes

srowen reviewed Aug 11, 2019

View reviewed changes

HeartSaVioR added 2 commits August 12, 2019 06:51

Address review comments

cc6e48a

Also addressed the change to structured streaming guide doc

041a3a4

HyukjinKwon reviewed Aug 12, 2019

View reviewed changes

docs/structured-streaming-programming-guide.md Outdated Show resolved Hide resolved

HyukjinKwon reviewed Aug 12, 2019

View reviewed changes

docs/structured-streaming-programming-guide.md Outdated Show resolved Hide resolved

HyukjinKwon reviewed Aug 12, 2019

View reviewed changes

docs/structured-streaming-programming-guide.md Outdated Show resolved Hide resolved

HyukjinKwon reviewed Aug 12, 2019

View reviewed changes

docs/structured-streaming-programming-guide.md Outdated Show resolved Hide resolved

Address part of review comments

b6681bd

zsxwing reviewed Aug 12, 2019

View reviewed changes

Remove explanation of guarantee on epochId as it's super hacky to r…

050e178

…ealize

HyukjinKwon approved these changes Aug 13, 2019

View reviewed changes

zsxwing reviewed Aug 19, 2019

View reviewed changes

Address review comment

48e908f

asfgit closed this in b37c8d5 Aug 20, 2019

HeartSaVioR deleted the SPARK-28650 branch August 20, 2019 08:50

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-28650][SS][DOC] Correct explanation of guarantee for ForeachWriter #25407

[SPARK-28650][SS][DOC] Correct explanation of guarantee for ForeachWriter #25407

HeartSaVioR commented Aug 11, 2019 •

edited

SparkQA commented Aug 11, 2019

dongjoon-hyun Aug 11, 2019

srowen Aug 11, 2019

HeartSaVioR Aug 11, 2019

dongjoon-hyun commented Aug 11, 2019

srowen left a comment

srowen Aug 11, 2019

srowen Aug 11, 2019

HeartSaVioR Aug 11, 2019

HeartSaVioR commented Aug 11, 2019

HeartSaVioR commented Aug 11, 2019

SparkQA commented Aug 12, 2019

zsxwing Aug 12, 2019

HeartSaVioR Aug 13, 2019

SparkQA commented Aug 12, 2019

HeartSaVioR commented Aug 13, 2019

HyukjinKwon left a comment

SparkQA commented Aug 13, 2019

HeartSaVioR commented Aug 16, 2019

srowen commented Aug 19, 2019

zsxwing Aug 19, 2019

HeartSaVioR Aug 19, 2019 •

edited

zsxwing Aug 20, 2019

HeartSaVioR Aug 20, 2019

zsxwing Aug 19, 2019

SparkQA commented Aug 20, 2019

zsxwing commented Aug 20, 2019

HeartSaVioR commented Aug 20, 2019

dongjoon-hyun commented Aug 20, 2019

[SPARK-28650][SS][DOC] Correct explanation of guarantee for ForeachWriter #25407

[SPARK-28650][SS][DOC] Correct explanation of guarantee for ForeachWriter #25407

Conversation

HeartSaVioR commented Aug 11, 2019 • edited

What changes were proposed in this pull request?

How was this patch tested?

SparkQA commented Aug 11, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dongjoon-hyun commented Aug 11, 2019

srowen left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

HeartSaVioR commented Aug 11, 2019

HeartSaVioR commented Aug 11, 2019

SparkQA commented Aug 12, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Aug 12, 2019

HeartSaVioR commented Aug 13, 2019

HyukjinKwon left a comment

Choose a reason for hiding this comment

SparkQA commented Aug 13, 2019

HeartSaVioR commented Aug 16, 2019

srowen commented Aug 19, 2019

Choose a reason for hiding this comment

HeartSaVioR Aug 19, 2019 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Aug 20, 2019

zsxwing commented Aug 20, 2019

HeartSaVioR commented Aug 20, 2019

dongjoon-hyun commented Aug 20, 2019

HeartSaVioR commented Aug 11, 2019 •

edited

HeartSaVioR Aug 19, 2019 •

edited