Spark 交换已经正确分布的分区答案

【问题标题】：Spark doing exchange of partitions already correctly distributedSpark 交换已经正确分布的分区
【发布时间】：2018-04-07 15:39:24
【问题描述】：

我通过两列连接 2 个数据集，结果是包含 550 亿行的数据集。之后，我必须通过与连接中使用的列不同的列对此 DS 进行一些聚合。问题是 Spark 在加入后进行交换分区（在 550 亿行上花费了太多时间），尽管数据已经正确分布，因为聚合列是唯一的。我知道聚合键已正确分配，有没有办法告诉 Spark 应用程序？

【问题讨论】：

标签： apache-spark apache-spark-sql spark-dataframe apache-spark-dataset databricks

【解决方案1】：

1) 转到 Spark UI 并检查“Locality Level”

2) 如果 Join 一个大数据和一个小数据使用 brodcast Join

3) 如果加入大中型数据，如果中型 RDD 不能完全装入内存，则使用过滤器

val keys = sc.broadcast(mediumRDD.map(_._1).collect.toSet)
val reducedRDD = largeRDD.filter{ case(key, value) => keys.value.contains(key) }
reducedRDD.join(mediumRDD)

4) 检查数据是否序列化

.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .set("spark.kryoserializer.buffer.max", "128m")
      .set("spark.kryoserializer.buffer", "64m")
      .registerKryoClasses(
        Array(classOf[ArrayBuffer[String]], classOf[ListBuffer[String]])

5) 检查 Spark UI 或在代码中添加以下代码以进行调试

df.rdd.getNumPartitions

Spark的应用UI，从下面的截图可以看出，“Total Tasks”代表分区的数量

【讨论】：