flink使用10-通过Bulk iterator计算圆周率

迭代处理是批量处理处理中的常见操作, Flink 的 迭代计算支持两种模式, 分别是 Bulk Iteration (全量迭代计算) 和 Delt Iteration (增量迭代计算). 下面就一个计算圆周率的例子 来说一下使用 Bulk Iteration 都有哪几个步骤.

在 Bulk Iteration 中, 主要的步骤其实是分为3步, 第一步是指定最大循环次数, 第二步是指定在循环时的一个计算处理的过程, 最后一步就是调用计算过程, 指定结束条件. 具体代码如下所示

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
public class BulkIteration {

public static void main(String[] args) throws Exception {
// 获取执行环境
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
// 构建输出数据
DataSource<Integer> data = env.fromElements(0);
// 1. 指定循环次数
IterativeDataSet<Integer> loop = data.iterate(1000);
// 2. 指循环计算过程
MapOperator<Integer, Integer> process = loop.map(new MapFunction<Integer, Integer>() {
@Override
public Integer map(Integer i) throws Exception {
double x = Math.random();
double y = Math.random();
int result = (x * x + y * y) < 1 ? 1 : 0;
return i + result;
}
});
// 3. 使用 closeWith 调用计算过程
List<Integer> collect = loop.closeWith(process).collect();
// 输出最终结果
for (Integer i : collect) {
System.out.println( i / 1000.0 * 4);
}
}
}
0%