Flink: Emit watermarks from the IcebergSource #8553

pvary · 2023-09-12T15:56:46Z

There are several cases when it is worthwhile to emitt watermarks from the source.
In these cases we need to make sure that the emitted watermarks are in order.
If we do this correctly we could be use:

Windowing operators on the source
Watermark Alignment to throttle runaway sources

The PR has 2 main additions to the current IcebergSource:

Adding a new IcebergSourceRecordEmitter interface to do the actual record emission
- Default implementation which does only the record emission
- EventTimeExtractorRecordEmitter which emits the watermark for every split and emits the record too
Adding a new IcebergEventTimeExtractor interface to extract the split watermark, and the record timestamp

The solution uses the OrderedSplitAssignerFactory to order the splits based on the split watermark, and enhances the IcebergSource so the users could provide the eventTimeExtractor

...a/org/apache/iceberg/flink/source/eventtimeextractor/IcebergTimestampEventTimeExtractor.java

flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/source/IcebergSource.java

flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/source/split/SplitComparators.java

...c/test/java/org/apache/iceberg/flink/source/TestIcebergSourceFailoverEventTimeExtractor.java

flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/source/IcebergSource.java

jerqi · 2023-09-13T09:24:49Z

Minor typo about title

Emitt -> Emit

pvary · 2023-09-13T10:31:34Z

Minor typo about title
Emitt -> Emit

Fixed, thanks!

pvary · 2023-09-14T12:41:51Z

cc: @sundargates - you might be interested in this

flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/source/reader/RecordEmmitters.java

stevenzwu · 2023-09-25T15:59:09Z

Here is the design doc that discussed the proposal from this PR compared to a previously approach: https://docs.google.com/document/d/1JCF77SkeB8Z4h5pXRCRcMh4aiAVuBzHM5IwohLwNCto.

We are going with this approach due to simpler implementation and reuse of most of what Flink framework provided in terms of watermark alignment.

...java/org/apache/iceberg/flink/source/eventtimeextractor/EventTimeExtractorRecordEmitter.java

.../main/java/org/apache/iceberg/flink/source/eventtimeextractor/IcebergEventTimeExtractor.java

...a/org/apache/iceberg/flink/source/eventtimeextractor/IcebergTimestampEventTimeExtractor.java

...k/v1.17/flink/src/main/java/org/apache/iceberg/flink/source/reader/RecordEmitterFactory.java

flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/source/split/SplitComparators.java

...java/org/apache/iceberg/flink/source/eventtimeextractor/EventTimeExtractorRecordEmitter.java

flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/source/reader/RecordEmitters.java

...c/test/java/org/apache/iceberg/flink/source/TestIcebergSourceFailoverEventTimeExtractor.java

...java/org/apache/iceberg/flink/source/eventtimeextractor/EventTimeExtractorRecordEmitter.java

...src/main/java/org/apache/iceberg/flink/source/reader/IcebergTimestampWatermarkExtractor.java

...17/flink/src/main/java/org/apache/iceberg/flink/source/reader/SerializableRecordEmitter.java

...nk/src/main/java/org/apache/iceberg/flink/source/reader/WatermarkExtractorRecordEmitter.java

flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/source/split/SplitComparators.java

flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/source/IcebergSource.java

...nk/src/main/java/org/apache/iceberg/flink/source/reader/WatermarkExtractorRecordEmitter.java

...17/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergWatermarkExtractor.java

...java/org/apache/iceberg/flink/source/eventtimeextractor/EventTimeExtractorRecordEmitter.java

flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/source/IcebergSource.java

...src/main/java/org/apache/iceberg/flink/source/reader/IcebergTimestampWatermarkExtractor.java

...17/flink/src/main/java/org/apache/iceberg/flink/source/reader/SerializableRecordEmitter.java

...st/java/org/apache/iceberg/flink/source/TestIcebergSourceFailoverWithWatermarkExtractor.java

...k/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceWithWatermarkExtractor.java

pvary · 2023-10-11T14:54:17Z

Created #8803 to enable the possibility to avoid keeping all of the stats when creating the ScanTasks

pvary · 2023-11-15T12:17:00Z

@stevenzwu, @sundargates: Finally, I was able to get #8803 in. So I updated the PR, and the new versions allows to set a Timestamp column name as a watermark source column:

 IcebergSource.<RowData>builder()
        .tableLoader(sourceTableResource.tableLoader())
        .watermarkColumn("ts")

Please review!
Thanks!

stevenzwu

still need to check the unit test code. didn't quite get it on the first look

...17/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergWatermarkExtractor.java

flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/source/split/SplitComparators.java

...17/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergWatermarkExtractor.java

...k/src/main/java/org/apache/iceberg/flink/source/reader/TimestampBasedWatermarkExtractor.java

...nk/src/main/java/org/apache/iceberg/flink/source/reader/WatermarkExtractorRecordEmitter.java

flink/v1.17/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceFailover.java

...link/src/main/java/org/apache/iceberg/flink/source/reader/ColumnStatsWatermarkExtractor.java

...st/java/org/apache/iceberg/flink/source/TestIcebergSourceFailoverWithWatermarkExtractor.java

...k/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceWithWatermarkExtractor.java

pvary · 2023-11-22T10:53:28Z

...k/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceWithWatermarkExtractor.java

+            });
+
+    // Use static variable to collect the windows, since other solutions were flaky
+    windows.clear();


This is somewhat inelegant, but the windowed.collectAsync() solution was flaky. Failed once per ~1000 runs. Even though the window operator was called, and emitted the records (checked by logs), the results did not show up in the resultIterator. Maybe some Flink glitch?

So opted for a static ConcurrentMap instead to collect the generated windows.

I am not sure how the windows variable work here. won't it be serialized and shipped to the task manager?

If the windowing testing is flaky, I would be inclined to just remove this. I feel the throttling test method below this is more relevant and clear.

If I would use a non-static windows variable, then it will be serialized, and I could not use it in my test. That is why I had to use a static variable - which only works correctly in tests which runs everything in one JVM.

Flakiness is solved by this change.

stevenzwu

one more nit comment on Javadoc.

otherwise, LGTM

stevenzwu · 2023-11-22T17:10:23Z

flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/source/IcebergSource.java

@@ -429,6 +444,30 @@ public Builder<T> setAll(Map<String, String> properties) {
      return this;
    }

+    /**
+     * Emits watermarks once per split based on the file statistics for the given split. The


nit: based on the file statistics -> based on the min value of column statistics from file metadata

stevenzwu · 2023-11-22T17:12:38Z

flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/source/IcebergSource.java

@@ -453,6 +492,18 @@ public IcebergSource<T> build() {
        contextBuilder.project(FlinkSchemaUtil.convert(icebergSchema, projectedFlinkSchema));
      }

+      SerializableRecordEmitter<T> emitter = SerializableRecordEmitter.defaultEmitter();
+      if (watermarkColumn != null) {


I am fine with tabling this as a follow-up.

The default combining logic is fine for batch processing where ordering doesn't matter. for streaming read and watermark alignment, ordering is important as it affects data buffering in the flink state. That is precisely what watermark alignment intended to solve.

Flink watermark alignment has been designed mostly with unbounded splits (like Kafka) in mind. also within an unbounded Kafka split, records are FIFO. there is a natural order within a split.

@pvary not sure every users understand the internal details. At least, we can at least document this option of disabling combining for better ordering in the doc then. Then users can make an informed choice. @dchristle can probably also chime in here and help review the doc change in a follow up PR.

stevenzwu · 2023-11-22T17:16:03Z

...link/src/main/java/org/apache/iceberg/flink/source/reader/ColumnStatsWatermarkExtractor.java

+    this.eventTimeFieldId = field.fieldId();
+    this.eventTimeFieldName = eventTimeFieldName;
+    // Use the timeUnit only for Long columns.
+    this.timeUnit = typeID.equals(TypeID.LONG) ? timeUnit : TimeUnit.MICROSECONDS;


@pvary let's create an issue to follow up on the time unit when PR #9008 is merged

Thanks for the pointer!
Created #9137 to tackle this

stevenzwu · 2023-11-22T17:39:52Z

...k/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceWithWatermarkExtractor.java

+            });
+
+    // Use static variable to collect the windows, since other solutions were flaky
+    windows.clear();


I am not sure how the windows variable work here. won't it be serialized and shipped to the task manager?

If the windowing testing is flaky, I would be inclined to just remove this. I feel the throttling test method below this is more relevant and clear.

pvary · 2023-11-23T09:41:46Z

Merged to main.
Thanks @stevenzwu, @dchristle, @sundargates, @gyfora and @jerqi for the reviews!

pvary requested a review from stevenzwu September 12, 2023 15:56

github-actions bot added the flink label Sep 12, 2023

dchristle reviewed Sep 12, 2023

View reviewed changes

pvary changed the title ~~Flink: Emitt watermarks from the IcebergSource~~ Flink: Emit watermarks from the IcebergSource Sep 13, 2023

dchristle reviewed Sep 15, 2023

View reviewed changes

flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/source/reader/RecordEmmitters.java Outdated Show resolved Hide resolved

stevenzwu reviewed Sep 27, 2023

View reviewed changes

pvary force-pushed the watermark branch from 0e31d6a to 87046c7 Compare September 27, 2023 16:46

sundargates reviewed Sep 27, 2023

View reviewed changes

...nk/src/main/java/org/apache/iceberg/flink/source/reader/WatermarkExtractorRecordEmitter.java Show resolved Hide resolved

stevenzwu reviewed Sep 27, 2023

View reviewed changes

pvary mentioned this pull request Oct 9, 2023

Some questions about Iceberg's capabilities in Flink #8754

Closed

pvary mentioned this pull request Nov 8, 2023

Flink: Add support for Flink 1.18 #8930

Closed

Peter Vary added 7 commits November 15, 2023 10:53

Flink: Emitt watermarks from the IcebergSource

d592d27

Fix review comments

b2ff85f

Use RecordEmmitters instead of factories to simplify the code

7771586

Fix the emmitter

cf8e080

Steven's comments

d5226a7

Sundar's comments

7283b56

Small fixes

4f11bba

pvary force-pushed the watermark branch 2 times, most recently from 4e92111 to 372f93d Compare November 15, 2023 12:06

Column name based split orders

0734faf

pvary force-pushed the watermark branch from 372f93d to 0734faf Compare November 15, 2023 12:30

stevenzwu reviewed Nov 16, 2023

View reviewed changes

pvary force-pushed the watermark branch from 48d19dd to 3ccdcf1 Compare November 21, 2023 16:23

stevenzwu reviewed Nov 21, 2023

View reviewed changes

Final comments and flaky fix

3f823f3

pvary commented Nov 22, 2023

View reviewed changes

stevenzwu approved these changes Nov 22, 2023

View reviewed changes

Javadoc change

d6dc3a0

pvary merged commit 0831eb0 into apache:main Nov 23, 2023
45 checks passed

pvary deleted the watermark branch November 23, 2023 09:41

jonathf mentioned this pull request Nov 23, 2023

Questions related to watermarking of Iceberg source #9138

Closed

pvary pushed a commit to pvary/iceberg that referenced this pull request Nov 23, 2023

Flink: Backport apache#8553 to v1.15, v1.16

2bb8fc4

pvary mentioned this pull request Nov 23, 2023

Flink: Backport #8553 to v1.15, v1.16 #9139

Closed

pvary pushed a commit to pvary/iceberg that referenced this pull request Nov 23, 2023

Flink: Backport apache#8553 to v1.15, v1.16

67d5cdc

pvary pushed a commit to pvary/iceberg that referenced this pull request Nov 24, 2023

Flink: Backport apache#8553 to v1.15, v1.16

3b26b8b

pvary pushed a commit to pvary/iceberg that referenced this pull request Nov 24, 2023

Flink: Backport apache#8553 to v1.15, v1.16

5332a0a

pvary pushed a commit to pvary/iceberg that referenced this pull request Nov 24, 2023

Flink: Backport apache#8553 to v1.15, v1.16

4f1f615

pvary pushed a commit to pvary/iceberg that referenced this pull request Nov 24, 2023

Flink: Backport apache#8553 to v1.15, v1.16

b92ea39

pvary pushed a commit to pvary/iceberg that referenced this pull request Nov 24, 2023

Flink: Backport apache#8553 to v1.15, v1.16

27333e3

pvary pushed a commit to pvary/iceberg that referenced this pull request Nov 24, 2023

Flink: Backport apache#8553 to v1.15, v1.16

4471040

pvary pushed a commit to pvary/iceberg that referenced this pull request Nov 27, 2023

Flink: Backport apache#8553 to v1.15, v1.16

6d21701

pvary added a commit that referenced this pull request Nov 28, 2023

Flink: Backport #8553 to v1.15, v1.16 (#9145)

5e059c1

pvary mentioned this pull request Nov 29, 2023

Flink: Document watermark generation feature #9179

Merged

lisirrx pushed a commit to lisirrx/iceberg that referenced this pull request Jan 4, 2024

Flink: Backport apache#8553 to v1.15, v1.16 (apache#9145)

872fd62

mas-chen pushed a commit to mas-chen/iceberg that referenced this pull request Jan 9, 2024

Flink: Emit watermarks from the IcebergSource (apache#8553)

daf73d7

rodmeneses pushed a commit to rodmeneses/iceberg that referenced this pull request Feb 19, 2024

Flink: Emit watermarks from the IcebergSource (apache#8553)

d4c5e5f

rodmeneses added a commit to rodmeneses/iceberg that referenced this pull request Feb 19, 2024

Flink: Backport apache#8553 to v1.16 and v1.15 (apache#9145)

8c21d0a

pvary mentioned this pull request Mar 14, 2024

Is there any way on Flink to read newly appended data only (NOT in current Iceberg table snapshot)? #9955

Open

czy006 mentioned this pull request Mar 23, 2024

[Subtask]: Store watermark in the snapshot level apache/amoro#1721

Closed

3 tasks

devangjhabakh pushed a commit to cdouglas/iceberg that referenced this pull request Apr 22, 2024

Flink: Emit watermarks from the IcebergSource (apache#8553)

3d02c1d

devangjhabakh pushed a commit to cdouglas/iceberg that referenced this pull request Apr 22, 2024

Flink: Backport apache#8553 to v1.15, v1.16 (apache#9145)

125aefe

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Flink: Emit watermarks from the IcebergSource #8553

Flink: Emit watermarks from the IcebergSource #8553

pvary commented Sep 12, 2023

jerqi commented Sep 13, 2023

pvary commented Sep 13, 2023

pvary commented Sep 14, 2023

stevenzwu commented Sep 25, 2023

pvary commented Oct 11, 2023 •

edited

Loading

pvary commented Nov 15, 2023

stevenzwu left a comment

pvary Nov 22, 2023

stevenzwu Nov 22, 2023

pvary Nov 23, 2023

stevenzwu left a comment

stevenzwu Nov 22, 2023

stevenzwu Nov 22, 2023 •

edited

Loading

stevenzwu Nov 22, 2023

pvary Nov 23, 2023

stevenzwu Nov 22, 2023

pvary commented Nov 23, 2023

Flink: Emit watermarks from the IcebergSource #8553

Flink: Emit watermarks from the IcebergSource #8553

Conversation

pvary commented Sep 12, 2023

jerqi commented Sep 13, 2023

pvary commented Sep 13, 2023

pvary commented Sep 14, 2023

stevenzwu commented Sep 25, 2023

pvary commented Oct 11, 2023 • edited Loading

pvary commented Nov 15, 2023

stevenzwu left a comment

Choose a reason for hiding this comment

pvary Nov 22, 2023

Choose a reason for hiding this comment

stevenzwu Nov 22, 2023

Choose a reason for hiding this comment

pvary Nov 23, 2023

Choose a reason for hiding this comment

stevenzwu left a comment

Choose a reason for hiding this comment

stevenzwu Nov 22, 2023

Choose a reason for hiding this comment

stevenzwu Nov 22, 2023 • edited Loading

Choose a reason for hiding this comment

stevenzwu Nov 22, 2023

Choose a reason for hiding this comment

pvary Nov 23, 2023

Choose a reason for hiding this comment

stevenzwu Nov 22, 2023

Choose a reason for hiding this comment

pvary commented Nov 23, 2023

pvary commented Oct 11, 2023 •

edited

Loading

stevenzwu Nov 22, 2023 •

edited

Loading