[Iceberg CDC sink] Split late data and create commit windows - #40006
[Iceberg CDC sink] Split late data and create commit windows#40006ahmedabu98 wants to merge 2 commits into
Conversation
ahmedabu98
changed the title
|
Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment |
|
Assigning reviewers: R: @chamikaramj for label java. Note: If you would like to opt out of this review, comment Available commands:
The PR bot will only process comments in the main thread (not review comments). |
| */ | ||
| final class CommitWindows | ||
| extends PTransform< | ||
| PCollection<KV<KV<String, Integer>, KV<byte[], CdcRecord>>>, CommitWindows.Result> { |
There was a problem hiding this comment.
PCollection<KV<KV<String, Integer>, KV<byte[], CdcRecord>>> is pretty hard to follow.
Shall we introduce new types to simplify this ? For example,
ShardIdentifier -> KV<String, Integer>
SortKeyAndRecord -> KV<byte[], CdcRecord>
Alternatively we can use Beam Rows or AutoValue classes with schema here.
| return new Result(input.getPipeline(), sorted, deadLetter, deadLetterSchema); | ||
| } | ||
|
|
||
| /** Applies the commit-window assignment for the input's boundedness; see the class Javadoc. */ |
There was a problem hiding this comment.
for the input's boundedness -> based on whether input is bounded or not ?
|
|
||
| /** | ||
| * Splits late data in {@link CommitWindows}: every {@linkplain PaneInfo.Timing#LATE late} pane is | ||
| * diverted to a DLQ side output; on-time and early panes pass through unchanged. |
There was a problem hiding this comment.
We can't actually commit late data instead of sending to DLQ ? (probably reasonable but just to double check).
| Coder<?> valueCoder = ((KvCoder<?, ?>) inputCoder).getValueCoder(); | ||
| checkArgument( | ||
| valueCoder instanceof KvCoder, "expected a KvCoder input value coder, got %s", valueCoder); | ||
| Coder<?> recordCoder = ((KvCoder<?, ?>) valueCoder).getValueCoder(); |
There was a problem hiding this comment.
This kind of casts can be brittel and can be avoided by introducing new types as mentioned in the other comment I believe.
| onTimeUnsorted.apply( | ||
| "SortBySeqKind", | ||
| SortValues.create( | ||
| BufferedExternalSorter.options().withMemoryMB(config.getSorterMemoryMB()))); |
There was a problem hiding this comment.
This could lead to OOMs since we are sorting entire windows per shard. Let's make sure there's enough memory for the default case.
| "EventTimeWindows", | ||
| Window.<KV<KV<String, Integer>, KV<byte[], CdcRecord>>>into( | ||
| FixedWindows.of( | ||
| checkStateNotNull( |
2 participants
Adds the CDC sink's windowing stage:
Part of #39979
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, commentfixes #<ISSUE NUMBER>instead.CHANGES.mdwith noteworthy changes.See the Contributor Guide for more tips on how to make review process smoother.
To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md
GitHub Actions Tests Status (on master branch)
See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.