創(chuàng)建SparkSession和sparkSQL的詳細(xì)過程
一、概述
spark 有三大引擎，spark core、sparkSQL、sparkStreaming，
spark core 的關(guān)鍵抽象是 SparkContext、RDD；
SparkSQL 的關(guān)鍵抽象是 SparkSession、DataFrame；
sparkStreaming 的關(guān)鍵抽象是 StreamingContext、DStream
SparkSession 是 spark2.0 引入的概念，主要用在 sparkSQL 中，當(dāng)然也可以用在其他場(chǎng)合，他可以代替 SparkContext；
SparkSession 其實(shí)是封裝了 SQLContext 和 HiveContext
(1) 在Spark1.6 中我們使用的叫Hive on spark，主要是依賴hive生成spark程序，有兩個(gè)核心組件SQLcontext和HiveContext。
這是Spark 1.x 版本的語(yǔ)法
//set up the spark configuration and create contexts
val sparkConf = new SparkConf().setAppName("SparkSessionZipsExample").setMaster("local")
// your handle to SparkContext to access other context like SQLContext
val sc = new SparkContext(sparkConf).set("spark.some.config.option", "some-value")
val sqlContext = new org.apache.spark.sql.SQLContext(sc)
（2）Spark2.0中我們使用的就是sparkSQL，是后繼的全新產(chǎn)品，解除了對(duì)Hive的依賴。
從Spark2.0以上的版本開始，spark是使用全新的SparkSession接口代替Spark1.6 中的SQLcontext和HiveContext 來實(shí)現(xiàn)對(duì)數(shù)據(jù)的加載、轉(zhuǎn)換、處理等工作，并且實(shí)現(xiàn)了SQLcontext和HiveContext的所有功能。
在新版本中并不需要之前那么繁瑣的創(chuàng)建很多對(duì)象，只需要?jiǎng)?chuàng)建一個(gè)SparkSession對(duì)象即可。SparkSession支持從不同的數(shù)據(jù)源加載數(shù)據(jù)，并把數(shù)據(jù)轉(zhuǎn)換成DataFrame，并支持把DataFrame轉(zhuǎn)換成SQLContext自身中的表。然后使用SQL語(yǔ)句來操作數(shù)據(jù)，也提供了HiveQL以及其他依賴于Hive的功能支持。
二、創(chuàng)建SparkSession
SparkSession 是 Spark SQL 的入口。使用 Dataset 或者 Dataframe 編寫 Spark SQL 應(yīng)用的時(shí)候，第一個(gè)要?jiǎng)?chuàng)建的對(duì)象就是 SparkSession。Builder 是 SparkSession 的構(gòu)造器。 通過 Builder, 可以添加各種配置，并通過 stop 函數(shù)來停止 SparkSession。
Builder 的方法如下：
import org.apache.spark.sql.SparkSession
val spark: SparkSession = SparkSession.builder
.appName("My Spark Application") //設(shè)置 application 的名字
.master("local[*]")
.enableHiveSupport() //增加支持 hive Support
.config("spark.sql.warehouse.dir", "target/spark-warehouse") //設(shè)置各種配置
.getOrCreate //獲取或者新建一個(gè) sparkSession
（1）設(shè)置參數(shù)
創(chuàng)建SparkSession之后可以通過 spark.conf.set 來設(shè)置運(yùn)行參數(shù)
//配置設(shè)置
spark.conf.set("spark.sql.shuffle.partitions", 6)
spark.conf.set("spark.executor.memory", "2g")
//獲取配置，可以使用Scala的迭代器來讀取configMap中的數(shù)據(jù)。
val configMap:Map[String, String] = spark.conf.getAll()
（2）讀取元數(shù)據(jù)
如果需要讀取元數(shù)據(jù)(catalog)，可以通過SparkSession來獲取。
spark.catalog.listTables.show(false) spark.catalog.listDatabases.show(false)
這里返回的都是Dataset，所以可以根據(jù)需要再使用Dataset API來讀取
catalog 和 schema 是兩個(gè)不同的概念
Catalog是目錄的意思，從數(shù)據(jù)庫(kù)方向說，相當(dāng)于就是所有數(shù)據(jù)庫(kù)的集合;
Schema是模式的意思, 從數(shù)據(jù)庫(kù)方向說, 類似Catelog下的某一個(gè)數(shù)據(jù)庫(kù);
（3）創(chuàng)建Dataset和Dataframe
通過SparkSession來創(chuàng)建Dataset和Dataframe有多種方法。
通過range()方法來創(chuàng)建dataset
通過createDataFrame()來創(chuàng)建dataframe。
// create a Dataset using spark.range starting from 5 to 100,
// with increments of 5
val numDS = spark.range(5, 100, 5)//創(chuàng)建dataset
// reverse the order and display first 5 items
numDS.orderBy(desc("id")).show(5)
//compute descriptive stats and display them
numDs.describe().show()
// create a DataFrame using spark.createDataFrame from a List or Seq
val langPercentDF = spark.createDataFrame( List( ("Scala", 35),
("Python", 30), ("R", 15), ("Java", 20)) )//創(chuàng)建dataframe
//rename the columns
val lpDF = langPercentDF.withColumnRenamed("_1", "language").
withColumnRenamed("_2", "percent")
//order the DataFrame in descending order of percentage
lpDF.orderBy(desc("percent")).show(false)
（4）讀取數(shù)據(jù)
可以用SparkSession讀取JSON、CSV、TXT 和 parquet表。
import spark.implicits //使RDD轉(zhuǎn)化為DataFrame以及后續(xù)SQL操作 //讀取JSON文件，生成DataFrame val jsonFile = args(0) val zipsDF = spark.read.json(jsonFile)
（5）使用SparkSQL
借助SparkSession用戶可以像SQLContext一樣使用Spark SQL的全部功能。
zipsDF.createOrReplaceTempView("zips_table")//對(duì)上面的dataframe創(chuàng)建一個(gè)表
zipsDF.cache()//緩存表
val resultsDF = spark.sql("SELECT city, pop, state, zip FROM zips_table")
//對(duì)表調(diào)用SQL語(yǔ)句
resultsDF.show(10)//展示結(jié)果
（6）存儲(chǔ)/讀取Hive表
下面的代碼演示了通過SparkSession來創(chuàng)建Hive表并進(jìn)行查詢的方法。
//drop the table if exists to get around existing table error
spark.sql("DROP TABLE IF EXISTS zips_hive_table")
//save as a hive table
spark.table("zips_table").write.saveAsTable("zips_hive_table")
//make a similar query against the hive table
val resultsHiveDF = spark.sql("SELECT city, pop, state,
zip FROM zips_hive_table WHERE pop > 40000")
resultsHiveDF.show(10)
三、 SQLContext
它是 sparkSQL 的入口點(diǎn)，sparkSQL 的應(yīng)用必須創(chuàng)建一個(gè) SQLContext 或者 HiveContext 的類實(shí)例
from pyspark import SparkContext, SparkConf
from pyspark.sql import SparkSession, SQLContext, HiveContext
conf = SparkConf().setAppName('test').setMaster('yarn')
sc = SparkContext(conf=conf)
sqlc = SQLContext(sc)
print(dir(sqlc))
# 'cacheTable', 'clearCache', 'createDataFrame', 'createExternalTable', 'dropTempTable', 'getConf', 'getOrCreate', 'newSession', 'range', 'read', 'readStream',
# 'registerDataFrameAsTable', 'registerFunction', 'registerJavaFunction', 'setConf', 'sparkSession', 'sql', 'streams', 'table', 'tableNames', 'tables', 'udf', 'uncacheTable'
### sqlcontext 讀取數(shù)據(jù)也自動(dòng)生成 df
data = sqlc.read.text('/usr/yanshw/test.txt')
print(type(data))
四、 HiveContext
它是 sparkSQL 的另一個(gè)入口點(diǎn)，它繼承自 SQLContext，用于處理 hive 中的數(shù)據(jù)
HiveContext 對(duì) SQLContext 進(jìn)行了擴(kuò)展，功能要強(qiáng)大的多
1. 它可以執(zhí)行 HiveSQL 和 SQL 查詢
2. 它可以操作 hive 數(shù)據(jù)，并且可以訪問 HiveUDF
3. 它不一定需要 hive，在沒有 hive 環(huán)境時(shí)也可以使用 HiveContext
注意，如果要處理 hive 數(shù)據(jù)，需要把 hive 的 hive-site.xml 文件放到 spark/conf 下，HiveContext 將從 hive-site.xml 中獲取 hive 配置信息；
如果 HiveContext 沒有找到 hive-site.xml，他會(huì)在當(dāng)前目錄下創(chuàng)建 spark-warehouse 和 metastore_db 兩個(gè)文件夾
from pyspark import SparkContext, SparkConf
from pyspark.sql import SparkSession, SQLContext, HiveContext
conf = SparkConf().setAppName('test').setMaster('yarn')
sc = SparkContext(conf=conf)
## 需要把 hive/conf/hive-site.xml 復(fù)制到 spark/conf 下
hivec = HiveContext(sc)
print(dir(hivec))
# 'cacheTable', 'clearCache', 'createDataFrame', 'createExternalTable', 'dropTempTable', 'getConf', 'getOrCreate', 'newSession', 'range', 'read', 'readStream','refreshTable',
# 'registerDataFrameAsTable', 'registerFunction', 'registerJavaFunction', 'setConf', 'sparkSession', 'sql', 'streams', 'table', 'tableNames', 'tables', 'udf', 'uncacheTable'
data = hivec.sql('''select * from hive1101.person limit 2''')
print(type(data))
SparkSession 創(chuàng)建
from pyspark.sql import SparkSession
### method 1
sess = SparkSession.builder \
.appName("aaa") \
.config("spark.driver.extraClassPath", sparkClassPath) \
.master("local") \
.enableHiveSupport() \ # sparkSQL 連接 hive 時(shí)需要這句
.getOrCreate()# builder 方式必須有這句
### method 2
conf = SparkConf().setAppName('myapp1').setMaster('local[4]')# 設(shè)定 appname 和 master
sess = SparkSession.builder.config(conf=conf).getOrCreate() # builder 方式必須有這句
### method 3
from pyspark import SparkContext, SparkConf
conf = SparkConf().setAppName('myapp1').setMaster('local[4]')# 設(shè)定 appname 和 master
sc = SparkContext(conf=conf)
sess = SparkSession(sc)
1）文件數(shù)據(jù)源
from pyspark import SparkContext, SparkConf
from pyspark.sql import SparkSession, SQLContext, HiveContext
conf = SparkConf().setAppName('test').setMaster('yarn')
sc = SparkContext(conf=conf)
#### 替代了 SQLContext 和 HiveContext，其實(shí)只是簡(jiǎn)單的封裝，提供了統(tǒng)一的接口
spark = SparkSession(sc)
print(dir(spark))
# 很多屬性，我把私有屬性刪了
# 'Builder','builder', 'catalog', 'conf', 'createDataFrame', 'newSession', 'range', 'read', 'readStream','sparkContext', 'sql', 'stop', 'streams', 'table', 'udf', 'version'
### sess 讀取數(shù)據(jù)自動(dòng)生成 df
data = spark.read.text('/usr/yanshw/test.txt')#read 可讀類型 [ 'csv', 'format', 'jdbc', 'json', 'load', 'option', 'options', 'orc', 'parquet', 'schema', 'table', 'text']
print(type(data)) # <class 'pyspark.sql.dataframe.DataFrame'>
2） Hive 數(shù)據(jù)源
## 也需要把 hive/conf/hive-site.xml 復(fù)制到 spark/conf 下
spark = SparkSession.builder.appName('test').master('yarn').enableHiveSupport().getOrCreate()
hive_data = spark.sql('select * from hive1101.person limit 2')
print(hive_data) # DataFrame[name: string, idcard: string]
SparkSession vs SparkContext
SparkSession 是 spark2.x 引入的新概念，SparkSession 為用戶提供統(tǒng)一的切入點(diǎn)，字面理解是創(chuàng)建會(huì)話，或者連接 spark
在 spark1.x 中，SparkContext 是 spark 的主要切入點(diǎn)，由于 RDD 作為主要的 API，我們通過 SparkContext 來創(chuàng)建和操作 RDD,
SparkContext 的問題在于：
1. 不同的應(yīng)用中，需要使用不同的 context，在 Streaming 中需要使用 StreamingContext，在 sql 中需要使用 sqlContext，在 hive 中需要使用 hiveContext，比較麻煩
2. 隨著 DataSet 和 DataFrame API 逐漸成為標(biāo)準(zhǔn) API，需要為他們創(chuàng)建接入點(diǎn)，即 SparkSession
SparkSession 實(shí)際上封裝了 SparkContext，另外也封裝了 SparkConf、sqlContext，隨著版本增加，可能更多，
所以我們盡量使用 SparkSession ，如果發(fā)現(xiàn)有些 API 不在 SparkSession 中，也可以通過 SparkSession 拿到 SparkContext 和其他 Context 等
在 shell 操作中，原生創(chuàng)建了 SparkSession，故無需再創(chuàng)建，創(chuàng)建了也不會(huì)起作用
在 shell 中，SparkContext 叫 sc，SparkSession 叫 spark。
到此這篇關(guān)于SparkSession和sparkSQL的文章就介紹到這了,更多相關(guān)SparkSession和sparkSQL內(nèi)容請(qǐng)搜索本站以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持本站！
版權(quán)聲明：本站文章來源標(biāo)注為YINGSOO的內(nèi)容版權(quán)均為本站所有，歡迎引用、轉(zhuǎn)載，請(qǐng)保持原文完整并注明來源及原文鏈接。禁止復(fù)制或仿造本網(wǎng)站，禁止在非www.sddonglingsh.com所屬的服務(wù)器上建立鏡像，否則將依法追究法律責(zé)任。本站部分內(nèi)容來源于網(wǎng)友推薦、互聯(lián)網(wǎng)收集整理而來，僅供學(xué)習(xí)參考，不代表本站立場(chǎng)，如有內(nèi)容涉嫌侵權(quán)，請(qǐng)聯(lián)系alex-e#qq.com處理。
關(guān)注官方微信