API迁移指南

从Flink 1.2迁移到Flink 1.3

自Flink 1.2以来，有一些API已被更改。大多数更改都记录在其特定文档中。以下是API更改的综合列表以及升级到Flink 1.3时迁移详细信息的链接。

TypeSerializer 界面变化

这主要适用TypeSerializer于为其状态实施自定义的用户。

从Flink 1.3开始，添加了两个与保存点恢复的串行器兼容性相关的其他方法。有关如何实现这些方法的更多详细信息，请参阅处理序列化程序升级和兼容性。

ProcessFunction 总是一个 RichFunction

在Flink 1.2中，引入了ProcessFunction其丰富的变体RichProcessFunction。自Flink 1.3以来，RichProcessFunction已被删除，ProcessFunction现在始终RichFunction可以访问生命周期方法和运行时上下文。

Flink CEP库API更改

Flink 1.3中的CEP库附带了许多新函数，这些函数导致API发生了一些变化。有关详细信息，请访问CEP迁移文档。

从Flink核心工件中删除了Logger依赖项

在Flink 1.3中，为了确保用户可以使用他们自己的自定义日志记录框架，核心Flink工件现在可以清除特定的记录器依赖项。

示例和快速入门原型已经指定了记录器，不应受到影响。对于其他自定义项目，请确保添加记录器依赖项。例如，在Maven中pom.xml，您可以添加：

  1. <dependency>
  2.     <groupId>org.slf4j</groupId>
  3.     <artifactId>slf4j-log4j12</artifactId>
  4.     <version>1.7.7</version>
  5. </dependency>
  7. <dependency>
  8.     <groupId>log4j</groupId>
  9.     <artifactId>log4j</artifactId>
  10.     <version>1.2.17</version>
  11. </dependency>

从Flink 1.1迁移到Flink 1.2

如状态文档中所述，Flink有两种类型的状态：被Keys化状态和非被Keys化状态（也称为算子状态）。这两种类型都可用于 算子和用户定义的函数。本文档将指导您完成将Flink 1.1函数代码迁移到Flink 1.2的过程，并介绍Flink 1.2中引入的一些重要内部更改，这些更改涉及Flink 1.1中对齐窗口 算子的弃用（请参阅对齐处理时间窗口 算子）。

迁移过程将有两个目标：

  • 允许您的函数利用Flink 1.2中引入的新函数，例如重新缩放，

  • 确保您的新Flink 1.2作业能够从其Flink 1.1前身生成的保存点恢复执行。

按照本指南中的步骤 算子操作后，您可以将正在运行的作业从Flink 1.1迁移到Flink 1.2，只需在Flink 1.1作业中使用保存点并将其作为起点提供给Flink 1.2作业。这将允许Flink 1.2作业从其Flink 1.1前任中断的位置恢复执行。

示例用户函数

作为本文档其余部分的运行示例，我们将使用CountMapper和BufferingSink函数。第一个是具有被Keys化状态的函数的示例，而第二个是具有非被Keys化状态的函数。Flink 1.1中上述两个函数的代码如下：

  1. public class CountMapper extends RichFlatMapFunction<Tuple2<String, Integer>, Tuple2<String, Integer>> {
  3.     private transient ValueState<Integer> counter;
  5.     private final int numberElements;
  7.     public CountMapper(int numberElements) {
  8.         this.numberElements = numberElements;
  9.     }
  11.     @Override
  12.     public void open(Configuration parameters) throws Exception {
  13.         counter = getRuntimeContext().getState(
  14.             new ValueStateDescriptor<>("counter", Integer.class, 0));
  15.     }
  17.     @Override
  18.     public void flatMap(Tuple2<String, Integer> value, Collector<Tuple2<String, Integer>> out) throws Exception {
  19.         int count = counter.value() + 1;
  20.         counter.update(count);
  22.         if (count % numberElements == 0) {
  23.             out.collect(Tuple2.of(value.f0, count));
  24.             counter.update(0); // reset to 0
  25.         }
  26.     }
  27. }
  29. public class BufferingSink implements SinkFunction<Tuple2<String, Integer>>,
  30.     Checkpointed<ArrayList<Tuple2<String, Integer>>> {
  32.     private final int threshold;
  34.     private ArrayList<Tuple2<String, Integer>> bufferedElements;
  36.     BufferingSink(int threshold) {
  37.         this.threshold = threshold;
  38.         this.bufferedElements = new ArrayList<>();
  39.     }
  41.     @Override
  42.     public void invoke(Tuple2<String, Integer> value) throws Exception {
  43.         bufferedElements.add(value);
  44.         if (bufferedElements.size() == threshold) {
  45.             for (Tuple2<String, Integer> element: bufferedElements) {
  46.                 // send it to the sink
  47.             }
  48.             bufferedElements.clear();
  49.         }
  50.     }
  52.     @Override
  53.     public ArrayList<Tuple2<String, Integer>> snapshotState(
  54.         long checkpointId, long checkpointTimestamp) throws Exception {
  55.         return bufferedElements;
  56.     }
  58.     @Override
  59.     public void restoreState(ArrayList<Tuple2<String, Integer>> state) throws Exception {
  60.         bufferedElements.addAll(state);
  61.     }
  62. }

这CountMapper是一个RichFlatMapFunction假定按表格分组的输入流(word, 1)。该函数为每个传入的键（ValueState<Integer> counter）保存一个计数器，如果某个单词的出现次数超过用户提供的阈值，则会发出一个包含单词本身和出现次数的元组。

的BufferingSink是一个SinkFunction接收元件（的潜在的输出CountMapper），并直到达到某个用户指定的阈值，将它们发射到最终水槽之前缓冲它们。这是避免对数据库或外部存储系统进行许多昂贵调用的常用方法。为了以容错方式进行缓冲，缓冲数据元保存在列表（bufferedElements）中，该列表定期检查点。

状态API迁移

要利用Flink 1.2的新函数，应修改上面的代码以使用新的状态抽象。完成这些更改后，您将能够更改作业的并行度（向上或向下扩展），并确保您的新版本的作业将从其前任停止的位置开始。

Keys状态：在深入研究迁移过程的细节之前需要注意的是，如果您的函数只有被Keys化状态，那么Flink 1.1的完全相同的代码也适用于Flink 1.2，完全支持新函数和完全向后兼容性。可以仅针对更好的代码组织进行更改，但这仅仅是一种风格问题。

如上所述，本节的其余部分重点介绍非被Keys化状态。

重新缩放和新状态抽象

第一个修改是从旧Checkpointed<T extends Serializable>状态接口到新状态接口的转换。在Flink 1.2中，有状态函数可以实现更通用的CheckpointedFunction接口或ListCheckpointed<T extends Serializable>接口，它在语义上更接近旧接口Checkpointed。

在这两种情况中，非键合状态被预期是一个List的序列化的对象，彼此独立的，因而在重新缩放获再分配。换句话说，这些对象是可以重新分区非被Keys化状态的最细粒度。例如，如果并行性1 BufferingSink包含数据元的检查点状态，(test1, 2)并且(test2, 2)当将并行性增加到2时，(test1, 2)可能最终在任务0中，而(test2, 2)将进入任务1。

有关被Keys化状态和非被Keys化状态重新缩放的原则的更多详细信息，请参阅状态文档。

ListCheckpointed

该ListCheckpointed接口需要实现两种方法：

  1. List<T> snapshotState(long checkpointId, long timestamp) throws Exception;
  3. void restoreState(List<T> state) throws Exception;

它们的语义与旧Checkpointed界面中的对应物相同。唯一的区别是现在snapshotState()应该将对象列表返回到检查点，如前所述，并且restoreState必须在恢复时处理此列表。如果状态不是重新分区，可以随时返回Collections.singletonList(MY_STATE)的snapshotState()。更新的代码BufferingSink包括在下面：

  1. public class BufferingSinkListCheckpointed implements
  2.         SinkFunction<Tuple2<String, Integer>>,
  3.         ListCheckpointed<Tuple2<String, Integer>>,
  4.         CheckpointedRestoring<ArrayList<Tuple2<String, Integer>>> {
  6.     private final int threshold;
  8.     private transient ListState<Tuple2<String, Integer>> checkpointedState;
  10.     private List<Tuple2<String, Integer>> bufferedElements;
  12.     public BufferingSinkListCheckpointed(int threshold) {
  13.         this.threshold = threshold;
  14.         this.bufferedElements = new ArrayList<>();
  15.     }
  17.     @Override
  18.     public void invoke(Tuple2<String, Integer> value) throws Exception {
  19.         this.bufferedElements.add(value);
  20.         if (bufferedElements.size() == threshold) {
  21.             for (Tuple2<String, Integer> element: bufferedElements) {
  22.                 // send it to the sink
  23.             }
  24.             bufferedElements.clear();
  25.         }
  26.     }
  28.     @Override
  29.     public List<Tuple2<String, Integer>> snapshotState(
  30.             long checkpointId, long timestamp) throws Exception {
  31.         return this.bufferedElements;
  32.     }
  34.     @Override
  35.     public void restoreState(List<Tuple2<String, Integer>> state) throws Exception {
  36.         if (!state.isEmpty()) {
  37.             this.bufferedElements.addAll(state);
  38.         }
  39.     }
  41.     @Override
  42.     public void restoreState(ArrayList<Tuple2<String, Integer>> state) throws Exception {
  43.         // this is from the CheckpointedRestoring interface.
  44.         this.bufferedElements.addAll(state);
  45.     }
  46. }

如代码所示，更新的函数也实现了CheckpointedRestoring接口。这是出于向后兼容性原因，更多细节将在本节末尾解释。

CheckpointedFunction

该CheckpointedFunction接口需要再次执行两种方法：

  1. void snapshotState(FunctionSnapshotContext context) throws Exception;
  3. void initializeState(FunctionInitializationContext context) throws Exception;

与在Flink 1.1中一样，snapshotState()每当执行检查点时都会调用它，但是每次初始化用户定义的函数时，都会调用initializeState()它（它的对应部分restoreState()），而不是仅在我们从故障中恢复的情况下。鉴于此，initializeState()不仅是初始化不同类型状态的地方，而且还包括状态恢复逻辑。CheckpointedFunction接口的实现BufferingSink如下所示。

  1. public class BufferingSink implements SinkFunction<Tuple2<String, Integer>>,
  2.         CheckpointedFunction, CheckpointedRestoring<ArrayList<Tuple2<String, Integer>>> {
  4.     private final int threshold;
  6.     private transient ListState<Tuple2<String, Integer>> checkpointedState;
  8.     private List<Tuple2<String, Integer>> bufferedElements;
  10.     public BufferingSink(int threshold) {
  11.         this.threshold = threshold;
  12.         this.bufferedElements = new ArrayList<>();
  13.     }
  15.     @Override
  16.     public void invoke(Tuple2<String, Integer> value) throws Exception {
  17.         bufferedElements.add(value);
  18.         if (bufferedElements.size() == threshold) {
  19.             for (Tuple2<String, Integer> element: bufferedElements) {
  20.                 // send it to the sink
  21.             }
  22.             bufferedElements.clear();
  23.         }
  24.     }
  26.     @Override
  27.     public void snapshotState(FunctionSnapshotContext context) throws Exception {
  28.         checkpointedState.clear();
  29.         for (Tuple2<String, Integer> element : bufferedElements) {
  30.             checkpointedState.add(element);
  31.         }
  32.     }
  34.     @Override
  35.     public void initializeState(FunctionInitializationContext context) throws Exception {
  36.         checkpointedState = context.getOperatorStateStore().
  37.             getSerializableListState("buffered-elements");
  39.         if (context.isRestored()) {
  40.             for (Tuple2<String, Integer> element : checkpointedState.get()) {
  41.                 bufferedElements.add(element);
  42.             }
  43.         }
  44.     }
  46.     @Override
  47.     public void restoreState(ArrayList<Tuple2<String, Integer>> state) throws Exception {
  48.         // this is from the CheckpointedRestoring interface.
  49.         this.bufferedElements.addAll(state);
  50.     }
  51. }

在initializeState把参数作为一个FunctionInitializationContext。这用于初始化非被Keys化状态“容器”。这是一个类型的容器，ListState其中非被Keys化状态对象将在检查点存储：

this.checkpointedState = context.getOperatorStateStore().getSerializableListState("buffered-elements");

在初始化容器之后，我们使用isRestored()上下文的方法来检查我们是否在失败后恢复。如果是这样true，即我们正在恢复，则应用恢复逻辑。

如修改的代码所示，在状态初始化期间恢复的BufferingSink这个ListState被保存在类变量中以供将来使用snapshotState()。在那里，ListState被清除由先前的检查点包含的所有对象，然后填充我们要设置检查点新的。

作为旁注，被Keys化状态也可以在initializeState()方法中初始化。这可以使用FunctionInitializationContext给定的参数来完成，而不是RuntimeContextFlink 1.1的情况。如果CheckpointedFunction要在CountMapper示例中使用该接口，则open()可以删除旧方法，new snapshotState()和initializeState()方法如下所示：

  1. public class CountMapper extends RichFlatMapFunction<Tuple2<String, Integer>, Tuple2<String, Integer>>
  2.         implements CheckpointedFunction {
  4.     private transient ValueState<Integer> counter;
  6.     private final int numberElements;
  8.     public CountMapper(int numberElements) {
  9.         this.numberElements = numberElements;
  10.     }
  12.     @Override
  13.     public void flatMap(Tuple2<String, Integer> value, Collector<Tuple2<String, Integer>> out) throws Exception {
  14.         int count = counter.value() + 1;
  15.         counter.update(count);
  17.         if (count % numberElements == 0) {
  18.             out.collect(Tuple2.of(value.f0, count));
  19.             counter.update(0); // reset to 0
  20.         }
  21.     }
  23.     @Override
  24.     public void snapshotState(FunctionSnapshotContext context) throws Exception {
  25.         // all managed, nothing to do.
  26.     }
  28.     @Override
  29.     public void initializeState(FunctionInitializationContext context) throws Exception {
  30.         counter = context.getKeyedStateStore().getState(
  31.             new ValueStateDescriptor<>("counter", Integer.class, 0));
  32.     }
  33. }

请注意，该snapshotState()方法为空，因为Flink本身负责在检查点时SNAPSHOT托管的被Keys化状态。

向后兼容Flink 1.1

到目前为止，我们已经看到如何修改我们的函数以利用Flink 1.2引入的新函数。剩下的问题是“我可以确保我的修改后的（Flink 1.2）作业将从我从Flink 1.1运行的作业停止的位置开始吗？”。

答案是肯定的，而这样做的方式非常简单。对于被Keys化状态，您必须什么都不做。Flink将负责从Flink 1.1恢复状态。对于非被Keys化状态，新函数必须实现CheckpointedRestoring接口，如上面的代码所示。这有一个方法，熟悉Flink 1.1 restoreState()的旧Checkpointed接口。如修改后的代码所示BufferingSink，该restoreState()方法与其前身相同。

对齐处理时间窗口 算子

在Flink 1.1中，只有在没有指定的逐出器或触发器的处理时间上运行时，timeWindow()被Key化的数据流上的命令才会实例化特殊类型WindowOperator。这可以是一个AggregatingProcessingTimeWindowOperator或一个AccumulatingProcessingTimeWindowOperator。这两个 算子都被称为对齐窗口 算子，因为它们假设它们的输入数据元按顺序到达。这在处理时间内有效，因为数据元在到达窗口算子时获得挂钟时间的时间戳。这些 算子仅限于使用内存状态后台，并且具有优化的数据结构，用于存储利用有序输入数据元到达的每窗口数据元。

在Flink 1.2中，不推荐使用对齐的窗口 算子，并且所有窗口 算子操作都通过泛型WindowOperator。此迁移不需要更改Flink 1.1作业的代码，因为Flink将透明地读取Flink 1.1保存点中对齐的窗口 算子存储的状态，将其转换为与泛型相兼容的格式WindowOperator，并使用通用的WindowOperator。

注意尽管已弃用，但您仍然可以使用Flink 1.2中的对齐窗口 算子，通过特殊WindowAssigners介绍来实现此目的。这些assigners是SlidingAlignedProcessingTimeWindows和TumblingAlignedProcessingTimeWindowsassigners，滑动和翻滚分别窗口。使用对齐窗口的Flink 1.2作业必须是一项新工作，因为在使用这些 算子时无法从Flink 1.1保存点恢复执行。

注意对齐的窗口 算子不提供重新缩放函数，也不提供与Flink 1.1的向后兼容性。

在Flink 1.2中使用对齐窗口 算子的代码如下所示：

  1. // for tumbling windows
  2. DataStream<Tuple2<String, Integer>> window1 = source
  3.     .keyBy(0)
  4.     .window(TumblingAlignedProcessingTimeWindows.of(Time.of(1000, TimeUnit.MILLISECONDS)))
  5.     .apply(your-function)
  7. // for sliding windows
  8. DataStream<Tuple2<String, Integer>> window1 = source
  9.     .keyBy(0)
  10.     .window(SlidingAlignedProcessingTimeWindows.of(Time.seconds(1), Time.milliseconds(100)))
  11.     .apply(your-function)
  1. // for tumbling windows
  2. val window1 = source
  3.     .keyBy(0)
  4.     .window(TumblingAlignedProcessingTimeWindows.of(Time.of(1000, TimeUnit.MILLISECONDS)))
  5.     .apply(your-function)
  7. // for sliding windows
  8. val window2 = source
  9.     .keyBy(0)
  10.     .window(SlidingAlignedProcessingTimeWindows.of(Time.seconds(1), Time.milliseconds(100)))
  11.     .apply(your-function)