Flink流处理之迭代案例

当前Flink将迭代的重心集中在批处理上，之前我们谈及了批量迭代和增量迭代主要是针对批处理（DataSet）API而言的，而且Flink为批处理中的迭代提供了针对性的优化。

可是对于流处理（DataStream），Flink相同提供了对迭代的支持。这一节我们主要来分析流处理中的迭代，我们将会看到流处理中的迭代相较于批处理有类似之处。但差异也是十分之明显。

可迭代的流处理程序同意定义“步函数”（step function）并将其内嵌到一个可迭代的流（IterativeStream）中。由于一个流处理程序可能永不终止，因此不同于批处理中的迭代机制，流处理中无法设置迭代的最大次数。取而代之的是，你能够指定等待反馈输入的最大时间间隔（假设超过该时间间隔没有反馈元素到来。那么该迭代将会终止）。通过应用split或filter转换，你能够指定流的哪一部分用于反馈给迭代头，哪一部分分发给下游。这里我们以filter作为演示样例来展示可迭代的流处理程序的API使用模式。

首先。基于输入流构建IterativeStream。这是一个迭代的起始。通常称之为迭代头：

IterativeStream<Integer> iteration = inputStream.iterate();

接着。我们指定一系列的转换操作用于表述在迭代过程中运行的逻辑（这里简单以map转换作为演示样例）。map API所接受的UDF就是我们上文所说的步函数：

DataStream<Integer> iteratedStream = iteration.map(/* this is executed many times */);

然后。作为迭代我们肯定须要有数据反馈给迭代头进行反复计算，所以我们从迭代过的流中过滤出符合条件的元素组成的部分流，我们称之为反馈流：

DataStream<Integer> feedbackStream = iteratedStream.filter(/* one part of the stream */);

将反馈流反馈给迭代头就意味着一个迭代的完整逻辑的完毕，那么它就能够“关闭”这个闭合的“环”了。通过调用IterativeStream的closeWith这一实例方法能够关闭一个迭代（也可表述为定义了迭代尾）。传递给closeWith的数据流将会反馈给迭代头：

iteration.closeWith(feedbackStream);

另外，一个惯用的模式是过滤出须要继续向前分发的部分流，这个过滤转换事实上定义的是“终止迭代”的逻辑条件，符合条件的元素将被分发给下游而不用于进行下一次迭代：

DataStream<Integer> output = iteratedStream.filter(/* some other part of the stream */);

跟分析批处理中的迭代一样，我们仍然以解决实际问题的案例作为切入点来看看流处理中的迭代跟批处理中的迭代有何不同。

首先描写叙述一下须要解决的问题：产生一个由一系列二元组（两个字段都是在一个区间内产生的正整数来作为斐波那契数列的两个初始值）构成的数据流。然后对该数据流中的二元组不断地迭代使其产生斐波那契数列，直到某次产生的值大于给定的阈值，则停止迭代并输出迭代次数。



这个案例假设拆分到对单个元素（二元组）的角度来看。其运行步骤例如以下图所看到的：





上面我们已经对问题的核心过程进行了分析。接下来我们会分步解决问题的构建迭代的流处理程序。

首先，我们先通过source函数创建初始的流对象inputStream：

DataStream<Tuple2<Integer, Integer>> inputStream = env.addSource(new RandomFibonacciSource());

该source函数会生成二元组序列，二元组的两个字段值是随机生成的作为斐波那契数列的初始值：

private static class RandomFibonacciSource
implements SourceFunction<Tuple2<Integer, Integer>> {

private Random random = new Random();
private volatile boolean isRunning = true;
private int counter = 0;

public void run(SourceContext<Tuple2<Integer, Integer>> ctx) throws Exception {
while (isRunning && counter < MAX_RANDOM_VALUE) {
int first = random.nextInt(MAX_RANDOM_VALUE / 2 - 1) + 1;
int second = random.nextInt(MAX_RANDOM_VALUE / 2 -1) + 1;

if (first > second) continue;

ctx.collect(new Tuple2<Integer, Integer>(first, second));
counter++;
Thread.sleep(50);
}
}

public void cancel() {
isRunning = false;
}
}

为了对新计算的斐波那契数列中的值以及累加的迭代次数进行存储，我们须要将二元组数据流转换为五元组数据流，并据此创建迭代对象：

IterativeStream<Tuple5<Integer, Integer, Integer, Integer, Integer>> iterativeStream =
inputStream.map(new TupleTransformMapFunction()).iterate(5000);

元组转换的map函数实现：

private static class TupleTransformMapFunction extends RichMapFunction<Tuple2<Integer,
Integer>, Tuple5<Integer, Integer, Integer, Integer, Integer>> {
public Tuple5<Integer, Integer, Integer, Integer, Integer> map(
Tuple2<Integer, Integer> inputTuples) throws Exception {
return new Tuple5<Integer, Integer, Integer, Integer, Integer>(
inputTuples.f0,
inputTuples.f1,
inputTuples.f0,
inputTuples.f1,
0);
}
}

在迭代流iterativeStream创建完毕之后，我们将基于它运行斐波那契数列的步函数并产生斐波那契数列流fibonacciStream：

DataStream<Tuple5<Integer, Integer, Integer, Integer, Integer>> fibonacciStream =
iterativeStream.map(new FibonacciCalcStepFunction());

当中用于计算斐波那契数列的步函数实现例如以下：

private static class FibonacciCalcStepFunction extends
RichMapFunction<Tuple5<Integer, Integer, Integer, Integer, Integer>,
Tuple5<Integer, Integer, Integer, Integer, Integer>> {
public Tuple5<Integer, Integer, Integer, Integer, Integer> map(
Tuple5<Integer, Integer, Integer, Integer, Integer> inputTuple) throws Exception {
return new Tuple5<Integer, Integer, Integer, Integer, Integer>(
inputTuple.f0,
inputTuple.f1,
inputTuple.f3,
inputTuple.f2 + inputTuple.f3,
++inputTuple.f4);
}
}

正如上文所述。后三个字段会产生变化。在计算之前，数列最后一个元素会被保留。也就是f3相应的元素，然后通过f2元素加上f3元素会产生最新值并更新f3元素。而f4则会累加。



上文我们对每一个元素计算斐波那契数列的新值并产生了fibonacciStream，可是我们须要对最新的两个值进行推断。看它们是否超过了指定的阈值。超过了阈值的元组将会被输出，而没有超过的则会再次參与迭代。因此这将产生两个不同的分支。我们也为此构建了分支流：

SplitStream<Tuple5<Integer, Integer, Integer, Integer, Integer>> branchedStream =
fibonacciStream.split(new FibonacciOverflowSelector());

而对是否超过阈值的元组进行推断并分离的实现例如以下：

private static class FibonacciOverflowSelector implements OutputSelector<
Tuple5<Integer, Integer, Integer, Integer, Integer>> {
public Iterable<String> select(
Tuple5<Integer, Integer, Integer, Integer, Integer> inputTuple) {
if (inputTuple.f2 < OVERFLOW_THRESHOLD && inputTuple.f3 < OVERFLOW_THRESHOLD) {
return Collections.singleton(ITERATE_FLAG);
}

return Collections.singleton(OUTPUT_FLAG);
}
}

在筛选方法select中，我们对不同的分支以不同的常量标识符进行标识：ITERATE_FLAG（还要继续迭代）和OUTPUT_FLAG（直接输出）。

产生了分支流之后。我们就能够从中检出不同的流分支做迭代或者输出处理。

对须要再次迭代的，就通过迭代流的closeWith方法反馈给迭代头：

iterativeStream.closeWith(branchedStream.select(ITERATE_FLAG));

而对于不须要的迭代就直接让其流向下游处理，这里我们仅仅是简单得将流“重构”了一下然后直接输出：

DataStream<Tuple3<Integer, Integer, Integer>> outputStream = branchedStream
.select(OUTPUT_FLAG).map(new BuildOutputTupleMapFunction());
outputStream.print();

所谓的重构就是将之前的五元组又一次缩减为三元组，实现例如以下：

private static class BuildOutputTupleMapFunction extends RichMapFunction<
Tuple5<Integer, Integer, Integer, Integer, Integer>,
Tuple3<Integer, Integer, Integer>> {
public Tuple3<Integer, Integer, Integer> map(Tuple5<Integer, Integer, Integer, Integer,
Integer> inputTuple) throws Exception {
return new Tuple3<Integer, Integer, Integer>(
inputTuple.f0,
inputTuple.f1,
inputTuple.f4);
}
}

终于我们将会得到类似例如以下的输出：

(7,14,5)

(18,37,3)

(3,46,3)

(23,32,3)

(31,43,2)

(13,45,2)

(37,42,2)

……



终于完整的主干程序代码例如以下：

public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment
.getExecutionEnvironment().setBufferTimeout(1);

DataStream<Tuple2<Integer, Integer>> inputStream = env.addSource(new RandomFibonacciSource());

IterativeStream<Tuple5<Integer, Integer, Integer, Integer, Integer>> iterativeStream =
inputStream.map(new TupleTransformMapFunction()).iterate(5000);

DataStream<Tuple5<Integer, Integer, Integer, Integer, Integer>> fibonacciStream =
iterativeStream.map(new FibonacciCalcStepFunction());

SplitStream<Tuple5<Integer, Integer, Integer, Integer, Integer>> branchedStream =
fibonacciStream.split(new FibonacciOverflowSelector());

iterativeStream.closeWith(branchedStream.select(ITERATE_FLAG));

DataStream<Tuple3<Integer, Integer, Integer>> outputStream = branchedStream
.select(OUTPUT_FLAG).map(new BuildOutputTupleMapFunction());

outputStream.print();
env.execute("Streaming Iteration Example");
}