我正在尝试使用 SparkSession 通过 Spark Notebook 将文件的 JSON 数据转换为 RDD。我已经有了 JSON 文件。
val spark = SparkSession
.builder()
.appName("jsonReaderApp")
.config("config.key.here", configValueHere)
.enableHiveSupport()
.getOrCreate()
val jread = spark.read.json("search-results1.json")
我是 spark 新手，不知道 config.key.here 和 configValueHere 使用什么。
【问题讨论】：
标签： json apache-spark spark-notebook
要为 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()
或者不导入SparkConf:
spark.sparkContext.getConf().getAll()
根据您使用的 API，请参阅以下内容之一：
您可以通过运行以下代码获得更深层次的 SparkSession 配置选项列表。大多数是相同的，但还有一些额外的。我不确定您是否可以更改这些。
spark.sparkContext._conf.getAll()
要获取 SparkContext 的所有“作为键值对的各种 Spark 参数”、“Spark 功能的主要入口点”、...“连接到 Spark 集群”、...以及“创建 RDD” ，该集群上的累加器和广播变量，”运行以下命令。
import pyspark
from pyspark import SparkConf, SparkContext
spark_conf = SparkConf().setAppName("test")
spark = SparkContext(conf = spark_conf)
SparkConf().getAll()
根据您使用的 API，请参阅以下内容之一：
您应该得到一个包含“作为键值对的各种 Spark 参数”的元组列表，类似于以下内容：
[(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/*')]
根据您使用的 API，请参阅以下内容之一：
有关 Spark 属性的完整列表，请参阅：
http://spark.apache.org/docs/latest/configuration.html#viewing-spark-properties
每个元组都是("spark.some.config.option", "some-value")，您可以在应用程序中设置：
spark = (
SparkSession
.builder
.appName("Your App Name")
.config("spark.some.config.option1", "some-value")
.config("spark.some.config.option2", "some-value")
.getOrCreate())
sc = spark.sparkContext
spark_conf = (
SparkConf()
.setAppName("Your App Name")
.set("spark.some.config.option1", "some-value")
.set("spark.some.config.option2", "some-value"))
sc = SparkContext(conf = spark_conf)
您还可以在spark-defaults.conf 文件中设置 Spark 参数：
spark.some.config.option1 some-value
spark.some.config.option2 "some-value"
然后使用spark-submit (pyspark) 运行您的 Spark 应用程序：
spark-submit \
--properties-file path/to/your/spark-defaults.conf \
--name "Your App Name" \
--py-files path/to/your/supporting/pyspark_files.zip \
--class Main path/to/your/pyspark_main.py
【讨论】：
spark.sparkContext.getConf().getAll() 提供的信息比 SparkConf().getAll() 更多。
这就是我在我的 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()
}
【讨论】：
设置一些配置的最简单方法：
spark.conf.set("spark.sql.shuffle.partitions", 500).
其中spark 指的是SparkSession，这样您就可以在运行时设置配置。当您想要一次又一次地更改配置以针对特定查询调整一些 spark 参数时，它非常有用。
【讨论】：
简单来说，“config”方法中设置的值会自动传播到 SparkConf 和 SparkSession 自己的配置中。
例如： 你可以参考 https://jaceklaskowski.gitbooks.io/mastering-apache-spark/content/spark-sql-settings.html 了解如何使用配置选项为 SparkSession 设置 Hive 仓库位置
要了解这个api可以参考：https://spark.apache.org/docs/2.0.1/api/java/org/apache/spark/sql/SparkSession.Builder.html
【讨论】：
val spark = SparkSession .builder() .appName("jsonReaderApp") .config("spark.sql.json.rdd2", 2) .getOrCreate() 和 val jread = spark.read.json("bin/flatjson.json") 唯一的问题是我无法使用从流中获取的 JSON 数据。我必须更改文件的格式；每行必须是一个对象，对象之间不能有逗号。
每个 Spark 配置选项都在以下位置进行了说明：http://spark.apache.org/docs/latest/configuration.html
您可以像上面的示例那样在运行时设置这些，也可以通过给 spark-submit 的配置文件设置这些
【讨论】：