Spark 2.0.0错误:PartitioningCollection要求其所有分区都具有相同的numPartitions

Question

我在Spark中加入一些DataFrames并且我一直收到以下错误:

PartitioningCollection requires all of its partitionings have the same numPartitions.

这似乎发生在我将两个DataFrame连接在一起后,每个看起来都相当合理,但加入后,如果我尝试从连接的DataFrame中获取一行,我会收到此错误.我真的只是想了解为什么会出现这个错误或者它背后的含义是什么,因为我似乎无法找到任何关于它的文档.

以下调用导致此异常:

val resultDataframe = dataFrame1
  .join(dataFrame2,     
    $"first_column" === $"second_column").take(2)

但我当然可以打电话

dataFrame1.take(2)

和

dataFrame2.take(2)

我也试过重新分区DataFrames,使用Dataset.repartition(numPartitions)或Dataset.coalesce(numParitions)在dataFrame1与dataFrame2加盟之前,以及resultDataFrame之后的加入,但似乎没有什么影响了错误.在一些粗略的谷歌搜索后,我无法找到其他人得到错误的参考...

Answer 1

我在最近几天遇到了同样的问题,当我在互联网上找不到任何引用时,我很失望.直到你的!

我要补充几件事:在数据帧(多个连接)上的一组非常复杂的操作之后,我得到了错误.此外,这些操作涉及从同一父数据帧生成的数据帧.我试图用一个最小的例子来复制它,但是从我的管道中提取它并不是一件容易的事.

我怀疑当DAG太复杂时,Spark可能在计算正确的计划时遇到麻烦.不幸的是,似乎如果它是Spark 2.0.0中的一个错误,那么夜间构建还没有修复它(我几天前尝试过2.0.2快照).

解决问题(暂时)的实用解决方案似乎是:在管道中写入磁盘(在某些时候)某些数据帧,然后再次读取它们.这有效地迫使Spark有一个更小,更易于管理的优化计划,而且它不再崩溃.当然,这只是一个临时修复.

Answer 2

我也有同样的问题.对我来说,它是从连接的select部分中删除一些列后发生的(而不是join子句本身).

我能够通过调用.repartition()数据帧来修复它.