-
Notifications
You must be signed in to change notification settings - Fork 2.1k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Flink: Write watermark to the snapshot summary #6253
Conversation
@hililiwei, I really like the idea of having watermarks written. Can you explain in more detail what you're doing here and where the watermark comes from? |
hi @rdblue, thank you so much for your feedback. Actually, I want to solve a problem that is common in streaming scenarios: how does the downstream application know that all of the specified period data has been written if the application is not reading incremental data in real time, but is microbatching? Watermarking is an easy solution to think of. When downstream applications get the watermark, they can know whether the event time or process time of the data has reached a critical value. But it's not enough. If our community accepts this PR solution, I would like to do one more thing, which is to support time-based partition commit. In some scenarios, when a new partition is written, it is usually necessary to notify the downstream application. For example, When all the data for this partition is written, commit this partition to iceberg, just as flink does for hive\filesystem. I once participated in the development of this part of Flink, and I hope to introduce it to iceberg sink. Because of iceberg's snapshot management feature, we may be able to do better than hive\filesystem. The current iceberg flink sink can only commit based on checkpoint. When the time-based commit is complete, it will provides a partition commit feature that allows configuring custom policies. Commit actions are based on a combination of triggers and policies. Back to your question.
|
there is a previous attempt on writing the watermark (data completeness) to the Iceberg table: https://github.com/apache/iceberg/pull/2109/files. Let me summarize the questions/points. Would be great to gather some inputs. @rdblue @pvary @dixingxing0 @zhangjun0x01 @liubo1022126
|
@hililiwei time-based partition commit seems quite complicated. trying to understand its value. With watermark info to mark the data completeness, downstream can decide which partition (hourly or daily) has the complete data and it is ok to trigger the processing of the completed hour or day. |
Hi @stevenzwu
|
What do you mean round robin?
How could downstream consumers know when a new partition show up?
can you elaborate the differences btw watermark and new complete partition for downstream consumers? |
Hi @stevenzwu
I mean, downstream applications need to get the watermark of the table at intervals to determine whether to start processing. For example, get the latest watermark every five minutes.
The writer determines whether to commit new partitions to the table based on a combination of triggers and policies. Once a new partition is committed, it means that the writer considers the new partition's data ready. For details: Downstream applications read tables using incremental read scan. When a new snapshot (including a new partition and data) is commited by the upstream, the downstream can get it.
Take a scenario in our current production as an example. When my table is partitioned by hour, if the data in [12:00,13:00) is not completely written, I do not want consumers to see it when querying the table. The query action may be an ad hoc query or a statistics task. I want to commit the snapshot after all the data in [12:00, 13:00) is written instead of commting the snapshot at each checkpoint. Thx |
I think it represents different meanings. I include it in the summary because it does not represent the water mark for the table, but simply represents the current flink task that generated the snapshot. |
in our practice,If the flink task turns on unaligned checkpoints, it may cause the watermark to advance too much. |
This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@iceberg.apache.org list. Thank you for your contributions. |
This pull request has been closed due to lack of activity. This is not a judgement on the merit of the PR in any way. It is just a way of keeping the PR queue manageable. If you think that is incorrect, or the pull request requires review, you can revive the PR at any time. |
In some scenarios, the task needs to determine that all data of a certain period has been written based on the watermark.
The PR writes the watermark of the task to the snapshot summary, like the flink job id.