Flink: Document watermark generation feature #9179

pvary · 2023-11-29T08:22:03Z

Documentation for #8553

docs/flink-queries.md

stevenzwu · 2023-11-29T17:18:33Z

docs/flink-queries.md

+### Emitting watermarks
+Emitting watermarks from the source itself could be beneficial for several purposes, like harnessing the
+[Flink Watermark Alignment](https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/dev/datastream/event-time/generating_watermarks/#watermark-alignment)
+feature to prevent runaway readers, or providing triggers for [Flink windowing](https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/dev/datastream/operators/windows/).


I don't think we should call out windows specifically for the benefit of emitting watermark from source itself. any event time and watermark strategy will have windows triggers

I would also remove prevent runaway readers, as it may not be very clear to users what "runaway readers" mean

I think it is very important to understand, that windowing and watermark generation based on records could cause surprising results - especially with batch reads, or in backfill situations. Without this feature there is not guarantee on the order of the files are read. Window triggering will only become reliable when the source controls the emitted watermarks.

I am not sure how detailed the description should be, but I think it is important to be noted here, so I am open for suggestions, if you think we should add more detail here.

This feature produces better watermark. But I think we don't have to explicitly say windowing here. If someone use low-level event timer triggers for stateful processing, this also helps.

docs/flink-queries.md

stevenzwu · 2023-11-29T17:26:21Z

docs/flink-queries.md

+        SOURCE_NAME,
+        TypeInformation.of(RowData.class));
+```
+
 ## Options


we also need to update the read options section.
https://iceberg.apache.org/docs/1.3.0/flink-configuration/#read-options

There is no corresponding read-option for this feature yet.

oh. then we would need to add them. cc @mas-chen

stevenzwu · 2023-11-29T17:27:21Z

@dchristle can you also help review this doc PR? your perspective can help improve the readability of the doc.

docs/flink-queries.md

mas-chen · 2023-11-29T18:37:24Z

docs/flink-queries.md

+            .set(FlinkReadOptions.SPLIT_FILE_OPEN_COST, String.valueOf(TableProperties.SPLIT_SIZE_DEFAULT))
+            // Watermark using long column
+            .watermarkColumn("long_column")
+            .watermarkTimeUnit(TimeUnit.MILLI_SCALE)


I'd include this in the previous example. I read this as a more advanced example as most users wouldn't need watermark alignment and so withTimestampAssigner could also be moved down here.

I would keep this 2 as a separate example.
If I understand correctly @stevenzwu thinks that the watermark alignment is the most important feature of this change, and @mas-chen thinks that the ordering / windowing is more important.

Probably this is a good indication that both benefits are important 😄

@mas-chen can you clarify your comment? I am not quite following.

@pvary it might be good to separate this into two code snippets. we can remove the two lines in the beginning.

StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(); TableLoader tableLoader = TableLoader.fromHadoopTable("hdfs://nn:8020/warehouse/path");

Oh I just think

.watermarkTimeUnit(TimeUnit.MILLI_SCALE)

should be advertised in the "basic" example. I think most people would just configure this, rather than the custom Timestamp assigner. This reduces code in the first example and keeps it simpler.

The 2nd example I consider as a more "advanced" example where we can show how to do the custom Timestamp assigner (and furthermore watermark alignment from the Flink perspective is an advanced feature--it requires lots of tuning and understanding of how it interacts with the watermark strategy--out of orderliness/idleness/etc).

@mas-chen .watermarkTimeUnit() is only needed for long type column, which we don't know what's the precision. the first example is Iceberg timestamp field which carries time unit inherently (currently only micro-second) and hence there is no need to ask user to set the time unit like the second example.

timestamp assigner is for Flink StreamRecord timestamp, it is related to watermark generation / advancement at all.

Thanks for the explantation, makes sense. Please disregard my comment!

docs/flink-queries.md

stevenzwu · 2023-11-30T20:03:01Z

docs/flink-queries.md

+        SOURCE_NAME,
+        TypeInformation.of(RowData.class));
+```
+
 ## Options


oh. then we would need to add them. cc @mas-chen

stevenzwu · 2023-11-30T20:05:40Z

docs/flink-queries.md

+            .set(FlinkReadOptions.SPLIT_FILE_OPEN_COST, String.valueOf(TableProperties.SPLIT_SIZE_DEFAULT))
+            // Watermark using long column
+            .watermarkColumn("long_column")
+            .watermarkTimeUnit(TimeUnit.MILLI_SCALE)


@mas-chen can you clarify your comment? I am not quite following.

@pvary it might be good to separate this into two code snippets. we can remove the two lines in the beginning.

StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(); TableLoader tableLoader = TableLoader.fromHadoopTable("hdfs://nn:8020/warehouse/path");

stevenzwu · 2023-11-30T20:08:17Z

docs/flink-queries.md

+StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
+TableLoader tableLoader = TableLoader.fromHadoopTable("hdfs://nn:8020/warehouse/path");
+
+// Ordered data file reads with windowing, using a timestamp column


there is no windowing in the snippet below. so may not be accurate to say it in the comment here.

If I understand this part correctly, it tries to demonstrate emit watermark from Iceberg source without enabling watermark alignment.

Changed the examples part... could you please check?

docs/flink-queries.md

stevenzwu · 2023-12-05T16:48:42Z

thanks @pvary for the documentation. thanks @mas-chen for the review

pvary · 2023-12-05T16:49:41Z

Thanks for the review and the merge @stevenzwu and @mas-chen !

Flink: Document watermark generation feature

feed5fd

github-actions bot added the docs label Nov 29, 2023

pvary requested a review from stevenzwu November 29, 2023 15:07

stevenzwu reviewed Nov 29, 2023

View reviewed changes

mas-chen reviewed Nov 29, 2023

View reviewed changes

Review comments

f2654b2

pvary force-pushed the water_doc branch from 0c6a37f to f2654b2 Compare November 30, 2023 12:44

stevenzwu reviewed Nov 30, 2023

View reviewed changes

Stats and examples

8a60078

stevenzwu approved these changes Dec 5, 2023

View reviewed changes

stevenzwu merged commit 8519224 into apache:main Dec 5, 2023
2 checks passed

pvary deleted the water_doc branch December 5, 2023 16:49

This was referenced Dec 11, 2023

streaming update jasonf20/iceberg#1

Closed

streaming update jasonf20/iceberg#2

Closed

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

Flink: Document watermark generation feature (apache#9179)

64a04c1

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

Flink: Document watermark generation feature (apache#9179)

46294f6

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Flink: Document watermark generation feature #9179

Flink: Document watermark generation feature #9179

pvary commented Nov 29, 2023

stevenzwu Nov 29, 2023

stevenzwu Nov 29, 2023

pvary Nov 30, 2023

stevenzwu Nov 30, 2023 •

edited

Loading

stevenzwu Nov 29, 2023

pvary Nov 30, 2023

stevenzwu Nov 30, 2023

stevenzwu commented Nov 29, 2023

mas-chen Nov 29, 2023

pvary Nov 30, 2023

stevenzwu Nov 30, 2023

mas-chen Nov 30, 2023

stevenzwu Nov 30, 2023

mas-chen Dec 1, 2023

stevenzwu Nov 30, 2023

stevenzwu Nov 30, 2023

stevenzwu Nov 30, 2023

pvary Dec 4, 2023

stevenzwu commented Dec 5, 2023

pvary commented Dec 5, 2023

Flink: Document watermark generation feature #9179

Flink: Document watermark generation feature #9179

Conversation

pvary commented Nov 29, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

stevenzwu Nov 30, 2023 • edited Loading

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

stevenzwu commented Nov 29, 2023

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

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

stevenzwu commented Dec 5, 2023

pvary commented Dec 5, 2023

stevenzwu Nov 30, 2023 •

edited

Loading