ApacheFlink 中的数据集联盟答案

【问题标题】：Union of DataSets in ApacheFlinkApacheFlink 中的数据集联盟
【发布时间】：2015-10-15 11:01:56
【问题描述】：

我正在尝试将 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()

其中graph是一个常规的scala集合但可以转换为DataSet； results 是一个DataSet[Vector]，不应该被收集，在邻居方法中是需要的

我总是得到一个 FlinkRuntime 异常：

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

【问题讨论】：

标签： scala bigdata apache-flink

【解决方案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)

【讨论】：