Flink实战(七) - Time & Windows编程_flink的tumblingeventtimewindows

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

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

1 处理时间（Processing time）

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

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

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

处理时间是最简单的时间概念，不需要流和机器之间的协调。它提供最佳性能和最低延迟。但分布式和异步环境中，处理时间不提供确定性，因为易受记录到达系统的速度（如从MQ）到记录在系统内的算子之间流动的速度的影响。和停电（调度或其他）。

2 事件时间（Event time）

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

此时间通常在进入Flink前内置在记录中，且可以从每个记录中提取该事件时间戳。在事件时间，时间进展取决于数据，而不是任何挂钟。

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

完美世界中，事件时间处理将产生完全一致和确定的结果，无论事件何时到达或顺序。但除非事件已知按顺序到达（按时间戳），否则事件时间处理会在等待无序事件时产生一些延迟。由于只能等待一段有限时间，因此限制了确定性事件时间应用程序的可能性。

假设所有数据都已到达，算子操作将按预期运行，即使在处理无序或延迟事件或重新处理历史数据时也会产生正确且一致的结果。
如每小时事件时间窗口将包含带有落入该小时的事件时间戳的所有记录，无论它们到达的顺序如何或何时处理它们。有时当事件时间程序实时处理实时数据时，它们将使用一些处理时间 算子操作，以确保它们及时进行。

3 摄取时间(Ingestion time)

事件进入Flink的时间。

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

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

• 与处理时间相比 ，成本稍高，但可提供更可预测的结果。因为使用稳定的时间戳（在源处分配一次），所以对记录的不同窗口算子操作将引用相同的时间戳，而在处理时间中，每个窗口算子可以将记录分配给不同的窗口（基于本地系统时钟和任何运输延误）
• 与事件时间相比，无法处理任何无序事件或后期数据，但程序不必指定如何生成水印

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

4 设置时间特性

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

package org.apache.flink.streaming.api;

import org.apache.flink.annotation.PublicEvolving;

/**
 * The time characteristic defines how the system determines time for time-dependent
 * order and operations that depend on time (such as time windows).
 */
@PublicEvolving
public enum TimeCharacteristic {

	ProcessingTime,

	IngestionTime,
  
	EventTime
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

在Flink的流式处理环境中，默认使用处理时间：

@Public
public class StreamExecutionEnvironment {

	/** The default name to use for a streaming job if no other name has been specified. */
	public static final String DEFAULT_JOB_NAME = "Flink Streaming Job";

	/** The time characteristic that is used if none other is set. */
	private static final TimeCharacteristic DEFAULT_TIME_CHARACTERISTIC = TimeCharacteristic.ProcessingTime;
1
2
3
4
5
6
7
8

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

KeyedStream.timeWindow(Time.seconds(30))。
1

以下示例显示Flink程序，该程序在每h时间窗口中聚合事件。窗口的行为适应时间特征：

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(...);
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

为了在事件时间运行此示例，程序要:

• 使用直接为数据定义事件时间的源并自行发出水印
• 或程序必须在源之后注入时间戳分配器和水印生成器

这些函数描述了如何访问事件时间戳，以及事件流表现出的无序程度。

5 Windows

5.1 简介

Windows是处理无限流的核心。Windows将流拆分为有限大小的“桶”，可在其上应用计算。

package org.apache.flink.streaming.api.functions.windowing;

import org.apache.flink.annotation.Public;
import org.apache.flink.api.common.functions.Function;
import org.apache.flink.streaming.api.windowing.windows.Window;
import org.apache.flink.util.Collector;

import java.io.Serializable;

/**
 * Base interface for functions that are evaluated over keyed (grouped) windows.
 *
 * @param <IN> The type of the input value.
 * @param <OUT> The type of the output value.
 * @param <KEY> The type of the key.
 * @param <W> The type of {@code Window} that this window function can be applied on.
 */
@Public
public interface WindowFunction<IN, OUT, KEY, W extends Window> extends Function, Serializable {

    void apply(KEY key, W window, Iterable<IN> input, Collector<OUT> out) throws Exception;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

重点看：

• 如何在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()返回给定窗口的最大允许时间戳：

@PublicEvolving
public class TimeWindow extends Window {

    private final long start;
    private final long end;

    public TimeWindow(long start, long end) {
        this.start = start;
        this.end = end;
    }

    /**
     * Gets the starting timestamp of the window. This is the first timestamp that belongs to this
     * window.
     *
     * @return The starting timestamp of this window.
     */
    public long getStart() {
        return start;
    }

    /**
     * Gets the end timestamp of this window. The end timestamp is exclusive, meaning it is the
     * first timestamp that does not belong to this window any more.
     *
     * @return The exclusive end timestamp of this window.
     */
    public long getEnd() {
        return end;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

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

6.1 滚动窗口

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

如指定大小为5min的翻滚窗口，则将评估当前窗口，并且每5min将启动一个新窗口：

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

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>);
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

package com.javaedge.java.chapter6;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

public class JavaWindowsApp {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> text = env.socketTextStream("localhost", 9999);
        text.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
                    @Override
                    public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
                        String[] tokens = value.toLowerCase().split(",");
                        for (String token : tokens) {
                            if (token.length() > 0) {
                                out.collect(new Tuple2<>(token, 1));
                            }
                        }
                    }
                }).keyBy(0)
                .timeWindow(Time.seconds(5))
                .sum(1)
                .print()
                .setParallelism(1);
        env.execute("JavaWindowsApp");
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

6.2 滑动窗口

该滑动窗口分配器分配元件以固定长度的窗口。与滚动窗口分配器类似，窗口大小由窗口大小参数配置
附加的窗口滑动参数控制滑动窗口的启动频率。因此，如果幻灯片小于窗口大小，则滑动窗口可以重叠。在这种情况下，数据元被分配给多个窗口。

如将10min的窗口滑动5min。有这玩意，你每隔5min就会得到一个窗口，其中包含过去10min内到达的事件，如下：

使用滑动窗口：

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>);
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

7 窗口函数

定义窗口分配器后，我们需要指定要在每个窗口上执行的计算。这是窗口函数的职责，窗口函数用于在系统确定窗口准备好进行处理后处理每个（可能是被Keys化的）窗口的数据元

的窗函数可以是一个ReduceFunction，AggregateFunction，FoldFunction或ProcessWindowFunction。前两个可以更有效地执行，因为Flink可以在每个窗口到达时递增地聚合它们的数据元.
ProcessWindowFunction获取Iterable窗口中包含的所有数据元以及有关数据元所属窗口的其他元信息。

具有ProcessWindowFunction的窗口转换不能像其他情况一样有效地执行，因为Flink必须在调用函数之前在内部缓冲窗口的所有数据元。这可以通过组合来减轻ProcessWindowFunction与ReduceFunction，AggregateFunction或FoldFunction以获得两个窗口元件的增量聚合并且该附加元数据窗口 ProcessWindowFunction接收。我们将查看每个变体的示例。

7.1 ReduceFunction

指定如何组合输入中的两个数据元以生成相同类型的输出数据元.
Flink使用ReduceFunction来递增地聚合窗口的数据元.

定义和使用

DataStream<Tuple2<String, Long>> input = ...;

input
    .keyBy(<key selector>)
    .window(<window assigner>)
    .reduce(new ReduceFunction<Tuple2<String, Long>> {
      public Tuple2<String, Long> reduce(Tuple2<String, Long> v1, Tuple2<String, Long> v2) {
        return new Tuple2<>(v1.f0, v1.f1 + v2.f1);
      }
    });
1
2
3
4
5
6
7
8
9
10

原来传递进来的数据是字符串，此处我们就使用数值类型，通过数值类型来演示增量的效果。

这里不是等待窗口所有的数据进行一次性处理，而是数据两两处理

package com.javaedge.java.chapter6;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

public class JavaWindowsReduceApp {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> text = env.socketTextStream("localhost", 9999);

        text.flatMap(new FlatMapFunction<String, Tuple2<Integer, Integer>>() {
                    @Override
                    public void flatMap(String value, Collector<Tuple2<Integer, Integer>> out) throws Exception {
                        String[] tokens = value.toLowerCase().split(",");
                        for (String token : tokens) {
                            if (token.length() > 0) {
                                out.collect(new Tuple2<>(1, Integer.parseInt(token)));
                            }
                        }
                    }
                }).keyBy(0)
                .timeWindow(Time.seconds(5))
                .reduce(new ReduceFunction<Tuple2<Integer, Integer>>() {
                    @Override
                    public Tuple2<Integer, Integer> reduce(Tuple2<Integer, Integer> value1, Tuple2<Integer, Integer> value2) throws Exception {
                        System.out.println("value1 = [" + value1 + "], value2 = [" + value2 + "]");
                        return new Tuple2<>(value1.f0, value1.f1 + value2.f1);
                    }
                })
                .print()
                .setParallelism(1);
        env.execute("JavaWindowsReduceApp");
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

输入：

javaedge@JavaEdgedeMac-mini ~ % nc -lk 9999
a,a,a,b,b,c
1,2,3,4,5
1
2
3

增量输出：

7.2 聚合函数

AggregateFunction是一个通用版本，ReduceFunction它有三种类型：

• 输入类型（IN）
• 累加器类型（ACC）
• 输出类型（OUT）

输入类型是输入流中数据元的类型，且AggregateFunction具有将一个输入数据元添加到累加器的方法。该接口还具有用于创建初始累加器的方法，用于将两个累加器合并到一个累加器中以及用于OUT从累加器提取输出（类型）。我们将在下面的示例中看到它的工作原理。

与之相同ReduceFunction，Flink将在窗口到达时递增地聚合窗口的输入数据元。

一个AggregateFunction可以被定义并这样使用：

/**
 * The accumulator is used to keep a running sum and a count. The {@code getResult} method
 * computes the average.
 */
private static class AverageAggregate
    implements AggregateFunction<Tuple2<String, Long>, Tuple2<Long, Long>, Double> {
  @Override
  public Tuple2<Long, Long> createAccumulator() {
    return new Tuple2<>(0L, 0L);
  }

  @Override
  public Tuple2<Long, Long> add(Tuple2<String, Long> value, Tuple2<Long, Long> accumulator) {
    return new Tuple2<>(accumulator.f0 + value.f1, accumulator.f1 + 1L);
  }

  @Override
  public Double getResult(Tuple2<Long, Long> accumulator) {
    return ((double) accumulator.f0) / accumulator.f1;
  }

  @Override
  public Tuple2<Long, Long> merge(Tuple2<Long, Long> a, Tuple2<Long, Long> b) {
    return new Tuple2<>(a.f0 + b.f0, a.f1 + b.f1);
  }
}

DataStream<Tuple2<String, Long>> input = ...;

input
    .keyBy(<key selector>)
    .window(<window assigner>)
    .aggregate(new AverageAggregate());
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

7.3 ProcessWindowFunction

ProcessWindowFunction获取包含窗口的所有数据元的Iterable，以及可访问时间和状态信息的Context对象，这使其能够提供比其他窗口函数更多的灵活性。这是以性能和资源消耗为代价的，因为数据元不能以递增方式聚合，而是需要在内部进行缓冲，直到窗口被认为已准备好进行处理。

public abstract class ProcessWindowFunction<IN, OUT, KEY, W extends Window> implements Function {

    public abstract void process(
            KEY key,
            Context context,
            Iterable<IN> elements,
            Collector<OUT> out) throws Exception;

   	/**
   	 * The context holding window metadata.
   	 */
   	public abstract class Context implements java.io.Serializable {
   	    /**
   	     * Returns the window that is being evaluated.
   	     */
   	    public abstract W window();

   	    /** Returns the current processing time. */
   	    public abstract long currentProcessingTime();

   	    /** Returns the current event-time watermark. */
   	    public abstract long currentWatermark();

   	    /**
   	     * State accessor for per-key and per-window state.
   	     *
   	     * <p><b>NOTE:</b>If you use per-window state you have to ensure that you clean it up
   	     * by implementing {@link ProcessWindowFunction#clear(Context)}.
   	     */
   	    public abstract KeyedStateStore windowState();

   	    /**
   	     * State accessor for per-key global state.
   	     */
   	    public abstract KeyedStateStore globalState();
   	}

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

该key参数是通过KeySelector为keyBy()调用指定的Keys提取的Keys。在元组索引键或字符串字段引用的情况下，此键类型始终是Tuple，您必须手动将其转换为正确大小的元组以提取键字段。

A ProcessWindowFunction可以像这样定义和使用：

DataStream<Tuple2<String, Long>> input = ...;

input
  .keyBy(t -> t.f0)
  .timeWindow(Time.minutes(5))
  .process(new MyProcessWindowFunction());

/* ... */

public class MyProcessWindowFunction
    extends ProcessWindowFunction<Tuple2<String, Long>, String, String, TimeWindow> {

  @Override
  public void process(String key, Context context, Iterable<Tuple2<String, Long>> input, Collector<String> out) {
    long count = 0;
    for (Tuple2<String, Long> in: input) {
      count++;
    }
    out.collect("Window: " + context.window() + "count: " + count);
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

该示例显示了ProcessWindowFunction对窗口中的数据元进行计数的情况。此外，窗口函数将有关窗口的信息添加到输出。

注意注意，使用ProcessWindowFunction简单的聚合（例如count）是非常低效的

package com.javaedge.java.chapter6;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class JavaWindowsProcessApp {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> text = env.socketTextStream("localhost", 9999);

        text.flatMap(new FlatMapFunction<String, Tuple2<Integer, Integer>>() {
                    @Override
                    public void flatMap(String value, Collector<Tuple2<Integer, Integer>> out) throws Exception {
                        String[] tokens = value.toLowerCase().split(",");
                        for (String token : tokens) {
                            if (token.length() > 0) {
                                out.collect(new Tuple2<>(1, Integer.parseInt(token)));
                            }
                        }
                    }
                }).keyBy(0)
                .timeWindow(Time.seconds(5))
                .process(new ProcessWindowFunction<Tuple2<Integer, Integer>, Object, Tuple, TimeWindow>() {
                    @Override
                    public void process(Tuple tuple, Context context, Iterable<Tuple2<Integer, Integer>> elements, Collector<Object> out) throws Exception {
                        long count = 0;
                        for (Tuple2<Integer, Integer> in : elements) {
                            count++;
                        }
                        out.collect("Window: " + context.window() + "count: " + count);
                    }
                })
                .print()
                .setParallelism(1);
        env.execute("JavaWindowsReduceApp");
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

8 水印

• 推荐阅读
  Flink流计算编程–watermark（水位线）简介

参考

Event Time

Windows