在Flink数据集中保存批量迭代的部分输出的可能性？

Question

val start = env.fromElements((0, BitSet.empty))
val end = start.iterateWithTermination(size) { inp =>
    val result = ObjData.mapPartition(new MyMapPartition).withBroadcastSet(inp, "concepts").groupBy(0).reduceGroup(new MyReduceGroup)
    (result,result)
}
end.count()

org.apache.flink.api.common.InvalidProgramException: A data set that is part of an iteration was used as a sink or action. Did you forget to close the iteration?

var start = env.fromElements((0, BitSet.empty))
var count = 1L
var all = count
while (count > 0){
    start = ObjData.mapPartition(new MyMapPartition).withBroadcastSet(start, "concepts").groupBy(0).reduceGroup(new MyReduceGroup)
    count = start.count()
    all = all + count
}
println("total nodes: " + all)

但是这种方法在最小的输入数据上特别慢，迭代版本需要<30秒，循环版本需要>3分钟。
我想flink无法创建执行循环的最佳计划。

有什么我应该尝试的变通方法吗？是否可以对flink进行一些修改，以便在hadoop等上保存部分结果？

Answer 1

不幸的是，目前不可能从批量迭代输出中间结果。您只能在迭代结束时输出最终结果。

而且，正如您正确地注意到的，Flink不能有效地展开while-loop或for-loop，所以这也不会起作用。

如果您的中间结果不是那么大，您可以尝试将中间结果添加到部分解决方案中，然后在迭代结束时输出所有结果。在TransitiveClosureNaive示例中实现了类似的方法，其中在迭代中发现的路径在下一个部分解决方案中累积。