水印的作用请教

hi everyone,

最近在使用Stream API或Table API&SQL编写任务，接Kafka数据流写HDFS的时候，关于水印有几个问题想请教社区帮忙答疑。

在我的理解中，水印本意是容忍事件时间的延迟（乱序程序），在不能容忍的时候触发窗口计算，以达到输出该窗口结果的目的。
那么，
1.在Kafka入HDFS的过程中，水印的作用具体是什么呢？貌似无窗口计算？
2.如果我在SQL或StreamAPI中使用ForBoundedOutOfOrderness中基于事件时间设置允许5分钟延迟的水印，那么Kafka各分区最小时间的水印到达后，比水印晚的数据还会写入HDFS（基于事件时间分区）对应的分区中吗？是丢弃？还是不丢弃它写入对应分区后再重新幂等提交该分区？

best,
amenhub

[tison]

取决于你的计算流图，watermark 通常只在以下情况有实际作用

True
& cond 1. 使用 EventTime
& cond 2. 流图中有 Timer 触发逻辑，例如 Window 算子或自定义 Timer

Best,
tison.


amenhub <[hidden email]> 于2021年2月1日周一 上午10:26写道：

> hi everyone,
>
> 最近在使用Stream API或Table API&SQL编写任务，接Kafka数据流写HDFS的时候，关于水印有几个问题想请教社区帮忙答疑。
>
> 在我的理解中，水印本意是容忍事件时间的延迟（乱序程序），在不能容忍的时候触发窗口计算，以达到输出该窗口结果的目的。
> 那么，
> 1.在Kafka入HDFS的过程中，水印的作用具体是什么呢？貌似无窗口计算？
>
> 2.如果我在SQL或StreamAPI中使用ForBoundedOutOfOrderness中基于事件时间设置允许5分钟延迟的水印，那么Kafka各分区最小时间的水印到达后，比水印晚的数据还会写入HDFS（基于事件时间分区）对应的分区中吗？是丢弃？还是不丢弃它写入对应分区后再重新幂等提交该分区？
>
> best,
> amenhub
>
>
>
>

谢谢回复！

也就是说如果我利用Flink从Kafka （Select *）采集数据到HDFS，不涉及Timer触发逻辑，使用水印的目的就只是为了使用事件时间以及HDFS中基于事件时间进行分区目录创建，对吗？

best,
amenhub



 
发件人： tison
发送时间： 2021-02-01 10:36
收件人： user-zh
主题： Re: 水印的作用请教
取决于你的计算流图，watermark 通常只在以下情况有实际作用
 
True
& cond 1. 使用 EventTime
& cond 2. 流图中有 Timer 触发逻辑，例如 Window 算子或自定义 Timer
 
Best,
tison.
 
 
amenhub <[hidden email]> 于2021年2月1日周一 上午10:26写道：
 

> hi everyone,
>
> 最近在使用Stream API或Table API&SQL编写任务，接Kafka数据流写HDFS的时候，关于水印有几个问题想请教社区帮忙答疑。
>
> 在我的理解中，水印本意是容忍事件时间的延迟（乱序程序），在不能容忍的时候触发窗口计算，以达到输出该窗口结果的目的。
> 那么，
> 1.在Kafka入HDFS的过程中，水印的作用具体是什么呢？貌似无窗口计算？
>
> 2.如果我在SQL或StreamAPI中使用ForBoundedOutOfOrderness中基于事件时间设置允许5分钟延迟的水印，那么Kafka各分区最小时间的水印到达后，比水印晚的数据还会写入HDFS（基于事件时间分区）对应的分区中吗？是丢弃？还是不丢弃它写入对应分区后再重新幂等提交该分区？
>
> best,
> amenhub
>
>
>
>

>>> 2.如果我在SQL或StreamAPI中使用ForBoundedOutOfOrderness中基于事件时间设置允许5分钟延迟的水印，那么Kafka各分区最小时间的水印到达后，比水印晚的数据还会写入HDFS（基于事件时间分区）对应的分区中吗？是丢弃？还是不丢弃它写入对应分区后再重新幂等提交该分区？
另外还是比较想了解，当指定水印及延迟的时候，比水印晚的数据，在流写HDFS的时候会怎么处理？



 
发件人： amenhub
发送时间： 2021-02-01 10:44
收件人： user-zh
主题： Re: Re: 水印的作用请教
谢谢回复！
 
也就是说如果我利用Flink从Kafka （Select *）采集数据到HDFS，不涉及Timer触发逻辑，使用水印的目的就只是为了使用事件时间以及HDFS中基于事件时间进行分区目录创建，对吗？
 
best,
amenhub
 
 
 
发件人： tison
发送时间： 2021-02-01 10:36
收件人： user-zh
主题： Re: 水印的作用请教
取决于你的计算流图，watermark 通常只在以下情况有实际作用
True
& cond 1. 使用 EventTime
& cond 2. 流图中有 Timer 触发逻辑，例如 Window 算子或自定义 Timer
Best,
tison.
amenhub <[hidden email]> 于2021年2月1日周一 上午10:26写道：

> hi everyone,
>
> 最近在使用Stream API或Table API&SQL编写任务，接Kafka数据流写HDFS的时候，关于水印有几个问题想请教社区帮忙答疑。
>
> 在我的理解中，水印本意是容忍事件时间的延迟（乱序程序），在不能容忍的时候触发窗口计算，以达到输出该窗口结果的目的。
> 那么，
> 1.在Kafka入HDFS的过程中，水印的作用具体是什么呢？貌似无窗口计算？
>
> 2.如果我在SQL或StreamAPI中使用ForBoundedOutOfOrderness中基于事件时间设置允许5分钟延迟的水印，那么Kafka各分区最小时间的水印到达后，比水印晚的数据还会写入HDFS（基于事件时间分区）对应的分区中吗？是丢弃？还是不丢弃它写入对应分区后再重新幂等提交该分区？
>
> best,
> amenhub
>
>
>
>

[tison]

请问你使用哪种 SinkConnector 写入 HDFS 呢？

Best,
tison.


amenhub <[hidden email]> 于2021年2月1日周一 上午10:58写道：

> >>>
> 2.如果我在SQL或StreamAPI中使用ForBoundedOutOfOrderness中基于事件时间设置允许5分钟延迟的水印，那么Kafka各分区最小时间的水印到达后，比水印晚的数据还会写入HDFS（基于事件时间分区）对应的分区中吗？是丢弃？还是不丢弃它写入对应分区后再重新幂等提交该分区？
> 另外还是比较想了解，当指定水印及延迟的时候，比水印晚的数据，在流写HDFS的时候会怎么处理？
>
>
>
>
> 发件人： amenhub
> 发送时间： 2021-02-01 10:44
> 收件人： user-zh
> 主题： Re: Re: 水印的作用请教
> 谢谢回复！
>
> 也就是说如果我利用Flink从Kafka （Select
> *）采集数据到HDFS，不涉及Timer触发逻辑，使用水印的目的就只是为了使用事件时间以及HDFS中基于事件时间进行分区目录创建，对吗？
>
> best,
> amenhub
>
>
>
> 发件人： tison
> 发送时间： 2021-02-01 10:36
> 收件人： user-zh
> 主题： Re: 水印的作用请教
> 取决于你的计算流图，watermark 通常只在以下情况有实际作用
> True
> & cond 1. 使用 EventTime
> & cond 2. 流图中有 Timer 触发逻辑，例如 Window 算子或自定义 Timer
> Best,
> tison.
> amenhub <[hidden email]> 于2021年2月1日周一 上午10:26写道：
> > hi everyone,
> >
> > 最近在使用Stream API或Table API&SQL编写任务，接Kafka数据流写HDFS的时候，关于水印有几个问题想请教社区帮忙答疑。
> >
> > 在我的理解中，水印本意是容忍事件时间的延迟（乱序程序），在不能容忍的时候触发窗口计算，以达到输出该窗口结果的目的。
> > 那么，
> > 1.在Kafka入HDFS的过程中，水印的作用具体是什么呢？貌似无窗口计算？
> >
> >
> 2.如果我在SQL或StreamAPI中使用ForBoundedOutOfOrderness中基于事件时间设置允许5分钟延迟的水印，那么Kafka各分区最小时间的水印到达后，比水印晚的数据还会写入HDFS（基于事件时间分区）对应的分区中吗？是丢弃？还是不丢弃它写入对应分区后再重新幂等提交该分区？
> >
> > best,
> > amenhub
> >
> >
> >
> >
>

StreamAPI使用的是StreamingFileSink，SQL就是FileSystem了



 
发件人： tison
发送时间： 2021-02-01 11:01
收件人： user-zh
主题： Re: Re: 水印的作用请教
请问你使用哪种 SinkConnector 写入 HDFS 呢？
 
Best,
tison.
 
 
amenhub <[hidden email]> 于2021年2月1日周一 上午10:58写道：
 

> >>>
> 2.如果我在SQL或StreamAPI中使用ForBoundedOutOfOrderness中基于事件时间设置允许5分钟延迟的水印，那么Kafka各分区最小时间的水印到达后，比水印晚的数据还会写入HDFS（基于事件时间分区）对应的分区中吗？是丢弃？还是不丢弃它写入对应分区后再重新幂等提交该分区？
> 另外还是比较想了解，当指定水印及延迟的时候，比水印晚的数据，在流写HDFS的时候会怎么处理？
>
>
>
>
> 发件人： amenhub
> 发送时间： 2021-02-01 10:44
> 收件人： user-zh
> 主题： Re: Re: 水印的作用请教
> 谢谢回复！
>
> 也就是说如果我利用Flink从Kafka （Select
> *）采集数据到HDFS，不涉及Timer触发逻辑，使用水印的目的就只是为了使用事件时间以及HDFS中基于事件时间进行分区目录创建，对吗？
>
> best,
> amenhub
>
>
>
> 发件人： tison
> 发送时间： 2021-02-01 10:36
> 收件人： user-zh
> 主题： Re: 水印的作用请教
> 取决于你的计算流图，watermark 通常只在以下情况有实际作用
> True
> & cond 1. 使用 EventTime
> & cond 2. 流图中有 Timer 触发逻辑，例如 Window 算子或自定义 Timer
> Best,
> tison.
> amenhub <[hidden email]> 于2021年2月1日周一 上午10:26写道：
> > hi everyone,
> >
> > 最近在使用Stream API或Table API&SQL编写任务，接Kafka数据流写HDFS的时候，关于水印有几个问题想请教社区帮忙答疑。
> >
> > 在我的理解中，水印本意是容忍事件时间的延迟（乱序程序），在不能容忍的时候触发窗口计算，以达到输出该窗口结果的目的。
> > 那么，
> > 1.在Kafka入HDFS的过程中，水印的作用具体是什么呢？貌似无窗口计算？
> >
> >
> 2.如果我在SQL或StreamAPI中使用ForBoundedOutOfOrderness中基于事件时间设置允许5分钟延迟的水印，那么Kafka各分区最小时间的水印到达后，比水印晚的数据还会写入HDFS（基于事件时间分区）对应的分区中吗？是丢弃？还是不丢弃它写入对应分区后再重新幂等提交该分区？
> >
> > best,
> > amenhub
> >
> >
> >
> >
>

[tison]

对于 StreamingFileSink 可以查看这两份资料

https://ci.apache.org/projects/flink/flink-docs-stable/dev/connectors/streamfile_sink.html
https://stackoverflow.com/questions/54763160/is-it-possible-that-bucketing-sink-create-bucket-on-event-time

默认的，watermark 和 (EventTime) Timestamp 信息会带到 BucketAssigner 实现所需的 Context
里，Flink 没有开箱即用的基于 EventTime 的分桶策略，你需要自己尝试实现。比水印晚的数据，可以自行实现为丢弃或追加到原有分区文件上。

对于 SQL 可以查看这份资料

https://ci.apache.org/projects/flink/flink-docs-stable/dev/table/connectors/filesystem.html

目前看起来开箱的逻辑迟到数据会追加而不是丢弃。有一些相关的配置可以调整 commit 也就是落盘的时机，但不影响落盘的数据。

Best,
tison.


amenhub <[hidden email]> 于2021年2月1日周一 上午11:07写道：

> StreamAPI使用的是StreamingFileSink，SQL就是FileSystem了
>
>
>
>
> 发件人： tison
> 发送时间： 2021-02-01 11:01
> 收件人： user-zh
> 主题： Re: Re: 水印的作用请教
> 请问你使用哪种 SinkConnector 写入 HDFS 呢？
>
> Best,
> tison.
>
>
> amenhub <[hidden email]> 于2021年2月1日周一 上午10:58写道：
>
> > >>>
> >
> 2.如果我在SQL或StreamAPI中使用ForBoundedOutOfOrderness中基于事件时间设置允许5分钟延迟的水印，那么Kafka各分区最小时间的水印到达后，比水印晚的数据还会写入HDFS（基于事件时间分区）对应的分区中吗？是丢弃？还是不丢弃它写入对应分区后再重新幂等提交该分区？
> > 另外还是比较想了解，当指定水印及延迟的时候，比水印晚的数据，在流写HDFS的时候会怎么处理？
> >
> >
> >
> >
> > 发件人： amenhub
> > 发送时间： 2021-02-01 10:44
> > 收件人： user-zh
> > 主题： Re: Re: 水印的作用请教
> > 谢谢回复！
> >
> > 也就是说如果我利用Flink从Kafka （Select
> > *）采集数据到HDFS，不涉及Timer触发逻辑，使用水印的目的就只是为了使用事件时间以及HDFS中基于事件时间进行分区目录创建，对吗？
> >
> > best,
> > amenhub
> >
> >
> >
> > 发件人： tison
> > 发送时间： 2021-02-01 10:36
> > 收件人： user-zh
> > 主题： Re: 水印的作用请教
> > 取决于你的计算流图，watermark 通常只在以下情况有实际作用
> > True
> > & cond 1. 使用 EventTime
> > & cond 2. 流图中有 Timer 触发逻辑，例如 Window 算子或自定义 Timer
> > Best,
> > tison.
> > amenhub <[hidden email]> 于2021年2月1日周一 上午10:26写道：
> > > hi everyone,
> > >
> > > 最近在使用Stream API或Table API&SQL编写任务，接Kafka数据流写HDFS的时候，关于水印有几个问题想请教社区帮忙答疑。
> > >
> > > 在我的理解中，水印本意是容忍事件时间的延迟（乱序程序），在不能容忍的时候触发窗口计算，以达到输出该窗口结果的目的。
> > > 那么，
> > > 1.在Kafka入HDFS的过程中，水印的作用具体是什么呢？貌似无窗口计算？
> > >
> > >
> >
> 2.如果我在SQL或StreamAPI中使用ForBoundedOutOfOrderness中基于事件时间设置允许5分钟延迟的水印，那么Kafka各分区最小时间的水印到达后，比水印晚的数据还会写入HDFS（基于事件时间分区）对应的分区中吗？是丢弃？还是不丢弃它写入对应分区后再重新幂等提交该分区？
> > >
> > > best,
> > > amenhub
> > >
> > >
> > >
> > >
> >
>

很详尽了，非常感谢 @tison !



 
发件人： tison
发送时间： 2021-02-01 11:43
收件人： user-zh
主题： Re: Re: 水印的作用请教
对于 StreamingFileSink 可以查看这两份资料
 
https://ci.apache.org/projects/flink/flink-docs-stable/dev/connectors/streamfile_sink.html
https://stackoverflow.com/questions/54763160/is-it-possible-that-bucketing-sink-create-bucket-on-event-time
 
默认的，watermark 和 (EventTime) Timestamp 信息会带到 BucketAssigner 实现所需的 Context
里，Flink 没有开箱即用的基于 EventTime 的分桶策略，你需要自己尝试实现。比水印晚的数据，可以自行实现为丢弃或追加到原有分区文件上。
 
对于 SQL 可以查看这份资料
 
https://ci.apache.org/projects/flink/flink-docs-stable/dev/table/connectors/filesystem.html
 
目前看起来开箱的逻辑迟到数据会追加而不是丢弃。有一些相关的配置可以调整 commit 也就是落盘的时机，但不影响落盘的数据。
 
Best,
tison.
 
 
amenhub <[hidden email]> 于2021年2月1日周一 上午11:07写道：
 

> StreamAPI使用的是StreamingFileSink，SQL就是FileSystem了
>
>
>
>
> 发件人： tison
> 发送时间： 2021-02-01 11:01
> 收件人： user-zh
> 主题： Re: Re: 水印的作用请教
> 请问你使用哪种 SinkConnector 写入 HDFS 呢？
>
> Best,
> tison.
>
>
> amenhub <[hidden email]> 于2021年2月1日周一 上午10:58写道：
>
> > >>>
> >
> 2.如果我在SQL或StreamAPI中使用ForBoundedOutOfOrderness中基于事件时间设置允许5分钟延迟的水印，那么Kafka各分区最小时间的水印到达后，比水印晚的数据还会写入HDFS（基于事件时间分区）对应的分区中吗？是丢弃？还是不丢弃它写入对应分区后再重新幂等提交该分区？
> > 另外还是比较想了解，当指定水印及延迟的时候，比水印晚的数据，在流写HDFS的时候会怎么处理？
> >
> >
> >
> >
> > 发件人： amenhub
> > 发送时间： 2021-02-01 10:44
> > 收件人： user-zh
> > 主题： Re: Re: 水印的作用请教
> > 谢谢回复！
> >
> > 也就是说如果我利用Flink从Kafka （Select
> > *）采集数据到HDFS，不涉及Timer触发逻辑，使用水印的目的就只是为了使用事件时间以及HDFS中基于事件时间进行分区目录创建，对吗？
> >
> > best,
> > amenhub
> >
> >
> >
> > 发件人： tison
> > 发送时间： 2021-02-01 10:36
> > 收件人： user-zh
> > 主题： Re: 水印的作用请教
> > 取决于你的计算流图，watermark 通常只在以下情况有实际作用
> > True
> > & cond 1. 使用 EventTime
> > & cond 2. 流图中有 Timer 触发逻辑，例如 Window 算子或自定义 Timer
> > Best,
> > tison.
> > amenhub <[hidden email]> 于2021年2月1日周一 上午10:26写道：
> > > hi everyone,
> > >
> > > 最近在使用Stream API或Table API&SQL编写任务，接Kafka数据流写HDFS的时候，关于水印有几个问题想请教社区帮忙答疑。
> > >
> > > 在我的理解中，水印本意是容忍事件时间的延迟（乱序程序），在不能容忍的时候触发窗口计算，以达到输出该窗口结果的目的。
> > > 那么，
> > > 1.在Kafka入HDFS的过程中，水印的作用具体是什么呢？貌似无窗口计算？
> > >
> > >
> >
> 2.如果我在SQL或StreamAPI中使用ForBoundedOutOfOrderness中基于事件时间设置允许5分钟延迟的水印，那么Kafka各分区最小时间的水印到达后，比水印晚的数据还会写入HDFS（基于事件时间分区）对应的分区中吗？是丢弃？还是不丢弃它写入对应分区后再重新幂等提交该分区？
> > >
> > > best,
> > > amenhub
> > >
> > >
> > >
> > >
> >
>