performance - SPARK mapToPair 和 saveToTextFile 我应该使用重新分区来优化性能

Question

spark.driver.memory=4g 
spark.executor.memory=39g
spark.yarn.executor.memoryoverhead=4
spark.dynamicAllocation.maxExecutors=10

Yarn Queue 有 750GB 和 150 个 vcore。

整体实现看起来像

JavaRDD<Collection<InternalObject>> finalRDD = parentJavaRDD
                                               .filter
                                               .mapToPair(<implementation of PairFunction>)
                                               .reduceByKey(<implemementation of Function2>)
                                               .values();
finalRDD
.map(<Impmenetation of Function)
.saveAsTextFile(outputHDFSPath);

当我在 YARN 历史服务器上看到 SPARK 执行程序日志时，我看到 9 个执行程序中有 1 个占用了 appx。与其他人在几分钟内完成相比，2 小时。



我还能在这里优化什么？
鉴于只有 1 个 executor 需要大约 2 小时，鉴于非常低的随机播放，我应该使用重新分区

.map(<Impmenetation of Function)
.repartition(finalRDD.partitions().size())
.saveAsTextFile(outputHDFSPath)

Answer 1

我建议使用近似正态分布的 key 重新分区您的数据，以便每个任务在几分钟内完成。

这将在两个方面有所帮助-

提高所有任务的整体写入性能。

如果任何任务失败，小块数据将很容易被 Spark 重新尝试

选择重新分区列的策略

选择重新分区列的策略完全取决于您正在处理的数据类型。

统计/行为信息。可提前从消费者处获得

如果您事先了解数据统计/行为，那么我建议您选择一个数据偏斜较少的列，并使用分区号(默认 200=spark.sql.shuffle.partitions)执行一些实验。在这里，您的目标应该是选择分区号，以便每个任务都应该在几分钟内完成。如果您不想执行任何实验，则可以使用默认分区号。

动态模式和数据行为

为此，您需要计算所有字段的偏度并选择具有最低偏度的字段。有多种统计技术可以找出正态分布，例如 - https://en.wikipedia.org/wiki/Kolmogorov%E2%80%93Smirnov_test等等。甚至可以使用 spark api -

// run a KS test for the sample versus a standard normal distribution
val testResult = Statistics.kolmogorovSmirnovTest(data, "norm", 0, 1)
println(testResult)

引用- https://spark.apache.org/docs/1.6.3/mllib-statistics.html#hypothesis-testing

如果您想使用一些虚拟逻辑，那么只需找到所有列的偏斜并选择重复率低的列。
- 列的示例逻辑

select a.f1, count(*) cnt from A a
 group by a.f1  
order by cnt desc
limit 1;

检查- Why is the final reduce step extremely slow in this MapReduce? (HiveQL, HDFS MapReduce)

附注。这将增加执行时间，因此建议尽可能使用选项 #1