使用最新spark版本时如何设置spark.sql.shuffle.partitions

Question

我想重置 pyspark 代码中的 spark.sql.shuffle.partitions 配置，因为我需要加入两个大表。但是下面的代码在最新的spark版本中不起作用，报错说“no method "setConf" in xxx”

#!/usr/bin/python
# -*- coding: utf-8 -*-
import sys
import pyspark
from pyspark.context import SparkContext
from pyspark.sql.session import SparkSession
sc = SparkContext('local')
spark = SparkSession(sc)

spark.sparkContext.setConf("spark.sql.shuffle.partitions", "1000")
spark.sparkContext.setConf("spark.default.parallelism", "1000")

# or using the follow, neither is working 
spark.setConf("spark.sql.shuffle.partitions", "1000")
spark.setConf("spark.default.parallelism", "1000")

我现在想知道如何重置“spark.sql.shuffle.partitions”。

Answer 1

SparkSession 提供了一个 RuntimeConfig 接口来设置和获取 Spark 相关参数。您的问题的答案是：

spark.conf.set("spark.sql.shuffle.partitions", 1000)

参考：https://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.sql.RuntimeConfig

我错过了您的问题是关于 pyspark。 Pyspark 有一个类似的接口spark.conf。 参考：https://spark.apache.org/docs/latest/api/python/pyspark.sql.html?highlight=sparksession#pyspark.sql.SparkSession.conf

Answer 2

请注意，当随机分区设置为大于 2000 时，我们在 Spark SQL “Group By”/“Distinct”实现中发现了一个缺陷。我们使用大约 3000 条记录的数据集进行了测试，其中包含 38 列有大约 1800 条 38 列的唯一记录。

当我们运行 38 列且“spark.sql.shuffle.partitions”设置为 2001 的“Distinct”或“Group By”查询时，不同记录的计数少于 1800，例如 1794。但是，当我们将其设置为 2000 时，相同的查询给我们的记录数为 1800。

所以基本上，当随机分区大于 2000 时，Spark 会错误地删除一些记录。

我们使用 Spark v2.3.1 进行了测试，并将很快提交 Bug Jira。我需要准备一个测试数据来演示，但我们已经用我们的真实数据集确认了它。