日韩中文字幕免费,999精品视频在这里,中文字幕免费一区二区

主頁 > 知識庫 > 創建SparkSession和sparkSQL的詳細過程

創建SparkSession和sparkSQL的詳細過程

一、概述

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

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

SparkSQL 的關鍵抽象是 SparkSession、DataFrame；

sparkStreaming 的關鍵抽象是 StreamingContext、DStream

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

SparkSession 其實是封裝了 SQLContext 和 HiveContext

(1) 在Spark1.6 中我們使用的叫Hive on spark，主要是依賴hive生成spark程序，有兩個核心組件SQLcontext和HiveContext。

這是Spark 1.x 版本的語法

//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，是后繼的全新產品，解除了對Hive的依賴。

從Spark2.0以上的版本開始，spark是使用全新的SparkSession接口代替Spark1.6 中的SQLcontext和HiveContext 來實現對數據的加載、轉換、處理等工作，并且實現了SQLcontext和HiveContext的所有功能。

在新版本中并不需要之前那么繁瑣的創建很多對象，只需要創建一個SparkSession對象即可。SparkSession支持從不同的數據源加載數據，并把數據轉換成DataFrame，并支持把DataFrame轉換成SQLContext自身中的表。然后使用SQL語句來操作數據，也提供了HiveQL以及其他依賴于Hive的功能支持。

二、創建SparkSession

SparkSession 是 Spark SQL 的入口。使用 Dataset 或者 Dataframe 編寫 Spark SQL 應用的時候，第一個要創建的對象就是 SparkSession。Builder 是 SparkSession 的構造器。通過 Builder, 可以添加各種配置，并通過 stop 函數來停止 SparkSession。

Builder 的方法如下：

import org.apache.spark.sql.SparkSession
val spark: SparkSession = SparkSession.builder
 .appName("My Spark Application")  //設置 application 的名字
 .master("local[*]")             
 .enableHiveSupport() //增加支持 hive Support      
 .config("spark.sql.warehouse.dir", "target/spark-warehouse") //設置各種配置
 .getOrCreate     //獲取或者新建一個 sparkSession

（1）設置參數

創建SparkSession之后可以通過 spark.conf.set 來設置運行參數

//配置設置
 spark.conf.set("spark.sql.shuffle.partitions", 6)
 spark.conf.set("spark.executor.memory", "2g")
 
 //獲取配置，可以使用Scala的迭代器來讀取configMap中的數據。 
 val configMap:Map[String, String] = spark.conf.getAll()

（2）讀取元數據

如果需要讀取元數據(catalog)，可以通過SparkSession來獲取。

spark.catalog.listTables.show(false)
 spark.catalog.listDatabases.show(false)

這里返回的都是Dataset，所以可以根據需要再使用Dataset API來讀取

catalog 和 schema 是兩個不同的概念
Catalog是目錄的意思，從數據庫方向說，相當于就是所有數據庫的集合;
Schema是模式的意思, 從數據庫方向說, 類似Catelog下的某一個數據庫;

（3）創建Dataset和Dataframe

通過SparkSession來創建Dataset和Dataframe有多種方法。

通過range()方法來創建dataset
通過createDataFrame()來創建dataframe。

// create a Dataset using spark.range starting from 5 to 100,
// with increments of 5
val numDS = spark.range(5, 100, 5)//創建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)) )//創建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）讀取數據

可以用SparkSession讀取JSON、CSV、TXT 和 parquet表。

import spark.implicits //使RDD轉化為DataFrame以及后續SQL操作
//讀取JSON文件，生成DataFrame
val jsonFile = args(0)
val zipsDF = spark.read.json(jsonFile)

（5）使用SparkSQL

借助SparkSession用戶可以像SQLContext一樣使用Spark SQL的全部功能。

zipsDF.createOrReplaceTempView("zips_table")//對上面的dataframe創建一個表
zipsDF.cache()//緩存表
 
val resultsDF = spark.sql("SELECT city, pop, state, zip FROM zips_table")
 
//對表調用SQL語句
resultsDF.show(10)//展示結果

（6）存儲/讀取Hive表

下面的代碼演示了通過SparkSession來創建Hive表并進行查詢的方法。

//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 的入口點，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 創建

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)

1）文件數據源

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'>

2） 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。

到此這篇關于SparkSession和sparkSQL的文章就介紹到這了,更多相關SparkSession和sparkSQL內容請搜索腳本之家以前的文章或繼續瀏覽下面的相關文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: