scala—apacheflink中数据集的并集

anhgbhbe 于 2021-06-24 发布在 Flink

关注(0)|答案(1)|浏览(373)

我在试着和一个 Seq[DataSet(Long,Long,Double)] 一个单身汉 DataSet[(Long,Long,Double)] 在Flink：

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

其中graph是一个常规的scala集合，但可以转换为数据集；结果是 DataSet[Vector] 不应该被收集并且是邻居方法中需要的
我总是得到一个flinkruntime例外：
当前无法处理输出超过64个的节点。org.apache.flink.optimizer.compilereException:当前无法处理输出超过64个的节点。在org.apache.flink.optimizer.dag.optimizernode.addoutgoingconnection（optimizernode。java:347)在org.apache.flink.optimizer.dag.singleinputnode.setinput（singleinputnode。java:202

scala apache-flink bigdata

来源：https://stackoverflow.com/questions/31616178/union-of-datasets-in-apacheflink

1条答案

按热度按时间

wfauudbj1#

flink目前不支持输入数据集超过64个的联合运算符。
作为一种解决方法，您可以按层次结构合并多达64个数据集，并在层次结构的级别之间插入标识Map器。比如：

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)

赞(0）回复(0）举报 2021-06-24

我来回答

scala—apacheflink中数据集的并集

1条答案

相关问题

热门标签

最新问答