生成水印
在本节中，您将了解 Flink 提供的 API，用于处理事件时间时间戳和水印。关于事件时间、处理时间和摄取时间的介绍，请参考事件时间的介绍。
水印策略介绍
为了使用事件时间，Flink需要知道事件的时间戳，这意味着流中的每个元素都需要分配其事件时间戳。这通常是通过使用TimestampAssigner从元素中的某个字段访问/提取时间戳来完成的。
时间戳分配与生成水印是同步进行的，水印告诉系统事件时间的进展。你可以通过指定一个WatermarkGenerator来配置。
Flink API期望一个WatermarkStrategy，其中包含一个TimestampAssigner和WatermarkGenerator。一些常见的策略作为WatermarkStrategy上的静态方法是开箱即用的，但用户也可以在需要时建立自己的策略。
为了完整起见，下面的是接口
public interface WatermarkStrategy<T> extends TimestampAssignerSupplier<T>, WatermarkGeneratorSupplier<T>{
/**
* Instantiates a {@link TimestampAssigner} for assigning timestamps according to this
* strategy.
*/
@Override
TimestampAssigner<T> createTimestampAssigner(TimestampAssignerSupplier.Context context);
/**
* Instantiates a WatermarkGenerator that generates watermarks according to this strategy.
*/
@Override
WatermarkGenerator<T> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context);
}
如前所述，你通常不会自己实现这个接口，而是使用WatermarkStrategy上的静态帮助方法来实现常见的水印策略，或者将自定义的TimestampAssigner与WatermarkGenerator捆绑在一起。例如，要使用有界无序水印和lambda函数作为时间戳分配器，你可以使用这个方法。
WatermarkStrategy
.forBoundedOutOfOrderness[(Long, String)](Duration.ofSeconds(20))
.withTimestampAssigner(new SerializableTimestampAssigner[(Long, String)] {
override def extractTimestamp(element: (Long, String), recordTimestamp: Long): Long = element._1
})
(在这里使用Scala Lambdas目前是行不通的，因为Scala很笨，很难支持这个。#fus)
指定一个TimestampAssigner是可选的，在大多数情况下，你其实并不想指定一个。例如，当使用Kafka或Kinesis时，你会直接从Kafka/Kinesis记录中获取时间戳。
我们将在后面的WatermarkGenerator编写中查看WatermarkGenerator接口。
注意：时间戳和水印都是在Kafka/Kinesis中获取的。时间戳和水印都被指定为自1970-01-01T00:00:00Z的Java纪元以来的毫秒。
使用水印策略
在Flink应用中，有两个地方可以使用WatermarkStrategy。1）直接在源上使用，2）在非源操作后使用。
第一个选项是比较好的，因为它允许源在水印逻辑中利用关于碎片/分区/分割的知识。源通常可以更精细地跟踪水印，源产生的整体水印也会更准确。直接在源上指定WatermarkStrategy通常意味着你必须使用源的特定接口/请参阅Watermark Strategies和Kafka Connector，以了解在Kafka Connector上如何工作，以及关于每个分区水印如何工作的更多细节。
第二个选项（在任意操作后设置WatermarkStrategy）只应在不能直接在源上设置策略时使用。
val env = StreamExecutionEnvironment.getExecutionEnvironment
val stream: DataStream[MyEvent] = env.readFile(
myFormat, myFilePath, FileProcessingMode.PROCESS_CONTINUOUSLY, 100,
FilePathFilter.createDefaultFilter())
val withTimestampsAndWatermarks: DataStream[MyEvent] = stream
.filter( _.severity == WARNING )
.assignTimestampsAndWatermarks(<watermark strategy>)
withTimestampsAndWatermarks
.keyBy( _.getGroup )
.window(TumblingEventTimeWindows.of(Time.seconds(10)))
.reduce( (a, b) => a.add(b) )
.addSink(...)
以这种方式使用WatermarkStrategy，可以获取一个流并生成一个带有时间戳元素和水印的新流。如果原始流已经有时间戳和/或水印，时间戳分配器会覆盖它们。
处理闲置数据源
如果其中一个输入分割/分区/碎片在一段时间内没有携带事件，这意味着WatermarkGenerator也没有得到任何新的信息来作为水印的基础。我们称之为空闲输入或空闲源。这是一个问题，因为有可能发生你的一些分区仍然携带事件。在这种情况下，水印将被保留下来，因为它是作为所有不同的并行水印的最小值计算的。
为了处理这个问题，你可以使用WatermarkStrategy来检测空闲，并将一个输入标记为空闲。WatermarkStrategy为此提供了一个方便的助手。
WatermarkStrategy .forBoundedOutOfOrderness[(Long, String)](Duration.ofSeconds(20)) .withIdleness(Duration.ofMinutes(1))
编写水印生成器
时间戳分配器（TimestampAssigner）是一个从事件中提取字段的简单函数，因此我们不需要详细研究它们。而WatermarkGenerator的编写就比较复杂了，我们将在接下来的两节中看如何做。这就是WatermarkGenerator的界面。
/**
* The {@code WatermarkGenerator} generates watermarks either based on events or
* periodically (in a fixed interval).
*
* <p><b>Note:</b> This WatermarkGenerator subsumes the previous distinction between the
* {@code AssignerWithPunctuatedWatermarks} and the {@code AssignerWithPeriodicWatermarks}.
*/
@Public
public interface WatermarkGenerator<T> {
/**
* 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.
*
* <p>The interval in which this method is called and Watermarks are generated
* depends on {@link ExecutionConfig#getAutoWatermarkInterval()}.
*/
void onPeriodicEmit(WatermarkOutput output);
}
有两种不同风格的水印生成器：周期性和标点式。
周期性生成器通常会通过onEvent()观察传入的事件，然后在框架调用onPeriodicEmit()时发出水印。
puncutated生成器会观察onEvent()中的事件，并等待流中携带水印信息的特殊标记事件或标点。当它看到这些事件之一时，就会立即发出一个水印。通常，标点生成器不会从onPeriodicEmit()发出水印。
接下来我们将看看如何实现每种样式的生成器。
编写一个周期性的水印生成器
周期性生成器观察流事件并周期性地生成水印（可能取决于流元素，或者纯粹基于处理时间）。
生成水印的间隔（每n毫秒）通过ExecutionConfig.setAutoWatermarkInterval(...)来定义。每次都会调用生成器的onPeriodicEmit()方法，如果返回的水印是非空的，并且大于前一个水印，就会发出一个新的水印。
这里我们展示了两个使用周期性水印生成器的简单例子。请注意，Flink提供了BoundedOutfOrdernessWatermarks，这是一个WatermarkGenerator，它的工作原理与下面所示的BoundedOutfOrdernessGenerator类似。你可以在这里阅读关于如何使用它。
/**
* This generator generates watermarks assuming that elements arrive out of order,
* but only to a certain degree. The latest elements for a certain timestamp t will arrive
* at most n milliseconds after the earliest elements for timestamp t.
*/
class BoundedOutOfOrdernessGenerator extends AssignerWithPeriodicWatermarks[MyEvent] {
val maxOutOfOrderness = 3500L // 3.5 seconds
var currentMaxTimestamp: Long = _
override def onEvent(element: MyEvent, eventTimestamp: Long): Unit = {
currentMaxTimestamp = max(eventTimestamp, currentMaxTimestamp)
}
override def onPeriodicEmit(): Unit = {
// emit the watermark as current highest timestamp minus the out-of-orderness bound
output.emitWatermark(new Watermark(currentMaxTimestamp - maxOutOfOrderness - 1));
}
}
/**
* This generator generates watermarks that are lagging behind processing time by a fixed amount.
* It assumes that elements arrive in Flink after a bounded delay.
*/
class TimeLagWatermarkGenerator extends AssignerWithPeriodicWatermarks[MyEvent] {
val maxTimeLag = 5000L // 5 seconds
override def onEvent(element: MyEvent, eventTimestamp: Long): Unit = {
// don't need to do anything because we work on processing time
}
override def onPeriodicEmit(): Unit = {
output.emitWatermark(new Watermark(System.currentTimeMillis() - maxTimeLag));
}
}
编写一个标点水印生成器
标点水印生成器将观察事件流，每当它看到一个携带水印信息的特殊元素时，就会发出一个水印。
这就是如何实现一个标点水印生成器，每当一个事件表明它携带某个标记时，它就会发射一个水印。
class PunctuatedAssigner extends AssignerWithPunctuatedWatermarks[MyEvent] {
override def onEvent(element: MyEvent, eventTimestamp: Long): Unit = {
if (event.hasWatermarkMarker()) {
output.emitWatermark(new Watermark(event.getWatermarkTimestamp()))
}
}
override def onPeriodicEmit(): Unit = {
// don't need to do anything because we emit in reaction to events above
}
}
注：可以对每个事件生成一个水印。但是，由于每个水印都会引起下游的一些计算，因此过多的水印会降低性能。
水印策略与Kafka连接器
当使用Apache Kafka作为数据源时，每个Kafka分区可能有一个简单的事件时间模式（升序时间戳或有界失序）。然而，当消耗来自Kafka的流时，多个分区经常会被并行消耗，交织来自分区的事件，并破坏每个分区的模式（这是Kafka的消费者客户端的固有工作方式）。
在这种情况下，你可以使用Flink的Kafka-partition-aware水印生成功能。使用该功能，在Kafka消费者内部，按Kafka分区生成水印，每个分区水印的合并方式与流洗牌的水印合并方式相同。
例如，如果每个Kafka分区的事件时间戳是严格的升序，那么用升序时间戳水印生成器生成每个分区的水印，会得到完美的整体水印。请注意，我们在示例中并没有提供TimestampAssigner，而是使用Kafka记录本身的时间戳。
下面的插图展示了如何使用per-Kafka-partition水印生成器，以及在这种情况下水印如何通过流式数据流传播。
val kafkaSource = new FlinkKafkaConsumer[MyType]("myTopic", schema, props)
kafkaSource.assignTimestampsAndWatermarks(
WatermarkStrategy
.forBoundedOutOfOrderness(Duration.ofSeconds(20)))
val stream: DataStream[MyType] = env.addSource(kafkaSource)
操作符如何处理水印
作为一般规则，操作者在向下游转发一个给定的水印之前，需要对其进行完全处理。例如，WindowOperator将首先评估所有应该被发射的窗口，只有在产生所有由水印触发的输出后，水印本身才会被发送到下游。换句话说，所有因发生水印而产生的元素将在水印之前被发射。
同样的规则也适用于TwoInputStreamOperator。然而，在这种情况下，运算器的当前水印被定义为其两个输入的最小值。
这种行为的细节由OneInputStreamOperator#processWatermark、TwoInputStreamOperator#processWatermark1和TwoInputStreamOperator#processWatermark2方法的实现来定义。
废弃的AssignerWithPeriodicWatermarks和AssignerWithPunctuatedWatermarks
在引入当前的WatermarkStrategy、TimestampAssigner和WatermarkGenerator抽象之前，Flink使用了AssignerWithPeriodicWatermarks和AssignerWithPunctuatedWatermarks。你仍然会在API中看到它们，但建议使用新的接口，因为它们提供了更清晰的分离关注点，而且还统一了水印生成的周期和标点样式。