rakulang, dartlang, nimlang, golang, rustlang, lang lang no see

Generating Watermarks

焉知非鱼

Generating Watermarks

在本节中,您将了解 Flink 提供的 API,用于处理事件时间时间戳和水印。关于事件时间、处理时间和摄取时间的介绍,请参考事件时间的介绍

水印策略介绍 #

为了使用事件时间,Flink 需要知道事件的时间戳,这意味着流中的每个元素都需要分配其事件时间戳(event timestamp)。这通常是通过使用 TimestampAssigner 从元素中的某个字段访问/提取时间戳(timestamp)来完成的。

时间戳分配与生成水印是同步进行的,水印告诉系统事件时间的进展。你可以通过指定一个 WatermarkGenerator 来配置。

Flink API 期望一个 WatermarkStrategy,其中包含一个 TimestampAssignerWatermarkGenerator。一些常见的策略作为 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 上的静态帮助方法来实现常见的水印策略,或者将自定义的 TimestampAssignerWatermarkGenerator 捆绑在一起。例如,要使用有界无序水印和 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 记录中获取时间戳。

我们将在后面的 Writing WatermarkGenerator中查看 WatermarkGenerator 接口。

注意:时间戳和水印都被指定为自 1970-01-01T00:00:00Z 的 Java 纪元以来的毫秒。

使用水印策略 #

在 Flink 应用中,有两个地方可以使用 WatermarkStrategy。1)直接在源上使用,2)在非源操作后使用。

第一个选项是比较好的,因为它允许源在水印逻辑中利用关于碎片/分区/分割的知识。源通常可以更精细地跟踪水印,源产生的整体水印也会更准确。直接在源上指定 WatermarkStrategy 通常意味着你必须使用源的特定接口/请参阅 Watermark Strategies 和 Kafka Connector,以了解在 Kafka Connector 上如何工作,以及关于每个分区水印如何工作的更多细节。

第二个选项(在任意操作后设置 WatermarkStrategy)只应在不能直接在源上设置策略时使用。

val env = StreamExecutionEnvironment.getExecutionEnvironment
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

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 )
        .timeWindow(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() 时,发射水印。

标点式生成器会观察 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)

img

运算符如何处理水印 #

作为一般规则,运算符(operator)在向下游转发一个给定的水印之前,需要对其进行完全处理。例如,WindowOperator 将首先评估所有应该被发射的窗口,只有在产生所有由水印触发的输出之后,水印本身才会被发送到下游。换句话说,所有因发生水印而产生的元素将在水印之前被发射。

同样的规则也适用于 TwoInputStreamOperator。然而,在这种情况下,运算符的当前水印被定义为其两个输入的最小值。

这种行为的细节由 OneInputStreamOperator#processWatermarkTwoInputStreamOperator#processWatermark1TwoInputStreamOperator#processWatermark2 方法的实现来定义。

废弃的 AssignerWithPeriodicWatermarks 和 AssignerWithPunctuatedWatermarks 方法 #

在引入当前的 WatermarkStrategyTimestampAssignerWatermarkGenerator 抽象之前,Flink 使用了 AssignerWithPeriodicWatermarksAssignerWithPeriodicWatermarks。你仍然会在 API 中看到它们,但建议使用新的接口,因为它们提供了更清晰的分离关注点,也统一了水印生成的周期和标点样式。