机器学习和图计算应用，都会使用到迭代计算，Flink 通过在迭代算子中定义 Step 函数来实现迭代算法，迭代算法包括 Iterate 和 DeltaIterate 两种类型。

Iterate

Iterate Operator 是一种简单的迭代形式：每一轮迭代，Step function 消费整个数据集（前一个迭代的结果或初始数据集），计算出下一轮迭代的输入（也称为 Next Partial Solution），满足迭代的终止条件后，会输出最终迭代结果。

Iterate

分析上图的流程：

1. 迭代输入：前一个迭代的结果或初始数据集。
2. Step function：Step function 将在每次迭代中执行。由算子 map、reduce、join 等组成。
3. Next Partial Solution：在每次迭代中，Step function 的输出将反馈到下一次迭代。
4. 迭代结果：最后一次迭代的输出被写入 DataSink 或用作后续算子的输入。

有多个选项可指定迭代的终止条件：

• 最大迭代次数：没有任何其他条件，迭代将执行多次。
• 自定义聚合器收敛：迭代允许指定自定义聚合器和收敛条件，如SUM聚合已发出记录的数目（聚合器），如果该数目为零，则终止（收敛条件）。

伪代码表示：

IterationState state = getInitialState();

while (!terminationCriterion()) {
    state = step(state);
}

setFinalState(state);

示例：递增数字

在以下示例中，迭代地递增一组数字：

Iterate Example

1. 迭代输入：初始输入从数据源读取的五个记录（整数1到5）。
2. Step function：单个 map 算子，将整数字段从i增加到i+1。将应用于输入的每个记录。
3. Next Partial Solution：Step function 的输出（map 算子的输出），反馈给下一次迭代。
4. 迭代结果：经过十次迭代，最初的数字将被增加十，新的数据集为整数11到15。

// 1st           2nd                       10th
map(1) -> 2      map(2) -> 3      ...      map(10) -> 11
map(2) -> 3      map(3) -> 4      ...      map(11) -> 12
map(3) -> 4      map(4) -> 5      ...      map(12) -> 13
map(4) -> 5      map(5) -> 6      ...      map(13) -> 14
map(5) -> 6      map(6) -> 7      ...      map(14) -> 15

Delta Iterate

Delta Iterate 实现了增量迭代。有选择地修改其解中的数据元并演化解决方案，而不是完全重新计算它。

在适用的情况下，这会使算法更高效，因为解中的每个数据元不会都出现在每次迭代中。这样可以专注于解中的的 hot part，并保持 cold part 不受影响。

Delta Iterate

分析上图的流程：

1. 迭代输入：读取初始 Workset 和 Solution Set 作为第一次迭代的输入。
2. Step function：在每次迭代中执行。
3. Next Workset / Update Solution Set：Next Workset 驱动迭代计算，并将反馈到下一个迭代。此外，Solution Set 将被更新并隐式转发（不需要重建）。两个数据集都可以由 Step function 的不同 算子更新。
4. 迭代结果：在最后一次迭代之后，Solution Set 被写入 DataSink 或用作后续算子的输入。

Delta Iterate 的默认达到指定的最大迭代次数或当生成的下一个工作集为空时，迭代将终止。还可以指定自定义聚合器和收敛标准。

伪代码表示：

IterationState workset = getInitialState();
IterationState solution = getInitialSolution();

while (!terminationCriterion()) {
    (delta, workset) = step(workset, solution);

    solution.update(delta)
}

setFinalState(solution);

示例：在图表中传播最小值

在以下示例中，每个顶点（Vertex）都有一个ID和一个颜色（Color）。每个顶点将其顶点ID传播到相邻顶点。目标是为子图中的每个定点分配最小ID。如果顶点接收的ID小于当前的ID，则它将变为接收到的ID的顶点的颜色。

Delta Iterate Example

初始输入同时设置 Workset 和 Solution Set。在上图中，颜色显示了 Solution Set 的演变。在每次迭代中，最小ID的颜色在各自的子图中传播。同时，每次迭代的工作量（交换和比较顶点ID）都会减少，对应于 Workset 减小，在三次迭代后，Workset 的大小从所有七个顶点变为零，此时迭代终止。

---

Reference：
https://ci.apache.org/projects/flink/flink-docs-release-1.6/dev/batch/iterations.html