[core][spark] Deduplicate a replayed Structured Streaming micro-batch - #9667
Open
zhuxiangyi wants to merge 1 commit into
Open
[core][spark] Deduplicate a replayed Structured Streaming micro-batch#9667zhuxiangyi wants to merge 1 commit into
zhuxiangyi wants to merge 1 commit into
Conversation
Structured Streaming guarantees exactly-once only if the sink is idempotent for a repeated batch id: when a query fails between the sink returning from addBatch and Spark recording the batch as completed, the restarted query replays that micro-batch with its original batch id. PaimonSink received the batch id but only used it to pace full compaction, and committed through newBatchWriteBuilder(), whose commit user is a fresh random UUID and whose commit identifier is always Long.MAX_VALUE. Neither can identify a replay, so the whole batch was committed a second time, duplicating its rows in an append table. Commit every micro-batch under a commit user that survives a restart, and use filterAndCommit with the batch id as the commit identifier, so a replay that Paimon already committed is skipped. The commit user is derived from the checkpoint location, falling back to the query id Spark persists in the checkpoint metadata when the location does not reach the sink options, and write.stream.commit-user overrides both, as an option of the writer or as a spark.paimon.write.stream.commit-user session conf, like the read side takes its read.stream.* options. filterAndCommit verified that every file it is about to commit still exists, a check meant for a committable restored from an engine's state that may reference files deleted long ago. A caller filtering a committable it has just produced knows those files exist, so InnerTableCommit can now turn the check off, and the Spark sink does. Otherwise every micro-batch would pay a file listing proportional to the number of files it wrote. The data files of a skipped replay stay uncommitted and are reclaimed by orphan file cleaning. A postpone bucket table committing through the staged committer cannot deduplicate and logs a warning per micro-batch.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Purpose
Closes #9666.
Structured Streaming delivers exactly-once only if the sink is idempotent for a repeated batch id.
When a query fails after the sink returns from
addBatchbut before Spark records the batch ascompleted, the restarted query replays that micro-batch with its original batch id.
PaimonSinkreceived the batch id but used it only to pace full compaction, and committed throughtable.newBatchWriteBuilder(), whose commit user is a fresh random UUID per builder and whosecommit identifier is always
BatchWriteBuilder.COMMIT_IDENTIFIER = Long.MAX_VALUE. Neither of thedimensions Paimon deduplicates on could identify a replay, so the batch was committed a second
time: every row duplicated in an append-only table, and silently wrong values in an
aggregationmerge-engine table (writing
(1, 10)and replaying that batch yieldsv = 20).The machinery already exists in core and is what the Flink sink uses; the Spark sink simply took
the batch write path.
Tests
PaimonSinkIdempotencyTest(new, 7 cases), each asserting the correct behaviour so that it failswithout the fix:
entry of the batch, which is exactly the checkpoint state a driver failure leaves behind;
options because it comes from
spark.sql.streaming.checkpointLocation;completeoutput mode;write.stream.commit-useras an option of the writer and as aspark.paimon.session conf;addBatchcalled twice with the same batch id through the API directly.Two of them assert the prefix of the commit user recorded in the snapshot, so that the case which
is meant to exercise the query id derivation cannot pass through the checkpoint derivation.
The full set of Spark streaming suites was run on the spark3 and spark4 profiles (Spark 3.2, 3.4,
3.5, 4.1): 34 suites, 349 tests.
API and Format
BatchWriteBuilderImplgainswithCommitUser, and itsnewCommit()return type is narrowed fromBatchTableCommittoInnerTableCommit. The narrowing keeps the caller in the connector free of adowncast that could only fail at runtime; it is source compatible, and
BatchWriteBuilderImplhasno subclasses.
InnerTableCommitgainscheckFilesExistence(boolean).filterAndCommitverified that every fileit is about to commit still exists, which guards a committable restored from an engine's state that
may reference files deleted long ago. A caller filtering a committable it has just produced knows
those files exist, so the Spark sink turns the check off; otherwise every micro-batch would pay a
file listing proportional to the number of files it wrote. The default is unchanged, so Flink keeps
the check.
Documentation
docs/docs/spark/structured-streaming.mdgains an "Exactly-once" section covering how the commituser is derived, the new
write.stream.commit-useroption, and the limits: starting from a newcheckpoint location gives a query a new commit user, the data files of a skipped replay are left to
orphan file cleaning, and a postpone bucket table committing through the staged committer cannot
deduplicate and logs a warning per micro-batch. The generated option reference is regenerated.