这篇文章主要介绍“flink中新的水印策略是什么”，在日常操作中，相信很多人在flink中新的水印策略是什么问题上存在疑惑，小编查阅了各式资料，整理出简单好用的操作方法，希望对大家解答”flink中新的水印策略是什么”的疑惑有所帮助！接下来，请跟着小编一起来学习吧！
为上高等地区用户提供了全套网页设计制作服务，及上高网站建设行业解决方案。主营业务为成都做网站、网站制作、上高网站设计，以传统方式定制建设网站，并提供域名空间备案等一条龙服务，秉承以专业、用心的态度为用户提供真诚的服务。我们深信只要达到每一位用户的要求，就会得到认可，从而选择与我们长期合作。这样，我们也可以走得更远！
在flink 1.11之前的版本中，提供了两种生成水印（Watermark）的策略，分别是AssignerWithPunctuatedWatermarks和AssignerWithPeriodicWatermarks，这两个接口都继承自TimestampAssigner接口。
用户想使用不同的水印生成方式，则需要实现不同的接口，但是这样引发了一个问题，对于想给水印添加一些通用的、公共的功能则变得复杂，因为我们需要给这两个接口都同时添加新的功能，这样还造成了代码的重复。
所以为了避免代码的重复，在flink 1.11 中对flink的水印生成接口进行了重构，
当我们构建了一个DataStream之后，使用assignTimestampsAndWatermarks方法来构造水印，新的接口需要传入一个WatermarkStrategy对象。
重庆分公司
重庆分公司
DataStream#assignTimestampsAndWatermarks(WatermarkStrategy)
WatermarkStrategy 这个接口是做什么的呢？这里面提供了很多静态的方法和带有缺省实现的方法，只有一个方法是非default和没有缺省实现的，就是下面的这个方法。
/**
* Instantiates a WatermarkGenerator that generates watermarks according to this strategy.
*/
@Override
WatermarkGenerator createWatermarkGenerator(WatermarkGeneratorSupplier.Context context);
所以默认情况下，我们只需要实现这个方法就行了，这个方法主要是返回一个 WatermarkGenerator，我们在进入这里边看看。
@Public
public interface WatermarkGenerator {
/**
* Called for every event, allows the watermark generator to examine and remember the
* event timestamps, or to emit a watermark based on the event itself.
*/
void onEvent(T event, long eventTimestamp, WatermarkOutput output);
/**
* Called periodically, and might emit a new watermark, or not.
*
* The interval in which this method is called and Watermarks are generated
* depends on {@link ExecutionConfig#getAutoWatermarkInterval()}.
*/
void onPeriodicEmit(WatermarkOutput output);
}
这个方法简单明了，主要是有两个方法：
我们自己实现一个简单的周期性的发射水印的例子：
在这个onEvent方法里，我们从每个元素里抽取了一个时间字段，但是我们并没有生成水印发射给下游，而是自己保存了在一个变量里，在onPeriodicEmit方法里，使用最大的日志时间减去我们想要的延迟时间作为水印发射给下游。
DataStream> withTimestampsAndWatermarks = dataStream.assignTimestampsAndWatermarks(
new WatermarkStrategy>(){
@Override
public WatermarkGenerator> createWatermarkGenerator(
WatermarkGeneratorSupplier.Context context){
return new WatermarkGenerator>(){
private long maxTimestamp;
private long delay = 3000;
@Override
public void onEvent(
Tuple2 event,
long eventTimestamp,
WatermarkOutput output){
maxTimestamp = Math.max(maxTimestamp, event.f1);
}
@Override
public void onPeriodicEmit(WatermarkOutput output){
output.emitWatermark(new Watermark(maxTimestamp - delay));
}
};
}
}); 为了方便开发，flink提供了一些内置的水印生成方法供我们使用。
通过静态方法forBoundedOutOfOrderness提供,入参接收一个Duration类型的时间间隔，也就是我们可以接受的最大的延迟时间.使用这种延迟策略的时候需要我们对数据的延迟时间有一个大概的预估判断。
WatermarkStrategy#forBoundedOutOfOrderness(Duration maxOutOfOrderness)
我们实现一个延迟3秒的固定延迟水印，可以这样做：
DataStream dataStream = ...... ;
dataStream.assignTimestampsAndWatermarks(WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(3)));
他的底层使用的WatermarkGenerator接口的一个实现类BoundedOutOfOrdernessWatermarks。我们看下源码中的这两个方法，是不是和我们上面自己写的很像.
@Override
public void onEvent(T event, long eventTimestamp, WatermarkOutput output) {
maxTimestamp = Math.max(maxTimestamp, eventTimestamp);
}
@Override
public void onPeriodicEmit(WatermarkOutput output) {
output.emitWatermark(new Watermark(maxTimestamp - outOfOrdernessMillis - 1));
}
通过静态方法forMonotonousTimestamps来提供.
WatermarkStrategy.forMonotonousTimestamps()
这个也就是相当于上述的延迟策略去掉了延迟时间，以event中的时间戳充当了水印。
在程序中可以这样使用：
DataStream dataStream = ...... ;
dataStream.assignTimestampsAndWatermarks(WatermarkStrategy.forMonotonousTimestamps());
它的底层实现是AscendingTimestampsWatermarks，其实它就是BoundedOutOfOrdernessWatermarks类的一个子类，没有了延迟时间，我们来看看具体源码的实现.
@Public
public class AscendingTimestampsWatermarks extends BoundedOutOfOrdernessWatermarks {
/**
* Creates a new watermark generator with for ascending timestamps.
*/
public AscendingTimestampsWatermarks() {
super(Duration.ofMillis(0));
}
}
上述我们讲了flink自带的两种水印生成策略，但是对于我们使用eventtime语义的时候，我们想从我们的自己的数据中抽取eventtime，这个就需要TimestampAssigner了.
@Public
@FunctionalInterface
public interface TimestampAssigner {
............
long extractTimestamp(T element, long recordTimestamp);
}
使用的时候我们主要就是从我们自己的元素element中提取我们想要的eventtime。
使用flink自带的水印策略和eventtime抽取类，可以这样用：
DataStream dataStream = ...... ;
dataStream.assignTimestampsAndWatermarks(
WatermarkStrategy
.>forBoundedOutOfOrderness(Duration.ofSeconds(5))
.withTimestampAssigner((event, timestamp)->event.f1));
在某些情况下，由于数据产生的比较少，导致一段时间内没有数据产生，进而就没有水印的生成，导致下游依赖水印的一些操作就会出现问题，比如某一个算子的上游有多个算子，这种情况下，水印是取其上游两个算子的较小值，如果上游某一个算子因为缺少数据迟迟没有生成水印，就会出现eventtime倾斜问题，导致下游没法触发计算。
所以filnk通过WatermarkStrategy.withIdleness()方法允许用户在配置的时间内（即超时时间内）没有记录到达时将一个流标记为空闲。这样就意味着下游的数据不需要等待水印的到来。
当下次有水印生成并发射到下游的时候，这个数据流重新变成活跃状态。
通过下面的代码来实现对于空闲数据流的处理
WatermarkStrategy
.>forBoundedOutOfOrderness(Duration.ofSeconds(20))
.withIdleness(Duration.ofMinutes(1)); 到此，关于“flink中新的水印策略是什么”的学习就结束了，希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习，快去试试吧！若想继续学习更多相关知识，请继续关注创新互联网站，小编会继续努力为大家带来更多实用的文章！
