scala - Spark Dataframe Join shuffle

Question

星火版本 1.6.0

我在两个具有 100 个分区的数据帧之间使用连接函数，该应用程序在一个集群上运行，我为每个 20 个执行程序使用 5 个内核，总共 100 个内核。

我的问题是，当我进行连接时，所有记录都在一个执行器上计算，而其他执行器不使用，如下图所示：

这会导致性能下降，因为所有数据都是使用一个执行程序针对其他 19 个可用执行程序计算的。

看起来 spark join 只在一个分区中“带来”所有记录，有没有办法避免这种情况？

为了确保它不会重新分配到 1，我还设置了这个 spark 属性：spark.sql.shuffle.partitions=100确实，两个输入数据帧有 100 个与输出数据帧相同的分区

标签： scalaapache-sparkdataframeapache-spark-sql

Answer 1

简短的回答：

这是因为您的数据，而不是因为火花。

长答案：

为了执行join操作，火花需要将具有相同键（您要加入的列的值）的数据移动到相同的工作人员。例如，如果您将 A 列与 B 列连接，则两个表中包含相同值的行将被移动到相同的工作人员，然后再连接。

此外 - 具有不同键的行也可能移动到同一个节点 - 这取决于您拥有的分区器。您可以在此处阅读更多信息- 但一般认为默认分区器存在 -HashPartitioner和RangePartitioner. 尽管使用了哪一个 - 它决定了哪个工人行。例如 - 如果您有 RangePartitioner 范围为 [0, 5)[5. 7)[7, 10] 然后键 1, 2, 3, 4 将全部交给同一个工人。如果您的数据中只有这些键 - 将只使用一名工作人员。