DataSets在ApacheFlink联盟

Question

我试图将一个Seq[DataSet(Long,Long,Double)]与Flink中的单个DataSet[(Long,Long,Double)]结合起来：

     val neighbors= graph.map(el => zKnn.neighbors(results,
      el.vector, 150, metric)).reduce(
     (a, b) => a.union(b)
      ).collect()

其中图是一个常规scala集合，但可以转换为DataSet；结果是一个DataSet[Vector]，不应该收集，并且在邻居方法中是必需的。

我总是得到FlinkRuntime的启示：

当前无法处理超过64个输出的节点。org.apache.flink.optimizer.CompilerException:目前无法处理超过64个输出的节点。在org.apache.flink.optimizer.dag.OptimizerNode.addOutgoingConnection(OptimizerNode.java:347) at org.apache.flink.optimizer.dag.SingleInputNode.setInput(SingleInputNode.java:202

Answer 1

Flink目前不支持有超过64个输入数据集的工会运营商。

作为一种解决方法，您可以将多达64个数据集分层合并，并在层次结构的各个级别之间插入一个标识映射器。类似于：

DataSet level1a = data1.union(data2.union(data3...(data64))).map(new IDMapper());
DataSet level1b = data65.union(data66...(data128))).map(new IDMapper());
DataSet level2 = level1a.union(level1b)