spark.sql.shuffle.partitions和spark.default.parallelism有什么区别？

Question

spark.sql.shuffle.partitions和spark.default.parallelism有什么不同

我尝试在SparkSQL中设置这两个任务，但第二阶段的任务号总是200。

Answer 1

在回答here时，spark.sql.shuffle.partitions配置了在为连接或聚合洗牌数据时使用的分区数量。

当用户未显式设置时，spark.default.parallelism是RDD中由join、reduceByKey和parallelize等转换返回的默认分区数。请注意，spark.default.parallelism似乎只适用于原始RDD，并且在处理数据帧时被忽略。

如果您正在执行的任务不是连接或聚合，并且您正在处理数据帧，那么设置这些设置将不会有任何效果。但是，您可以通过在代码中调用df.repartition(numOfPartitions) (不要忘记将其分配给一个新的val)来设置分区的数量。

要更改代码中的设置，只需执行以下操作：

sqlContext.setConf("spark.sql.shuffle.partitions", "300")
sqlContext.setConf("spark.default.parallelism", "300")

或者，也可以在使用spark-submit将作业提交到集群时进行更改

./bin/spark-submit --conf spark.sql.shuffle.partitions=300 --conf spark.default.parallelism=300

Answer 2

spark.default.parallelism是spark设置的默认分区数，默认为200。如果您想要增加分区的数量，那么可以在spark配置中或在运行spark SQL时应用spark.sql.shuffle.partitions属性来设置分区的数量。

正常情况下，当我们遇到内存拥塞时会使用此大小，我们会看到以下错误: spark error:java.lang.IllegalArgumentException: Size exceeds Integer.MAX_VALUE

因此设置您可以为每个分区分配256MB的分区，并可用于为您的进程设置。

另外，如果分区数接近2000，则将其增加到2000以上。因为spark对分区< 2000和>2000应用了不同的逻辑，这将通过减少内存占用来提高代码性能，因为如果> 2000，数据默认是高度压缩的。