插入前火花洗牌数据答案

【问题标题】：Spark shuffling data before insert插入前火花洗牌数据
【发布时间】：2021-09-22 06:02:30
【问题描述】：

CalcDf().show 对show 本身产生 13 个阶段 (0-12) +1(13)。

当我尝试将结果写入表格时，我假设应该只有 13 个阶段 (0-12)，而不是我看到了额外的阶段 (13)。它来自哪里，它有什么作用？我没有执行任何重新分区或其他需要随机播放的操作。据我了解，火花应该只是将 1100 个文件写入表中，但这不是正在发生的事情。

CalcDf()
.write
.mode(SaveMode.Overwrite)
.insertInto("tn")

CalcDf() 逻辑

val dim = spark.sparkContext.broadcast(
spark.table("dim")
.as[Dim]
.map(r => r.id-> r.col)
.collect().toMap
)

spark.table("table")
.as[CustomCC]
.groupByKey(_.id)
.flatMapGroups{case(k, iterator) => CustomCC.mapRows(iterator, dim)}
.withColumn("time_key", lit("2021-07-01"))

【问题讨论】：

添加表tn的详细信息，是否已分区？
@vikas 它已分区。但是对于每一行.withColumn("time_key", lit("2021-07-01"))，硬编码的密钥都是相同的。

标签： scala apache-spark hadoop

【解决方案1】：

前一阶段 #12 已完成随机写入，因此任何后续阶段都必须通过随机读取从中读取数据（您在 #13 中注意到了这一点）。

为什么要增加一个阶段？

因为第 12 阶段有随机写入而不是输出

为了解第 12 阶段，请提供有关如何构建 CalDf 的信息。

编辑

groupByKey 将进行随机写入，以在单个执行程序 JVM 上获取相同的 id。

stage 13 是读取这个打乱后的数据并计算 map 操作。

任务计数的差异可归因于操作。
在 show() 中，它还没有读取整个洗牌的数据。可能是因为它显示 20 行（默认）而在 insertInto(...) 中，它对整个数据进行操作，因此读取所有数据。

stage #13 不仅仅是因为它正在写入文件，它实际上是在进行计算。

【讨论】：