0 https://github.com/Wasabi1234/Flink-Tutorial

掌握Flink中三种常用的Time处理方式，掌握Flink中滚动窗口以及滑动窗口的使用，了解Flink中的watermark。

Flink 在流处理工程中支持不同的时间概念。

1 处理时间(Processing time)

执行相应算子操作的机器的系统时间.

当流程序在处理时间运行时，所有基于时间的 算子操作（如时间窗口）将使用运行相应算子的机器的系统时钟。每小时处理时间窗口将包括在系统时钟指示整个小时之间到达特定算子的所有记录。

例如，如果应用程序在上午9:15开始运行，则第一个每小时处理时间窗口将包括在上午9:15到上午10:00之间处理的事件，下一个窗口将包括在上午10:00到11:00之间处理的事件

处理时间是最简单的时间概念，不需要流和机器之间的协调

它提供最佳性能和最低延迟。但是，在分布式和异步环境中，处理时间不提供确定性，因为它容易受到记录到达系统的速度（例如从消息队列）到记录在系统内的算子之间流动的速度的影响。和停电（调度或其他）。

2 事件时间(Event time)

每个单独的事件在其生产设备上发生的时间.

此时间通常在进入Flink之前内置在记录中，并且可以从每个记录中提取该事件时间戳。

在事件时间，时间的进展取决于数据，而不是任何挂钟。

事件时间程序必须指定如何生成事件时间水印，这是表示事件时间进度的机制.

在一个完美的世界中，事件时间处理将产生完全一致和确定的结果，无论事件何时到达，或者顺序.

但是，除非事件已知按顺序到达（按时间戳），否则事件时间处理会在等待无序事件时产生一些延迟。由于只能等待一段有限的时间，因此限制了确定性事件时间应用程序的可能性。

假设所有数据都已到达，算子操作将按预期运行，即使在处理无序或延迟事件或重新处理历史数据时也会产生正确且一致的结果。

例如，每小时事件时间窗口将包含带有落入该小时的事件时间戳的所有记录，无论它们到达的顺序如何，或者何时处理它们。（有关更多信息，请参阅有关迟发事件的部分。）

请注意，有时当事件时间程序实时处理实时数据时，它们将使用一些处理时间 算子操作，以确保它们及时进行。

3 摄取时间(Ingestion time)

事件进入Flink的时间.

在源算子处，每个记录将源的当前时间作为时间戳，并且基于时间的算子操作（如时间窗口）引用该时间戳。

在概念上位于事件时间和处理时间之间。

与处理时间相比 ，它成本稍微高一些，但可以提供更可预测的结果。因为使用稳定的时间戳（在源处分配一次），所以对记录的不同窗口 算子操作将引用相同的时间戳，而在处理时间中，每个窗口算子可以将记录分配给不同的窗口（基于本地系统时钟和任何运输延误）

与事件时间相比，无法处理任何无序事件或后期数据，但程序不必指定如何生成水印。

在内部，摄取时间与事件时间非常相似，但具有自动时间戳分配和自动水印生成函数

4 设置时间特性

Flink DataStream程序的第一部分通常设置基本时间特性

显然,在Flink的流式处理环境中,默认使用处理时间

该设置定义了数据流源的行为方式（例如，它们是否将分配时间戳），以及窗口 算子操作应该使用的时间概念,比如

KeyedStream.timeWindow(Time.seconds(30))。
以下示例显示了一个Flink程序，该程序在每小时时间窗口中聚合事件。窗口的行为适应时间特征。

• Java

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
// 可选的:
// env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);
// env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
DataStream<MyEvent> stream = env.addSource(new FlinkKafkaConsumer09<MyEvent>(topic, schema, props));
stream
    .keyBy( (event) -> event.getUser() )
    .timeWindow(Time.hours(1))
    .reduce( (a, b) -> a.add(b) )
    .addSink(...);

• Scala

val env = StreamExecutionEnvironment.getExecutionEnvironment
env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime)
// alternatively:
// env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime)
// env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
val stream: DataStream[MyEvent] = env.addSource(new FlinkKafkaConsumer09[MyEvent](topic, schema, props))
stream
    .keyBy( _.getUser )
    .timeWindow(Time.hours(1))
    .reduce( (a, b) => a.add(b) )
    .addSink(...)

请注意，为了在事件时间运行此示例，程序需要使用直接为数据定义事件时间的源并自行发出水印，或者程序必须在源之后注入时间戳分配器和水印生成器。这些函数描述了如何访问事件时间戳，以及事件流表现出的无序程度。

5 Windows

5.1 简介

Windows是处理无限流的核心。Windows将流拆分为有限大小的“桶”，我们可以在其上应用计算。我们重点介绍如何在Flink中执行窗口，以及程序员如何从其提供的函数中获益最大化。

窗口Flink程序的一般结构如下所示

第一个片段指的是被Keys化流

而第二个片段指的是非被Keys化流

正如所看到的，唯一的区别是keyBy(…)呼吁Keys流和window(…)成为windowAll(…)非被Key化的数据流。这也将作为页面其余部分的路线图。

Keyed Windows

Non-Keyed Windows

在上面，方括号（[…]）中的命令是可选的。这表明Flink允许您以多种不同方式自定义窗口逻辑，以便最适合您的需求。

5.2 窗口生命周期

简而言之，只要应该属于此窗口的第一个数据元到达，就会创建一个窗口，当时间（事件或处理时间）超过其结束时间戳加上用户指定 时，窗口将被完全删除allowed lateness（请参阅允许的延迟））。Flink保证仅删除基于时间的窗口而不是其他类型，例如全局窗口（请参阅窗口分配器）。例如，使用基于事件时间的窗口策略，每5分钟创建一个非重叠（或翻滚）的窗口，并允许延迟1分钟，Flink将创建一个新窗口，用于间隔12:00和12:05当具有落入此间隔的时间戳的第一个数据元到达时，当水印通过12:06 时间戳时它将删除它。

此外，每个窗口将具有Trigger和一个函数（ProcessWindowFunction，ReduceFunction， AggregateFunction或FoldFunction）连接到它。该函数将包含要应用于窗口内容的计算，而Trigger指定窗口被认为准备好应用该函数的条件。

触发策略可能类似于“当窗口中的数据元数量大于4”时，或“当水印通过窗口结束时”。

触发器还可以决定在创建和删除之间的任何时间清除窗口的内容。在这种情况下，清除仅指窗口中的数据元，而不是窗口元数据。这意味着仍然可以将新数据添加到该窗口。

除了上述内容之外，您还可以指定一个Evictor，它可以在触发器触发后以及应用函数之前和/或之后从窗口中删除数据元。

5.3 被Keys化与非被Keys化Windows

要指定的第一件事是您的流是否应该键入。必须在定义窗口之前完成此 算子操作。使用the keyBy(…)将您的无限流分成逻辑被Key化的数据流。如果keyBy(…)未调用，则表示您的流不是被Keys化的。

对于被Key化的数据流，可以将传入事件的任何属性用作键（此处有更多详细信息）。拥有被Key化的数据流将允许您的窗口计算由多个任务并行执行，因为每个逻辑被Key化的数据流可以独立于其余任务进行处理。引用相同Keys的所有数据元将被发送到同一个并行任务。

在非被Key化的数据流的情况下，您的原始流将不会被拆分为多个逻辑流，并且所有窗口逻辑将由单个任务执行，即并行度为1。

6 窗口分配器

指定流是否已键入后，下一步是定义一个窗口分配器.

窗口分配器定义如何将数据元分配给窗口,这是通过WindowAssigner 在window(…)（对于被Keys化流）或windowAll()（对于非被Keys化流）调用中指定您的选择来完成的

WindowAssigner负责将每个传入数据元分配给一个或多个窗口

Flink带有预定义的窗口分配器，用于最常见的用例，即

滚动窗口

滑动窗口

会话窗口

全局窗口

还可以通过扩展WindowAssigner类来实现自定义窗口分配器。所有内置窗口分配器（全局窗口除外）都根据时间为窗口分配数据元，这可以是处理时间或事件时间。请查看我们关于活动时间的部分，了解处理时间和事件时间之间的差异以及时间戳和水印的生成方式。

基于时间的窗口具有开始时间戳（包括）和结束时间戳（不包括），它们一起描述窗口的大小。

在代码中，Flink在使用TimeWindow基于时间的窗口时使用，该窗口具有查询开始和结束时间戳的方法maxTimestamp()返回给定窗口的最大允许时间戳

下图显示了每个分配者的工作情况。紫色圆圈表示流的数据元，这些数据元由某个键（在这种情况下是用户1，用户2和用户3）划分。x轴显示时间的进度。

6.1 滚动窗口

一个滚动窗口分配器的每个数据元分配给指定的窗口的窗口大小。滚动窗口具有固定的尺寸，不重叠.

例如，如果指定大小为5分钟的翻滚窗口，则将评估当前窗口，并且每五分钟将启动一个新窗口，如下图所示

以下代码段显示了如何使用滚动窗口。

• Java

DataStream<T> input = ...;
// tumbling event-time windows
input
    .keyBy(<key selector>)
    .window(TumblingEventTimeWindows.of(Time.seconds(5)))
    .<windowed transformation>(<window function>);
// tumbling processing-time windows
input
    .keyBy(<key selector>)
    .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
    .<windowed transformation>(<window function>);
// daily tumbling event-time windows offset by -8 hours.
input
    .keyBy(<key selector>)
    .window(TumblingEventTimeWindows.of(Time.days(1), Time.hours(-8)))
    .<windowed transformation>(<window function>);

• Scala

val input: DataStream[T] = ...
// tumbling event-time windows
input
    .keyBy(<key selector>)
    .window(TumblingEventTimeWindows.of(Time.seconds(5)))
    .<windowed transformation>(<window function>)
// tumbling processing-time windows
input
    .keyBy(<key selector>)
    .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
    .<windowed transformation>(<window function>)
// daily tumbling event-time windows offset by -8 hours.
input
    .keyBy(<key selector>)
    .window(TumblingEventTimeWindows.of(Time.days(1), Time.hours(-8)))
    .<windowed transformation>(<window function>)

Scala

Java

6.2 滑动窗口

该滑动窗口分配器分配元件以固定长度的窗口。与滚动窗口分配器类似，窗口大小由窗口大小参数配置

附加的窗口滑动参数控制滑动窗口的启动频率。因此，如果幻灯片小于窗口大小，则滑动窗口可以重叠。在这种情况下，数据元被分配给多个窗口。

例如，您可以将大小为10分钟的窗口滑动5分钟。有了这个，你每隔5分钟就会得到一个窗口，其中包含过去10分钟内到达的事件，如下图所示。

以下代码段显示了如何使用滑动窗口

• Java

DataStream<T> input = ...;
// 滑动 事件时间 窗口
input
    .keyBy(<key selector>)
    .window(TumblingEventTimeWindows.of(Time.seconds(5)))
    .<windowed transformation>(<window function>);
//  滑动 处理时间 窗口
input
    .keyBy(<key selector>)
    .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
    .<windowed transformation>(<window function>);
// daily tumbling event-time windows offset by -8 hours.
input
    .keyBy(<key selector>)
    .window(TumblingEventTimeWindows.of(Time.days(1), Time.hours(-8)))
    .<windowed transformation>(<window function>);

• Scala

val input: DataStream[T] = ...
// tumbling event-time windows
input
    .keyBy(<key selector>)
    .window(TumblingEventTimeWindows.of(Time.seconds(5)))
    .<windowed transformation>(<window function>)
// tumbling processing-time windows
input
    .keyBy(<key selector>)
    .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
    .<windowed transformation>(<window function>)
// daily tumbling event-time windows offset by -8 hours.
input
    .keyBy(<key selector>)
    .window(TumblingEventTimeWindows.of(Time.days(1), Time.hours(-8)))
    .<windowed transformation>(<window function>)