spark 2.1.0 会话配置设置 (pyspark)答案

【问题标题】：spark 2.1.0 session config settings (pyspark)spark 2.1.0 会话配置设置 (pyspark)
【发布时间】：2017-06-12 16:26:12
【问题描述】：

我正在尝试覆盖 spark 会话/spark 上下文默认配置，但它正在选择整个节点/集群资源。

 spark  = SparkSession.builder
                      .master("ip")
                      .enableHiveSupport()
                      .getOrCreate()

 spark.conf.set("spark.executor.memory", '8g')
 spark.conf.set('spark.executor.cores', '3')
 spark.conf.set('spark.cores.max', '3')
 spark.conf.set("spark.driver.memory",'8g')
 sc = spark.sparkContext

当我将配置放入 spark submit 时它工作正常

spark-submit --master ip --executor-cores=3 --diver 10G code.py

【问题讨论】：

什么是资源管理器？ Spark Standalone/YARN
2.0 的其他方式是conf = (SparkConf().set("spark.executor.cores", "3")); spark = SparkSession.builder .master("ip").conf(conf=conf) .enableHiveSupport() .getOrCreate()
抱歉，都试过了，都没有成功。能不能试一次。我刚刚将我的 spark 更新为 2.2.0 快照以解决 64KB 代码大小问题 (SPARK-16845)。

标签： python apache-spark pyspark spark-dataframe

【解决方案1】：

您实际上并没有使用此代码覆盖任何内容。只是为了让您自己看看，请尝试以下操作。

一旦你启动 pyspark shell 输入：

sc.getConf().getAll()

这将显示所有当前配置设置。然后尝试您的代码并再次执行此操作。没有任何变化。

您应该做的是创建一个新配置并使用它来创建 SparkContext。这样做：

conf = pyspark.SparkConf().setAll([('spark.executor.memory', '8g'), ('spark.executor.cores', '3'), ('spark.cores.max', '3'), ('spark.driver.memory','8g')])
sc.stop()
sc = pyspark.SparkContext(conf=conf)

然后你可以像上面一样检查自己：

sc.getConf().getAll()

这应该反映您想要的配置。

【讨论】：

在 spark 2.1.0/2.2.0 中我们可以像这样定义 sc = pyspark.SparkContext 。没有传递参数的选项。
你是说不能传进去吗？文档仍将其列为参数，请参阅here
[见这里spark.apache.org/docs/latest/api/python/…。我不清楚现在的入口点是什么？
如果你指的是this这一行，那是指检查现有的火花上下文对象。因此，例如，当您启动 pyspark 时，sparkcontext 已经作为 sc 存在。键入sc 本质上等于键入SparkSession.SparkContext 并返回当前上下文对象。我的理解是您想创建一个具有不同配置的上下文。
我已经做了一些小改动，它工作了..谢谢你.. spark = SparkSession.builder.config(conf=conf1).getOrCreate() sc = spark.sparkContext 这里 conf1 是你上面定义的（conf = **）

【解决方案2】：

在 Spark 2.3.1

中更新配置

要更改默认 spark 配置，您可以按照以下步骤操作：

导入所需的类

from pyspark.conf import SparkConf
from pyspark.sql import SparkSession

获取默认配置

spark.sparkContext._conf.getAll()

更新默认配置

conf = spark.sparkContext._conf.setAll([('spark.executor.memory', '4g'), ('spark.app.name', 'Spark Updated Conf'), ('spark.executor.cores', '4'), ('spark.cores.max', '4'), ('spark.driver.memory','4g')])

停止当前 Spark 会话

spark.sparkContext.stop()

创建 Spark 会话

spark = SparkSession.builder.config(conf=conf).getOrCreate()

【讨论】：

也适用于 2.2.0。感谢您提供此答案。
我在没有重新创建会话的情况下使用了spark.sparkContext._conf.setAll([('spark.executor.memory', '4g'), ...])（这意味着我没有使用你的最后两个步骤）。当我之后查看配置时（spark.sparkContext._conf.getAll()），我也看到了设置的参数。但是，我不确定我是否只是覆盖了对象，只是将配置保存在 sparkConf..
@Markus：你可以在 Spark UI 中查看配置
你在 conf 中使用了 'spark' 变量，然后在 spark 中使用了 'conf' 变量，哈哈。启动会话后如何更改 spark 配置？？
@Markus，您覆盖了 spark.sparkContext._conf 对象中的条目，但这确实影响了您的 spark 对象的真实属性。 SparkSession 对象的真正属性是传递给对象构造函数的属性。

【解决方案3】：

你也可以在启动 pyspark 时设置配置，就像 spark-submit 一样：

pyspark --conf property=value

这是一个例子

-bash-4.2$ pyspark
Python 3.6.8 (default, Apr 25 2019, 21:02:35) 
[GCC 4.8.5 20150623 (Red Hat 4.8.5-36)] on linux
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /__ / .__/\_,_/_/ /_/\_\   version 2.4.0-cdh6.2.0
      /_/

Using Python version 3.6.8 (default, Apr 25 2019 21:02:35)
SparkSession available as 'spark'.
>>> spark.conf.get('spark.eventLog.enabled')
'true'
>>> exit()


-bash-4.2$ pyspark --conf spark.eventLog.enabled=false
Python 3.6.8 (default, Apr 25 2019, 21:02:35) 
[GCC 4.8.5 20150623 (Red Hat 4.8.5-36)] on linux
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /__ / .__/\_,_/_/ /_/\_\   version 2.4.0-cdh6.2.0
      /_/

Using Python version 3.6.8 (default, Apr 25 2019 21:02:35)
SparkSession available as 'spark'.
>>> spark.conf.get('spark.eventLog.enabled')
'false'

【讨论】：

【解决方案4】：

在配置中将“spark.driver.host”设置为“localhost”对我有用

spark = SparkSession \
    .builder \
    .appName("MyApp") \
    .config("spark.driver.host", "localhost") \
    .getOrCreate()

【讨论】：

【解决方案5】：

我有一个非常不同的要求，我必须检查我是否获取了执行器和驱动程序内存大小的参数，如果获取，则必须仅用执行器和驱动程序的更改来替换配置。步骤如下：

导入库

from pyspark.conf import SparkConf
from pyspark.sql import SparkSession

定义 Spark 并获取默认配置

spark = (SparkSession.builder
        .master("yarn")
        .appName("experiment") 
        .config("spark.hadoop.fs.s3a.multiobjectdelete.enable", "false")
        .getOrCreate())

conf = spark.sparkContext._conf.getAll()

检查是否存在执行程序和驱动程序大小（我在这里给出伪代码 1 条件检查，其余您可以创建案例）然后使用基于参数的给定配置或跳到默认配置。

if executor_mem is not None and driver_mem  is not None:
    conf = spark.sparkContext._conf.setAll([('spark.executor.memory',executor_mem),('spark.driver.memory',driver_mem)])
    spark.sparkContext.stop()
    spark = SparkSession.builder.config(conf=conf).getOrCreate()
else:
    spark = spark

不要忘记停止 spark 上下文，这将确保执行程序和驱动程序的内存大小在您传入参数时有所不同。希望这会有所帮助！

【讨论】：

【解决方案6】：

我知道这是一个很小的旧帖子，并且有一些已经被接受的答案，但我只是想发布一个相同的工作代码。

from pyspark.sql import SparkSession

spark = SparkSession \
    .builder \
    .appName("MyApp") \
    .config("spark.driver.host", "localhost") \
    .getOrCreate()

default_conf = spark.sparkContext._conf.getAll()
print(default_conf)

conf = spark.sparkContext._conf.setAll([('spark.executor.memory', '4g'),
                                        ('spark.app.name', 'Spark Updated Conf'),
                                        ('spark.executor.cores', '4'),
                                        ('spark.cores.max', '4'),
                                        ('spark.driver.memory','4g')])

spark.sparkContext.stop()

spark = SparkSession \
    .builder \
    .appName("MyApp") \
    .config(conf=conf) \
    .getOrCreate()


default_conf = spark.sparkContext._conf.get("spark.cores.max")
print("updated configs " , default_conf)

【讨论】：