spark教程(八)-SparkSession

spark 有三大引擎，spark core、sparkSQL、sparkStreaming，

spark core 的關鍵抽象是 SparkContext、RDD；

SparkSQL 的關鍵抽象是 SparkSession、DataFrame；

sparkStreaming 的關鍵抽象是 StreamingContext、DStream

 

SparkSession 是 spark2.0 引入的概念，主要用在 sparkSQL 中，當然也可以用在其他場合，他可以代替 SparkContext；

SparkSession 其實是封裝了 SQLContext 和 HiveContext

 

SQLContext

它是 sparkSQL 的入口點，sparkSQL 的應用必須創建一個 SQLContext 或者 HiveContext 的類實例

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 讀取數據也自動生成 df
data = sqlc.read.text('/usr/yanshw/test.txt')
print(type(data))

 

HiveContext

它是 sparkSQL 的另一個入口點，它繼承自 SQLContext，用於處理 hive 中的數據

 

HiveContext 對 SQLContext 進行了擴展，功能要強大的多

1. 它可以執行 HiveSQL 和 SQL 查詢

2. 它可以操作 hive 數據，並且可以訪問 HiveUDF

3. 它不一定需要 hive，在沒有 hive 環境時也可以使用 HiveContext

 

注意，如果要處理 hive 數據，需要把 hive 的 hive-site.xml 文件放到 spark/conf 下，HiveContext 將從 hive-site.xml 中獲取 hive 配置信息；

如果 HiveContext 沒有找到 hive-site.xml，他會在當前目錄下創建 spark-warehouse 和 metastore_db 兩個文件夾

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 復制到 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

它實現了對二者的封裝

SparkSession 的創建

class SparkSession(__builtin__.object):
     def __init__(self, sparkContext, jsparkSession=None):
     ''' Creates a new SparkSession.
     |
     |      >>> from datetime import datetime
     |      >>> spark = SparkSession(sc)
     |      >>> allTypes = sc.parallelize([Row(i=1, s="string", d=1.0, l=1,
     |      ...     b=True, list=[1, 2, 3], dict={"s": 0}, row=Row(a=1),
     |      ...     time=datetime(2014, 8, 1, 14, 1, 5))])
     |      >>> df = allTypes.toDF()
     |      >>> df.createOrReplaceTempView("allTypes")
     |      >>> spark.sql('select i+1, d+1, not b, list[1], dict["s"], time, row.a '
     |      ...            'from allTypes where b and i > 0').collect()
     |      [Row((i + CAST(1 AS BIGINT))=2, (d + CAST(1 AS DOUBLE))=2.0, (NOT b)=False, list[1]=2,             dict[s]=0, time=datetime.datetime(2014, 8, 1, 14, 1, 5), a=1)]
     |      >>> df.rdd.map(lambda x: (x.i, x.s, x.d, x.l, x.b, x.time, x.row.a, x.list)).collect()
     |      [(1, u'string', 1.0, 1, True, datetime.datetime(2014, 8, 1, 14, 1, 5), 1, [1, 2, 3])]'''

示例代碼

from pyspark.sql import SparkSession


### method 1
sess = SparkSession.builder \
    .appName("aaa") \
    .config("spark.driver.extraClassPath", sparkClassPath) \
    .master("local") \
    .enableHiveSupport() \  # sparkSQL 連接 hive 時需要這句
    .getOrCreate()      # builder 方式必須有這句

### method 2
conf = SparkConf().setAppName('myapp1').setMaster('local[4]')   # 設定 appname 和 master
sess = SparkSession.builder.config(conf=conf).getOrCreate() # builder 方式必須有這句

### method 3
from pyspark import SparkContext, SparkConf
conf = SparkConf().setAppName('myapp1').setMaster('local[4]')   # 設定 appname 和 master
sc = SparkContext(conf=conf)
sess = SparkSession(sc)

 

文件數據源

from pyspark import SparkContext, SparkConf
from pyspark.sql import SparkSession, SQLContext, HiveContext

conf = SparkConf().setAppName('test').setMaster('yarn')
sc = SparkContext(conf=conf)

#### 替代了 SQLContext 和 HiveContext，其實只是簡單的封裝，提供了統一的接口
spark = SparkSession(sc)
print(dir(spark))
# 很多屬性，我把私有屬性刪了
# 'Builder','builder', 'catalog', 'conf', 'createDataFrame', 'newSession', 'range', 'read', 'readStream','sparkContext', 'sql', 'stop', 'streams', 'table', 'udf', 'version'

### sess 讀取數據自動生成 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'>

 

Hive 數據源

## 也需要把 hive/conf/hive-site.xml 復制到 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 為用戶提供統一的切入點，字面理解是創建會話，或者連接 spark

在 spark1.x 中，SparkContext 是 spark 的主要切入點，由於 RDD 作為主要的 API，我們通過 SparkContext 來創建和操作 RDD,

SparkContext 的問題在於：

1. 不同的應用中，需要使用不同的 context，在 Streaming 中需要使用 StreamingContext，在 sql 中需要使用 sqlContext，在 hive 中需要使用 hiveContext，比較麻煩

2. 隨着 DataSet 和 DataFrame API 逐漸成為標准 API，需要為他們創建接入點，即 SparkSession

 

SparkSession 實際上封裝了 SparkContext，另外也封裝了 SparkConf、sqlContext，隨着版本增加，可能更多，

所以我們盡量使用 SparkSession ，如果發現有些 API 不在 SparkSession 中，也可以通過 SparkSession 拿到 SparkContext 和其他 Context 等

 

在 shell 操作中，原生創建了 SparkSession，故無需再創建，創建了也不會起作用

在 shell 中，SparkContext 叫 sc，SparkSession 叫 spark

 

通過 spark 拿到 sc

>>> dir(spark)
['Builder', '__class__', '__delattr__', '__dict__', '__doc__', '__enter__', '__exit__', '__format__', '__getattribute__', '__hash__', '__init__', '__module__', '__new__', '__reduce__', '__r
educe_ex__', '__repr__', '__setattr__', '__sizeof__', '__str__', '__subclasshook__', '__weakref__', '_convert_from_pandas', '_createFromLocal', '_createFromRDD', '_create_from_pandas_with_arrow', 
'_create_shell_session', '_get_numpy_record_dtype', '_inferSchema', '_inferSchemaFromList', '_instantiatedSession', '_jsc', '_jsparkSession', '_jvm', '_jwrapped', '_repr_html_', '_sc', '_wrapped', 
'builder', 'catalog', 'conf', 'createDataFrame', 'newSession', 'range', 'read', 'readStream', 'sparkContext', 'sql', 'stop', 'streams', 'table', 'udf', 'version']

spark.sparkContext  # 即 sc

 

dataframe 是 spark2.x 中新增的數據格式，由 SparkSession 直接讀取，不管文件是什么類型，txt也好，csv也罷，輸出格式都是 dataframe

而 SparkContext 不管讀什么文件，輸出格式都是 RDD