什么是SparkSession配置选项

Question

我正在尝试使用SparkSession将文件的JSON数据转换为使用Spark Notebook的RDD.我已经有了JSON文件.

 val spark = SparkSession
   .builder()
   .appName("jsonReaderApp")
   .config("config.key.here", configValueHere)
   .enableHiveSupport()
   .getOrCreate()
val jread = spark.read.json("search-results1.json")

我很新兴火花,不知道该用什么config.key.here和configValueHere.

Answer 1

SparkSession

要获得SparkSession的所有"各种Spark参数作为键值对","使用Dataset和DataFrame API编程Spark的入口点",运行以下命令(这是使用spark python api,scala将非常相似) .

import pyspark
from pyspark import SparkConf
from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()
SparkConf().getAll()

根据您使用的API,请参阅以下内容之一:

1. https://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.sql.SparkSession
2. https://spark.apache.org/docs/latest/api/python/pyspark.sql.html?#pyspark.sql.SparkSession
3. https://spark.apache.org/docs/latest/api/java/org/apache/spark/sql/SparkSession.html

您可以通过运行以下代码获得SparkSession配置选项的更深层次列表.大多数是相同的,但还有一些额外的.我不确定你是否可以改变它们.

spark.sparkContext.getConf().getAll()

SparkContext

要获得SparkContext的所有"各种Spark参数作为键值对","Spark功能的主要入口点","与Spark集群的连接","......"以创建RDD,累加器和在该集群上广播变量,"运行以下内容.

spark.sparkContext._conf.getAll()  

根据您使用的API,请参阅以下内容之一:

1. https://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.SparkContext
2. https://spark.apache.org/docs/latest/api/python/pyspark.html#pyspark.SparkContext
3. https://spark.apache.org/docs/latest/api/java/org/apache/spark/SparkContext.html

火花参数

您应该获得一个包含"各种Spark参数作为键值对"的元组列表,类似于以下内容:

import pyspark
from pyspark import SparkConf, SparkContext 
spark_conf = SparkConf().setAppName("test")
spark = SparkContext(conf = spark_conf)
SparkConf().getAll()

根据您使用的API,请参阅以下内容之一:

1. https://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.SparkConf
2. https://spark.apache.org/docs/latest/api/python/pyspark.html?#pyspark.SparkConf
3. https://spark.apache.org/docs/latest/api/java/org/apache/spark/SparkConf.html

另请参阅:http:
//spark.apache.org/docs/latest/configuration.html#runtime-environment

设置Spark参数

每个元组都SparkConf可以设置:

SparkSession

[(u'spark.eventLog.enabled', u'true'),
 (u'spark.yarn.appMasterEnv.PYSPARK_PYTHON', u'/<yourpath>/parcels/Anaconda-4.2.0/bin/python'),
 ...
 ...
 (u'spark.yarn.jars', u'local:/<yourpath>/lib/spark2/jars/*')]

SparkContext

spark = (SparkSession
          .builder
          .appName("Your App Name")
          .config("spark.some.config.option1", "some-value")
          .config("spark.some.config.option2", "some-value")
          .getOrCreate())  

Answer 2

这就是我在 Scala 中添加 spark 或 hive 设置的方式：

{
    val spark = SparkSession
        .builder()
        .appName("StructStreaming")
        .master("yarn")
        .config("hive.merge.mapfiles", "false")
        .config("hive.merge.tezfiles", "false")
        .config("parquet.enable.summary-metadata", "false")
        .config("spark.sql.parquet.mergeSchema","false")
        .config("hive.merge.smallfiles.avgsize", "160000000")
        .enableHiveSupport()
        .config("hive.exec.dynamic.partition", "true")
        .config("hive.exec.dynamic.partition.mode", "nonstrict")
        .config("spark.sql.orc.impl", "native")
        .config("spark.sql.parquet.binaryAsString","true")
        .config("spark.sql.parquet.writeLegacyFormat","true")
        //.config(“spark.sql.streaming.checkpointLocation”, “hdfs://pp/apps/hive/warehouse/dev01_landing_initial_area.db”)
        .getOrCreate()
}

Answer 3

设置一些配置的最简单方法：

spark.conf.set("spark.sql.shuffle.partitions", 500)。

其中spark引用 a SparkSession，这样您就可以在运行时设置配置。当您想要一次又一次更改配置以调整特定查询的一些 Spark 参数时，它非常有用。