处理dataskew而不加盐的连接键在火花

Question

我正在尝试将一个100万行数据帧与一个30行数据帧进行内部连接，这两个表都有相同的连接键，spark正在尝试执行排序合并连接，因此我的所有数据最终都在同一个执行器中，例如，Job从未完成

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     |
+-------+---------------+

即使我重新分区/广播数据，火花在加入时将所有数据带到一个执行器，数据在一个执行器上会发生倾斜。我还尝试将sortMergeJoinspark.sql.join.prefer关闭为false。但是我仍然看到我的数据在一个执行器上发生倾斜。有人能帮我吗？

Answer 1

1. 为什么所有数据都要移动到一个执行器？如果您在DF1中只有相同的id（id:1），并使用id连接DF2。根据HashPartitioner，id为1的数据将始终一起移动。
2. 你有广播加入吗？在spark UI中检查

Answer 2

只是这样做，它工作正常。数据按原样，没有分区。

 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 个执行程序的 N 个工作线程上的分区。它不是散列的，如果所有值都相同，则通过指定导致偏度的列来调用哈希。即 1 个 Worker 上所有数据的相同分区。

QED：所以，并不是所有人都只在一个执行器上工作，除非您只有一个用于Spark应用程序或哈希的执行器。

之后，我在我的实验性笔记本电脑上运行了local[4]，数据由4个内核提供服务，因此可以说是4个执行器。无盐渍，平行4。所以，奇怪的是你无法得到它，除非你散列。

您可以看到4个并行任务，因此，如果在一个真实的集群上，并不是所有的任务都在一个执行器上。