Checkpoint

• 目的
  为了保证程序发生故障时状态不丢也不错，它是保证状态一致性而不是数据一致性。
• 原理
  使用异步屏障快照Asynchronous Barrier Snapshotting（简称 ABS）算法（依赖于Chandy-Lamport算法的变种）实现分布式快照。
• 流程
  1）JobManager周期性产生Barrier，并广播给所有Source算子。
  2）Source算子收到Barrier后，生成自己的状态快照(包含数据源对应的offset/partition等信息)，然后把Barrier广播给下游所有非Source算子
  3）非Source算子收到某条上游算子的Barrier后，会阻塞此上游算子的输入流，把再Barrier之后流过来的数据先放到算子的缓冲区。等收到上游所有算子的Barrier后，此时才会进行生成自己的状态快照，然后把此算子的Barrier广播给下游所有非Source算子。(这里是Barrier对齐机制，保证数据不会被重复处理。当然，如果为了效率，也可以不进行对齐，此时数据会至少处理一次，可能导致数据被重复处理。对于数据的EXACTLY_ONCE来说，在1.11版本对于Barrier对齐机制进行了优化，因为对齐机制会导致checkpoint时间过长以及当作业出现反压时，从而加重作业的反压。此时引入了Unaligned Checkpoint机制，这个机制会导致接受到第一个Barrier时，不会阻塞此流后续数据的计算。这种方法也由坏处就是要持久化一部分缓存数据)
  4）当所有Sink算子完成checkpoint后，且向JobManager发送确认消息后，该次checkpoint完成。

State

• 状态类型
  1）原生状态（Raw State）
  Raw State是开发者自己管理的，需要自己序列化。
  2）托管状态（Managed State）
  Managed State是由Flink管理的，Flink帮忙存储、恢复和优化。Managed State再进行细分，由两种类型：Keyed State和Operator State。
  Keyed State：一个SubTask有多个State，每一个Key对应一个State。有ValueState，ListState，MapState等
  Operator State：一个SubTask有一个State。有ListState,BroadcastState等
• 状态后端
  Flink 内置了以下这些开箱即用的 state backends ：
  ①HashMapStateBackend：状态数据以 Java 对象的形式存储在堆中。
  ②EmbeddedRocksDBStateBackend：状态数据保存在 RocksDB 数据库中，数据被以序列化字节数组的方式存储，RocksDB 数据库默认将数据存储在 TaskManager 的数据目录。
  如果不设置，默认使用 HashMapStateBackend。
  在Flink1.13版本对状态后端进行了改进，帮助用户更好理解本地状态存储和 checkpoint 存储的区分。
  1）MemoryStateBackend
  旧版本的 MemoryStateBackend 等价于使用 HashMapStateBackend 和 JobManagerCheckpointStorage。
  2）FsStateBackend
  旧版本的 FsStateBackend 等价于使用 HashMapStateBackend 和 FileSystemCheckpointStorage。
  3）RocksDBStateBackend
  旧版本的 RocksDBStateBackend 等价于使用 EmbeddedRocksDBStateBackend 和 FileSystemCheckpointStorage。使用RocksDB状态后端可以设置增量检查点，RocksDB是一个基于LSM树的KV存储。

Time

• 时间语义
  Flink在1.12版本后默认使用Event Time
  1）处理时间（Process Time）数据进入Flink被处理的系统时间（Operator处理数据的系统时间）
  2）事件时间（Event Time）数据在数据源产生的时间，一般由事件中的时间戳描述，比如用户日志中的TimeStamp。
  3）摄取时间（Ingestion Time）数据进入Flink的时间，记录被Source节点观察到的系统时间。
• 水位线
  flink1.11中对flink的水印生成接口进行了重构，创建watermark主要有以下三种方式
  1）使用createWatermarkGenerator 创建watermark。
  2）使用固定延时策略生成水印，调用WatermarkStrategy中的静态方法forBoundedOutOfOrderness。
  3）使用单调递增的方式生成水印，调用WatermarkStrategy中的静态方法forMonotonousTimestamps。

Window

• 分类
  1）Keyed Windows

stream
       .keyBy(...)               <-  keyed versus non-keyed windows
       .window(...)              <-  required: "assigner"
      [.trigger(...)]            <-  optional: "trigger" (else default trigger)
      [.evictor(...)]            <-  optional: "evictor" (else no evictor)
      [.allowedLateness(...)]    <-  optional: "lateness" (else zero)
      [.sideOutputLateData(...)] <-  optional: "output tag" (else no side output for late data)
       .reduce/aggregate/apply()      <-  required: "function"
      [.getSideOutput(...)]      <-  optional: "output tag"

2）Non-Keyed Windows

stream
       .windowAll(...)           <-  required: "assigner"
      [.trigger(...)]            <-  optional: "trigger" (else default trigger)
      [.evictor(...)]            <-  optional: "evictor" (else no evictor)
      [.allowedLateness(...)]    <-  optional: "lateness" (else zero)
      [.sideOutputLateData(...)] <-  optional: "output tag" (else no side output for late data)
       .reduce/aggregate/apply()      <-  required: "function"
      [.getSideOutput(...)]      <-  optional: "output tag"

• 生命周期
  开始：当应该属于该窗口的第一个元素到达时，就会创建一个窗口。
  结束：当时间超过其结束时间戳加上用户指定的允许延迟时，该窗口将被完全删除。
  每个窗口都有一个触发器和一个函数。函数是用于窗口内数据的计算，触发器是决定此窗口的函数多会进行计算的条件。
• 类型
  1）Tumbling Windows(滚动窗口)

input
    .keyBy(<key selector>)
    .window(TumblingEventTimeWindows.of(Time.seconds(5)))
    .<windowed transformation>(<window function>);

2）Sliding Windows(滑动窗口)

input
    .keyBy(<key selector>)
    .window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
    .<windowed transformation>(<window function>);

3）Session Windows(会话窗口)

input
    .keyBy(<key selector>)
    .window(EventTimeSessionWindows.withGap(Time.minutes(10)))
    .<windowed transformation>(<window function>);

4）Global Windows（全局窗口）

input
    .keyBy(<key selector>)
    .window(GlobalWindows.create())
    .<windowed transformation>(<window function>);

• 窗口函数
  1）增量窗口函数

①ReduceFunction

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);
      }
    });

②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());

2）全量窗口函数
ProcessWindowFunction
这个函数更加灵活，但是延迟性更高，且牺牲一定存储空间。

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

    /**
     * Evaluates the window and outputs none or several elements.
     *
     * @param key The key for which this window is evaluated.
     * @param context The context in which the window is being evaluated.
     * @param elements The elements in the window being evaluated.
     * @param out A collector for emitting elements.
     *
     * @throws Exception The function may throw exceptions to fail the program and trigger recovery.
     */
    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();
   	}

}

3）增量和全量窗口函数混合使用
兼顾两者优点
ReduceFunction+ProcessWindowFunction
或
AggregateFunction+ProcessWindowFunction