处理数据倾斜而不加盐火花中的连接键

Question

我正在尝试将一百万行数据框与 30 行数据框进行内部连接，并且两个表都具有相同的连接键，spark 正在尝试执行排序合并连接，因此我的所有数据最终都在同一个执行程序和作业中例如，永远不会结束

DF1(million rows dataframe registered as TempView DF1)
+-------+-----------+
|   id  |  price    | 
+-------+-----------+
|    1  |   30      |
|    1  |   10      |
|    1  |   12      |
|    1  |   15      |
+-------+-----------+

DF2(30 rows dataframe registered as TempView DF2)
+-------+-----------+
|   id  |  Month    | 
+-------+-----------+
|    1  |   Jan     |
|    1  |   Feb     |
+-------+-----------+

我试过了

广播

spark.sql("Select /*+ BROADCAST(Df2) */ Df1.* from Df1 inner join Df2 on Df1.id=Df2.id").createTempView("temp")

重新分区

Df1.repartition(200)

查询执行计划

00 Project [.......................]
01 +- SortMergeJoin [.............................],Inner
02    :- Project [.............................]
03    :  +-Filter is notnull[JoinKey]
04    :    +- FileScan orc[..........................]
05    +-Project [.............................]
06      +-BroadcastHashJoin [..........................], LeftOuter, BuildRight
07        :- BroadCastHashJoin [......................],LeftSemi, BuildRight

分区数的输出

spark.table("temp").withColumn("partition_id",spark_partition_id).groupBy
("partition_id").count
+-------+---------------+
|    21 |300,00,000     |
+-------+---------------+

即使我重新分区/广播数据，spark 在加入时会将所有数据带到一个执行器，并且数据会在一个执行器处出现偏差。我还尝试将 spark.sql.join.preferSortMergeJoin 关闭为 false。但我仍然看到我的数据在一位执行人身上出现了偏差。谁能帮帮我？

Answer 1

只要这样做，它就可以正常工作。数据保持原样，没有分区。

 import org.apache.spark.sql.functions.broadcast
 // Simulate some data
 val df1 = spark.range(1000000).rdd.map(x => (1, "xxx")).toDF("one", "val")
 val df2 = spark.range(30).rdd.map(x => (1, "yyy")).toDF("one", "val2")
 // Data is as is, has no partitioning applied

 val df3 = df1.join(broadcast(df2), "one")  
 df3.count // An action to kick it all along

 // Look at final counts of partitions
 val rddcounts = df3.rdd.mapPartitions(iter => Array(iter.size).iterator, true) 
 rddcounts.collect

返回：

res26: Array[Int] = Array(3750000, 3750000, 3750000, 3750000, 3750000, 3750000, 3750000, 3750000)

这依赖于默认并行度，CE Databricks 集群上为 8。

广播应该在任何情况下都可以工作，因为小桌子很小。

即使这样：

val df = spark.range(1000000).rdd.map(x => (1, "xxx")).toDF("one", "val")
val df1 = df.repartition(50)

它与 50 个分区并行工作。这是循环分区，这意味着集群将获得分布在 N 个 Worker 上的分区，其中至少有 N 个 Executor。它不是散列的，如果所有值都相同，则通过指定导致偏斜的列来调用散列。 IE。所有数据的 1 个 Worker 上的同一分区。

QED：因此，并非所有人都只在一个 Executor 上工作，除非您只有一个用于 Spark 应用程序的 Executor 或应用了散列。

之后我在我的实验笔记本电脑上运行本地 [4]，数据由 4 个内核提供服务，因此实际上是 4 个执行器。不加盐，平行 4。所以，很奇怪你不能得到它，除非你散列。

如果在真实集群上，您可以看到 4 个并行任务，因此并非全部都在 1 个执行器上。

Answer 2

1. 为什么所有数据都转移到一个执行器？如果你只有相同的 DF1 中的 id(id:1) 并使用 id 加入 DF2 。根据 HashPartitioner id=1 的数据会一直一起移动。
2. 你发生广播加入了吗？在 spark UI 中检查它