You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by GitBox <gi...@apache.org> on 2021/09/16 10:16:44 UTC
[GitHub] [iceberg] liubo1022126 opened a new pull request #3130: Flink: support write watermark in snapshot
liubo1022126 opened a new pull request #3130:
URL: https://github.com/apache/iceberg/pull/3130
thanks @stevenzwu @rdblue
This PR is from Flink: support write watermark in snapshot #3093, based on the master and revert unrelated formatting change.
The current point of discussion is how to implement watermark, my idea is https://github.com/apache/iceberg/pull/3093#issuecomment-916904105, @stevenzwu @openinx How do you think?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] liubo1022126 commented on pull request #3130: Flink: support write watermark in snapshot
Posted by GitBox <gi...@apache.org>.
liubo1022126 commented on pull request #3130:
URL: https://github.com/apache/iceberg/pull/3130#issuecomment-927585109
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a change in pull request #3130: Flink: support write watermark in snapshot
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #3130:
URL: https://github.com/apache/iceberg/pull/3130#discussion_r719920777
##########
File path: core/src/main/java/org/apache/iceberg/TableProperties.java
##########
@@ -228,4 +228,10 @@ private TableProperties() {
public static final String MERGE_CARDINALITY_CHECK_ENABLED = "write.merge.cardinality-check.enabled";
public static final boolean MERGE_CARDINALITY_CHECK_ENABLED_DEFAULT = true;
+
+ public static final String WATERMARK_FIELD_NAME = "write.watermark.field";
+ public static final String WATERMARK_FIELD_NAME_DEFAULT = "";
+
+ public static final String WATERMARK_VALUE = "write.watermark";
Review comment:
In the past, we have 3 Flink streaming jobs (1 for each AWS region) writing to the same table. We need to write to 3 different watermark table properties (1 for each region). Watermark consumer then use the min value to determine the overall table watermark.
A provider pattern similar to `WRITE_LOCATION_PROVIDER_IMPL` can work. A default impl could be a single property name. not sure if there is a simpler way to achieve this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a change in pull request #3130: Flink: support write watermark in snapshot
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #3130:
URL: https://github.com/apache/iceberg/pull/3130#discussion_r728624766
##########
File path: core/src/main/java/org/apache/iceberg/TableProperties.java
##########
@@ -228,4 +228,10 @@ private TableProperties() {
public static final String MERGE_CARDINALITY_CHECK_ENABLED = "write.merge.cardinality-check.enabled";
public static final boolean MERGE_CARDINALITY_CHECK_ENABLED_DEFAULT = true;
+
+ public static final String WATERMARK_FIELD_NAME = "write.watermark.field";
+ public static final String WATERMARK_FIELD_NAME_DEFAULT = "";
+
+ public static final String WATERMARK_VALUE = "write.watermark";
Review comment:
> t's here, If we treat all physical jobs as a logical job, then each physical job is a parallelism in the logical job. So anyway, if we need to obtain information about different physical jobs, it is necessary to write different watermarks into table properties. I think what we can do a little more is to automatically aggregate different watermark values during the streaming job commit to get the final result.
This is an interesting idea definitely with metrics. Conceptually, it makes sense to me.
If the Flink sink also update the aggregated `write.watermark`, how does commit collision retry work? We can't move the watermark aggregation logic into core Iceberg commit retry logic. It seems that we have to implement commit retry logic in the Flink sink now due to this special handling on the aggregated watermark table property.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] liubo1022126 commented on pull request #3130: Flink: support write watermark in snapshot
Posted by GitBox <gi...@apache.org>.
liubo1022126 commented on pull request #3130:
URL: https://github.com/apache/iceberg/pull/3130#issuecomment-927826834
> @liubo1022126 I would add option 3: set the watermark as table property (not snapshot summary). This is what we did at Netflix before.
>
> * pro: Storing watermark in table properties make it easier for consumer
> * con: It needs `Table#newTransaction` to commit append and table properties update together. Transaction change might require bigger refactor in the IcebergFilesCommitter.
@stevenzwu ok, I'm agree with it.
what about mark it in both the snapshot of the append operation and the table properties. In append snapshot, we can get the ability of trend analysis. Mark the latest watermark in the table properties, user can use it very conveniently.
`private Long watermark = WATERMARK_VALUE_DEFAULT;`
yes, long seems better.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] liubo1022126 removed a comment on pull request #3130: Flink: support write watermark in snapshot
Posted by GitBox <gi...@apache.org>.
liubo1022126 removed a comment on pull request #3130:
URL: https://github.com/apache/iceberg/pull/3130#issuecomment-927585109
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] liubo1022126 commented on pull request #3130: Flink: support write watermark in snapshot
Posted by GitBox <gi...@apache.org>.
liubo1022126 commented on pull request #3130:
URL: https://github.com/apache/iceberg/pull/3130#issuecomment-924621782
thanks @coolderli , First of all, I agree with you.
But the calculation method of watermark is a combination of in-parallel and between-parallel. So if the iceberg-core is modified to support, will be more difficult.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] liubo1022126 commented on a change in pull request #3130: Flink: support write watermark in snapshot
Posted by GitBox <gi...@apache.org>.
liubo1022126 commented on a change in pull request #3130:
URL: https://github.com/apache/iceberg/pull/3130#discussion_r725449357
##########
File path: core/src/main/java/org/apache/iceberg/TableProperties.java
##########
@@ -228,4 +228,10 @@ private TableProperties() {
public static final String MERGE_CARDINALITY_CHECK_ENABLED = "write.merge.cardinality-check.enabled";
public static final boolean MERGE_CARDINALITY_CHECK_ENABLED_DEFAULT = true;
+
+ public static final String WATERMARK_FIELD_NAME = "write.watermark.field";
+ public static final String WATERMARK_FIELD_NAME_DEFAULT = "";
+
+ public static final String WATERMARK_VALUE = "write.watermark";
Review comment:
sorry, I just came back from holiday.
I think it’s a good idea to use different suffixes to distinguish different streaming jobs when writing watermark. Because job processing will not be blocked by other job problems. For example:
it's here, If we treat all physical jobs as a logical job, then each physical job is a parallelism in the logical job. So anyway, if we need to obtain information about different physical jobs, it is necessary to write different watermarks into table properties.
I think what we can do a little more is to automatically aggregate different watermark values during the streaming job commit to get the final result. For example:
That is to say, when using watermark value downstream, only need to use `write.watermark` directly, don't care about upstream logic.
**In code design, we may need to create a new class for the process of watermark, not simply write the watermark processing into IcebergFilesCommitter class like this pr. Divided into a simple watermark processing and multiple regions.**
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] coolderli commented on pull request #3130: Flink: support write watermark in snapshot
Posted by GitBox <gi...@apache.org>.
coolderli commented on pull request #3130:
URL: https://github.com/apache/iceberg/pull/3130#issuecomment-921706483
I mean users maybe don't know what kind of engine writes this table. For users, it is just a table and can know the data advancement at any time.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] liubo1022126 commented on a change in pull request #3130: Flink: support write watermark in snapshot
Posted by GitBox <gi...@apache.org>.
liubo1022126 commented on a change in pull request #3130:
URL: https://github.com/apache/iceberg/pull/3130#discussion_r725449357
##########
File path: core/src/main/java/org/apache/iceberg/TableProperties.java
##########
@@ -228,4 +228,10 @@ private TableProperties() {
public static final String MERGE_CARDINALITY_CHECK_ENABLED = "write.merge.cardinality-check.enabled";
public static final boolean MERGE_CARDINALITY_CHECK_ENABLED_DEFAULT = true;
+
+ public static final String WATERMARK_FIELD_NAME = "write.watermark.field";
+ public static final String WATERMARK_FIELD_NAME_DEFAULT = "";
+
+ public static final String WATERMARK_VALUE = "write.watermark";
Review comment:
sorry, I just came back from holiday.
I think it’s a good idea to use different suffixes to distinguish different streaming jobs when writing watermark. Because job processing will not be blocked by other job problems. For example:
it's here, If we treat all physical jobs as a logical job, then each physical job is a parallelism in the logical job. So anyway, if we need to obtain information about different physical jobs, it is necessary to write different watermarks into table properties.
I think what we can do a little more is to automatically aggregate different watermark values during the streaming job commit to get the final result. For example:
That is to say, when using watermark value downstream, only need to use `write.watermark` directly, don't care about upstream logic.
**In code design, we may need to create a new class for the process of watermark, not simply write the watermark processing into IcebergFilesCommitter like this pr. And divided into different implementations: simple watermark processing and multiple regions watermark processing.**
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a change in pull request #3130: Flink: support write watermark in snapshot
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #3130:
URL: https://github.com/apache/iceberg/pull/3130#discussion_r728624766
##########
File path: core/src/main/java/org/apache/iceberg/TableProperties.java
##########
@@ -228,4 +228,10 @@ private TableProperties() {
public static final String MERGE_CARDINALITY_CHECK_ENABLED = "write.merge.cardinality-check.enabled";
public static final boolean MERGE_CARDINALITY_CHECK_ENABLED_DEFAULT = true;
+
+ public static final String WATERMARK_FIELD_NAME = "write.watermark.field";
+ public static final String WATERMARK_FIELD_NAME_DEFAULT = "";
+
+ public static final String WATERMARK_VALUE = "write.watermark";
Review comment:
> t's here, If we treat all physical jobs as a logical job, then each physical job is a parallelism in the logical job. So anyway, if we need to obtain information about different physical jobs, it is necessary to write different watermarks into table properties. I think what we can do a little more is to automatically aggregate different watermark values during the streaming job commit to get the final result.
This is an interesting idea definitely with merits. Conceptually, it makes sense to me.
If the Flink sink also update the aggregated `write.watermark`, how does commit collision retry work? We can't move the watermark aggregation logic into core Iceberg commit retry logic. It seems that we have to implement commit retry logic in the Flink sink now due to this special handling on the aggregated watermark table property.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] coolderli commented on pull request #3130: Flink: support write watermark in snapshot
Posted by GitBox <gi...@apache.org>.
coolderli commented on pull request #3130:
URL: https://github.com/apache/iceberg/pull/3130#issuecomment-921702642
@liubo1022126 I have another question. How could spark or spark-streaming support the watermark? Could it be unified on the iceberg?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on pull request #3130: Flink: support write watermark in snapshot
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on pull request #3130:
URL: https://github.com/apache/iceberg/pull/3130#issuecomment-927527676
@liubo1022126 I would add option 3: set the watermark as table property (not snapshot summary). This is what we did at Netflix before.
* pro: Storing watermark in table properties make it easier for consumer
* con: It needs `Table#newTransaction` to commit append and table properties update together. Transaction change might require bigger refactor in the IcebergFilesCommitter.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] liubo1022126 removed a comment on pull request #3130: Flink: support write watermark in snapshot
Posted by GitBox <gi...@apache.org>.
liubo1022126 removed a comment on pull request #3130:
URL: https://github.com/apache/iceberg/pull/3130#issuecomment-927585644
> @liubo1022126 I would add option 3: set the watermark as table property (not snapshot summary). This is what we did at Netflix before.
>
> * pro: Storing watermark in table properties make it easier for consumer
> * con: It needs `Table#newTransaction` to commit append and table properties update together. Transaction change might require bigger refactor in the IcebergFilesCommitter.
@stevenzwu ok, I'm agree with it.
what about mark it in both the snapshot of the append operation and the table properties. In append snapshot, we can get the ability of trend analysis. Mark the latest watermark in the table properties, user can use it very conveniently.
`private Long watermark = WATERMARK_VALUE_DEFAULT;`
watermark is a timestamp, so type is long.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] liubo1022126 commented on pull request #3130: Flink: support write watermark in snapshot
Posted by GitBox <gi...@apache.org>.
liubo1022126 commented on pull request #3130:
URL: https://github.com/apache/iceberg/pull/3130#issuecomment-921432832
There is a problem with this pr, When the write job is started from a non-state, It will get the watermark in the current snapshot, if the operation of the current snapshot is not append operation(delete, rewrite), it can't get watermark in the current snapshot, because the advancement information of watermark is only implemented in streaming write.
Now I can think of two solutions:
1. Add watermark transfer support for various operations, like delete, rewrite and so on: But this idea have a problem, In the scenario of streaming write while file rewrite, It is common for new data to be written in the process of rewrite start to end. that is to say, when snapshot s1 has watermark w1 when file rewrite begin, then streaming write commit new snapshot s2 has watermark w2, then file rewrite complete for snapshot s3 with watermark w3, to calculate w3, in addition to passing w1 to w3, we must also perform additional calculations on w2. It looks very complicated and hard to understand.
2. Keep the current implementation in this pr, only record watermark in append operation: When we need to get the watermark of the current table, we will backtrack the table snapshot until we get the watermark value in the most recent append operation.
I prefer solution 2. what about you think? @stevenzwu @rdblue @openinx
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a change in pull request #3130: Flink: support write watermark in snapshot
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #3130:
URL: https://github.com/apache/iceberg/pull/3130#discussion_r716357813
##########
File path: core/src/main/java/org/apache/iceberg/io/WriteResult.java
##########
@@ -28,11 +28,15 @@
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.util.CharSequenceSet;
+import static org.apache.iceberg.TableProperties.WATERMARK_VALUE_DEFAULT;
+
public class WriteResult implements Serializable {
private DataFile[] dataFiles;
private DeleteFile[] deleteFiles;
private CharSequence[] referencedDataFiles;
+ private Long watermark = WATERMARK_VALUE_DEFAULT;
Review comment:
nit: use primitive type `long`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] liubo1022126 commented on a change in pull request #3130: Flink: support write watermark in snapshot
Posted by GitBox <gi...@apache.org>.
liubo1022126 commented on a change in pull request #3130:
URL: https://github.com/apache/iceberg/pull/3130#discussion_r725449357
##########
File path: core/src/main/java/org/apache/iceberg/TableProperties.java
##########
@@ -228,4 +228,10 @@ private TableProperties() {
public static final String MERGE_CARDINALITY_CHECK_ENABLED = "write.merge.cardinality-check.enabled";
public static final boolean MERGE_CARDINALITY_CHECK_ENABLED_DEFAULT = true;
+
+ public static final String WATERMARK_FIELD_NAME = "write.watermark.field";
+ public static final String WATERMARK_FIELD_NAME_DEFAULT = "";
+
+ public static final String WATERMARK_VALUE = "write.watermark";
Review comment:
sorry, I just came back from holiday.
I think it’s a good idea to use different suffixes to distinguish different streaming jobs when writing watermark. Because job processing will not be blocked by other job problems. For example:
it's here, If we treat all physical jobs as a logical job, then each physical job is a parallelism in the logical job. So anyway, if we need to obtain information about different physical jobs, it is necessary to write different watermarks into table properties.
I think what we can do a little more is to automatically aggregate different watermark values during the streaming job commit to get the final result. For example:
That is to say, when using watermark value downstream, only need to use `write.watermark` directly, don't care about upstream logic.
**In code design, we may need to create a new class for the process of watermark, not simply write the watermark processing into IcebergFilesCommitter like this pr. And divided into different implementations: simple watermark implementation and multiple regions watermark implementation.**
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org