设置火花配置

Question

我正在尝试在 pyspark shell 中设置一些 spark 参数的配置。

我尝试了以下

spark.conf.set("spark.executor.memory", "16g")

为了检查执行器内存是否已设置，我做了以下 spark.conf.get("spark.executor.memory")

返回"16g"。

我试图通过sc 使用 sc._conf.get("spark.executor.memory")

然后返回"4g"。

为什么这两个返回不同的值以及设置这些配置的正确方法是什么。

另外，我正在摆弄一堆参数，比如 "spark.executor.instances" "spark.executor.cores" "spark.executor.memory" "spark.executor.memoryOverhead" "spark.driver.memory" "spark.driver.cores" "spark.driver.memoryOverhead" "spark.memory.offHeap.size" "spark.memory.fraction" "spark.task.cpus" "spark.memory.offHeap.enabled " "spark.rpc.io.serverThreads" "spark.shuffle.file.buffer"

有没有办法为所有变量设置配置。

编辑

我需要以编程方式设置配置。完成 spark-submit 或启动 pyspark shell 后如何更改它？我正在尝试减少我的作业的运行时间，我正在经历多次迭代，更改 spark 配置并记录运行时间。

Answer 1

您可以使用以下方法设置环境变量：（例如在spark-env.sh 中，只能单独使用）

SPARK_EXECUTOR_MEMORY=16g

您也可以设置spark-defaults.conf:

spark.executor.memory=16g

但是这些解决方案是硬编码的并且几乎是静态的，您希望为不同的作业设置不同的参数，但是，您可能需要设置一些默认值。

最好的办法是使用spark-submit:

spark-submit --executor-memory 16G 

以编程方式定义变量的问题是，其中一些需要在启动时定义，否则优先规则将接管，并且您在启动作业后所做的更改将被忽略。

编辑：

创建 SparkContext 时会查找每个执行程序的内存量。

和

一旦将 SparkConf 对象传递给 Spark，它就会被克隆，并且用户不能再对其进行修改。 Spark 不支持在运行时修改配置。

见：SparkConf Documentation

您是否尝试在创建 SparkContext 之前更改变量，然后运行迭代，停止 SparkContext 并更改变量以再次迭代？

import org.apache.spark.{SparkContext, SparkConf}

val conf = new SparkConf.set("spark.executor.memory", "16g")
val sc = new SparkContext(conf)
...
sc.stop()
val conf2 = new SparkConf().set("spark.executor.memory", "24g")
val sc2 = new SparkContext(conf2)

您可以使用以下方式调试您的配置：sc.getConf.toDebugString

见：Spark Configuration

任何指定为标志或属性文件中的值都将传递给应用程序并与通过 SparkConf 指定的值合并。直接在 SparkConf 上设置的属性具有最高优先级，然后是传递给 spark-submit 或 spark-shell 的标志，然后是 spark-defaults.conf 文件中的选项。

您需要确保您的变量没有以更高的优先级定义。

优先顺序：

• conf/spark-defaults.conf
• --conf 或 -c - spark-submit 使用的命令行选项
• SparkConf

我希望这会有所帮助。

Answer 2

在 Pyspark 中，

假设我想在代码中增加驱动程序内存和执行程序。我可以这样做：

conf = spark.sparkContext._conf.setAll([('spark.executor.memory', '23g'), ('spark.driver.memory','9.7g')])

查看更新的设置：

spark.sparkContext._conf.getAll()