Spark：作业卡在 100 的最后 2 个任务上答案

【问题标题】：Spark : Job stuck on the last 2 tasks of 100Spark：作业卡在 100 的最后 2 个任务上
【发布时间】：2018-06-27 17:48:14
【问题描述】：

我是 Spark 的新手，我必须支持由我们的顾问编写的应用程序。我阅读并观看了大量有关 Spark 的信息，但我仍然在为正确调整工作而苦苦挣扎。

场景：

包含 5 条清理规则的 Java 类，我们将这些规则应用于包含 4 亿条记录的 RDD。
分区数设置为 1000。
最后的“动作”是在 S3 上写入，但在此之前我们将分区数减少到 100 个。
Spark UI 显示进度，但不幸的是在保存的最后阶段，任务卡在 98/100
我没有使用 .collect()，但我使用的是 .map() 和 Spark SQL。

这是我用来编写的代码：

rdd.coalesce(100)
   .write().mode("append")
   .partitionBy("year", "month")
   .format(SPARK_AVRO_PACKAGE)
   .save(appProps.getProperty(PAGEVIEW_CLEANSED));

我不确定是否应该努力改进代码或调整 spark/cluster 的性能。

更新：我认为这段代码的 sn-p 是我遇到的问题的原因。我在 SO (Spark not distributing load to tasks evenly) 上找到了类似的帖子，我只是不确定如何在我的情况下使用广播。

Dataset<Row> duplicatePrefetchPrerenderHashDS = 
            hashedPageViewDS
              .select(hashedPageViewDS.col(PREFETCH_PRERENDER_HASH))
              .groupBy(hashedPageViewDS.col(PREFETCH_PRERENDER_HASH))
              .count()
              .withColumnRenamed("count", "cnt")
              .where("cnt>1");

【问题讨论】：

短 - 数据倾斜。但我们不知道具体在哪里，除非您显示上游代码。
我怎么能和你分享这么多代码？
这是创建minimal reproducible example 的重点。但为了更简单，请查找编写器代码上游的第一个 join / groupByKey / groupBy + agg + 非归约函数（如 collect_list）/ Window 函数应用程序。当然coalesce 可能会导致其自身的问题。一种或另一种方式 - 问题不在这里或不单独在这里。
感谢您的支持。我可以确认我们使用合并来减少在写入 S3 和 join / groupBy (Spark SQL) 之前启动 Spark 作业 (1000) 时创建的分区数量。您认为自定义分区器是否有助于在所有分区之间平均重新分配记录？
您可以尝试检查结果分区大小以进行调试： rdd.foreachPartition(x=> println(x.size)) 如果此代码卡住，则您没有 s3 问题。如果尺寸真的不相等。使用 repartition(100) 而不是 coalesce(100)，它会给你等于分区

标签： apache-spark bigdata

【解决方案1】：

您可以采取多种方法：

您可以尝试并执行distributeBy("year", "month") 这将确保只有1 个分区写入每个文件夹。如果数据在年份和月份之间均匀分布。
如果问题实际上是某些年份的偏差。然后我会说使用 repartition(1000) 和使用 DistributeBy("year", "month","COL1")。在上面的示例中，COL1 将是一个几乎均匀分布的列，例如 DAY of MONTH 或 DATE。现在不是向每个分区写入 200 个（默认随机排序值）文件，COL1 将决定不写入文件（30 是 DAY OF MONTH）
另一个有用的方法是使用 repartition(100) 而不是 coalesce(100)，因为 repartition 会均匀分布数据，从而导致更多分区在输出中包含每个分区的数据。

【讨论】：

我已将 coalesce 更改为 repartition 但我的行为相同。我很想知道如何在 Java 中使用 distributeBy，值得一试！谢谢！
DataFrame API 中的等效分发是df.repartition($"key", 2)