Spark SQL 是 Spark 用來處理結(jié)構(gòu)化數(shù)據(jù)（結(jié)構(gòu)化數(shù)據(jù)可以來自外部結(jié)構(gòu)化數(shù)據(jù)源也可以通 過 RDD 獲?。┑囊粋€(gè)模塊，它提供了一個(gè)編程抽象叫做 DataFrame 并且作為分布式 SQL 查 詢引擎的作用。
外部的結(jié)構(gòu)化數(shù)據(jù)源包括 JSON、Parquet(默認(rèn))、RMDBS、Hive等。當(dāng)前 Spark SQL 使用 Catalyst 優(yōu)化器來對(duì) SQL 進(jìn)行優(yōu)化，從而得到更加高效的執(zhí)行方案。并且可以將結(jié)果存儲(chǔ)到外部系統(tǒng)。
潯陽(yáng)網(wǎng)站建設(shè)公司創(chuàng)新互聯(lián)建站,潯陽(yáng)網(wǎng)站設(shè)計(jì)制作，有大型網(wǎng)站制作公司豐富經(jīng)驗(yàn)。已為潯陽(yáng)上1000家提供企業(yè)網(wǎng)站建設(shè)服務(wù)。企業(yè)網(wǎng)站搭建\外貿(mào)網(wǎng)站制作要多少錢，請(qǐng)找那個(gè)售后服務(wù)好的潯陽(yáng)做網(wǎng)站的公司定做！
- 容易整合
- 統(tǒng)一的數(shù)據(jù)訪問方式
- 兼容hive
- 標(biāo)準(zhǔn)的數(shù)據(jù)連接
- spark sql 的前身是shark。但是spark sql拋棄了原有shark的代碼，汲取了shark的一些優(yōu)點(diǎn)，如：列存儲(chǔ)（In-Memory Columnar Storage）、Hive 兼容性等，重新開發(fā) SparkSQL。
- spark -1.1 2014 年 9 月 11 日，發(fā)布 Spark1.1.0。Spark 從 1.0 開始引入 SparkSQL（Shark 不再支持升級(jí)與維護(hù)）。Spark1.1.0 變化較大是 SparkSQL 和 MLlib
- spark -1.3 增加了dataframe新
- spark -1.4 增加了窗口分析函數(shù)
- spark - 1.5 鎢絲計(jì)劃。Hive 中有 UDF 與 UDAF，Spark 中對(duì) UDF 支持較早
- spark 1.6 執(zhí)行的 sql 中可以增加"--"注釋，Spark-1.5/1.6 的新特性，引入 DataSet 的概念
- spark 2.x SparkSQL+DataFrame+DataSet(正式版本)，Structured Streaming(DataSet)，引入 SparkSession 統(tǒng)一了 RDD，DataFrame，DataSet 的編程入口
SparkSession 是 Spark-2.0 引如的新概念。SparkSession 為用戶提供了統(tǒng)一的切入點(diǎn)，來讓用戶學(xué)習(xí) Spark 的各項(xiàng)功能。
隨著 DataSet 和 DataFrame 的 API 逐漸成為標(biāo)準(zhǔn)的 API，SparkSession 作為 DataSet 和 DataFrame API 的切入點(diǎn)，SparkSession 封裝了 SparkConf、SparkContext 和 SQLContext。為了向后兼容，SQLContext 和 HiveContext 也被保存下來。
特點(diǎn)：
- 為用戶提供一個(gè)統(tǒng)一的切入點(diǎn)使用 Spark 各項(xiàng)功能
- 允許用戶通過它調(diào)用 DataFrame 和 Dataset 相關(guān) API 來編寫程序
- 減少了用戶需要了解的一些概念，可以很容易的與 Spark 進(jìn)行交互
- 與 Spark 交互之時(shí)不需要顯示的創(chuàng)建 SparkConf、SparkContext 以及 SQlContext，這些對(duì) 象已經(jīng)封閉在 SparkSession 中
- SparkSession 提供對(duì) Hive 特征的內(nèi)部支持：用 HiveQL 寫 SQL 語(yǔ)句，訪問 Hive UDFs，從 Hive 表中讀取數(shù)據(jù)
SparkSession的創(chuàng)建：
在spark-shell中SparkSession 會(huì)被自動(dòng)初始化一個(gè)對(duì)象叫做 spark，為了向后兼容，Spark-Shell 還提供了一個(gè) SparkContext 的初始化對(duì)象，方便用戶操作：
在代碼開發(fā)的時(shí)候創(chuàng)建：
重慶分公司
重慶分公司
val conf = new SparkConf()
val spark: SparkSession = SparkSession.builder()
.appName("_01spark_sql")
.config(conf)
.getOrCreate()這里主要說的是RDD的局限性：
- RDD是不支持spark-sql的
- RDD 僅表示數(shù)據(jù)集，RDD 沒有元數(shù)據(jù)，也就是說沒有字段語(yǔ)義定義
- RDD 需要用戶自己優(yōu)化程序，對(duì)程序員要求較高
- 從不同數(shù)據(jù)源讀取數(shù)據(jù)相對(duì)困難，讀取到不同格式的數(shù)據(jù)都必須用戶自己定義轉(zhuǎn)換方式 合并多個(gè)數(shù)據(jù)源中的數(shù)據(jù)也較困難
DataFrame 被稱為 SchemaRDD。以行為單位構(gòu)成的分布式數(shù)據(jù)集合，按照列賦予不同的名稱。對(duì) select、fileter、aggregation 和 sort 等操作符的抽象。其中 Schema 是就是元數(shù)據(jù)，是語(yǔ)義描述信息。DataFrame是分布式的Row對(duì)象的集合.
DataFrame = RDD+Schema = SchemaRDD
優(yōu)勢(shì)：
- DataFrame 是一種特殊類型的 Dataset，DataSet[Row] = DataFrame
- DataFrame 自帶優(yōu)化器 Catalyst，可以自動(dòng)優(yōu)化程序
- DataFrame 提供了一整套的 Data Source API
特點(diǎn)：
- 支持 單機(jī) KB 級(jí)到集群 PB 級(jí)的數(shù)據(jù)處理
- 支持多種數(shù)據(jù)格式和存儲(chǔ)系統(tǒng)
- 通過 Spark SQL Catalyst 優(yōu)化器可以進(jìn)行高效的代碼生成和優(yōu)化
- 能夠無(wú)縫集成所有的大數(shù)據(jù)處理工具
- 提供 Python, Java, Scala, R 語(yǔ)言 API
由于 DataFrame 的數(shù)據(jù)類型統(tǒng)一是 Row，所以 DataFrame 也是有缺點(diǎn)的。Row 運(yùn)行時(shí)類型檢查，比如 salary 是字符串類型，下面語(yǔ)句也只有運(yùn)行時(shí)才進(jìn)行類型檢查。 dataframe.filter("salary>1000").show()
Dataset擴(kuò)展了 DataFrame API，提供了編譯時(shí)類型檢查，面向?qū)ο箫L(fēng)格的 API。
Dataset 可以和 DataFrame、RDD 相互轉(zhuǎn)換。DataFrame=Dataset[Row],可見 DataFrame 是一種特殊的 Dataset。
這里小編要重點(diǎn)強(qiáng)調(diào)一下二者的區(qū)別，但是在學(xué)習(xí)spark-sql的時(shí)候就對(duì)二者的關(guān)系不太清楚，而且在面試的時(shí)候也問到了這個(gè)問題，真的是一番血淚史啊。
通過查看多個(gè)前輩對(duì)二者的總結(jié)我大概的總結(jié)一下二者的區(qū)別：
- Dataset可以認(rèn)為是DataFrame的一個(gè)特例，主要區(qū)別是Dataset每一個(gè)record存儲(chǔ)的是一個(gè)強(qiáng)類型值而不是一個(gè)Row
- DataSet可以在編譯時(shí)檢查類型，而DataFrame只有在正真運(yùn)行的時(shí)候才會(huì)檢查
- DataFrame每一行的類型都是Row，不解析我們就無(wú)法知曉其中有哪些字段，每個(gè)字段又是什么類型。我們只能通過getAs[類型]或者row(i)的方式來獲取特定的字段內(nèi)容（超級(jí)大弊端）；而dataSet每一行的類型是不一定的，在自定義了case class之后就可以很自由的獲取每一行的信息。
好了 廢話說了一堆，不如直接上代碼：
object SparkSqlTest {
def main(args: Array[String]): Unit = {
//屏蔽多余的日志
Logger.getLogger("org.apache.hadoop").setLevel(Level.WARN)
Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
Logger.getLogger("org.project-spark").setLevel(Level.WARN)
val conf: SparkConf = new SparkConf()
conf.setMaster("local[2]")
.setAppName("SparkSqlTest")
//設(shè)置spark的序列化器
.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
//將自定義的對(duì)象，加入序列化器中
.registerKryoClasses(Array(classOf[Person]))
//構(gòu)建SparkSession對(duì)象
val spark: SparkSession = SparkSession.builder()
.config(conf).getOrCreate()
//創(chuàng)建sparkContext對(duì)象
val sc: SparkContext = spark.sparkContext
val list = List(
new Person("委xx", 18),
new Person("吳xx", 20),
new Person("戚xx", 30),
new Person("王xx", 40),
new Person("薛xx", 18)
)
//創(chuàng)建DataFrame
//構(gòu)建元數(shù)據(jù)
val schema = StructType(List(
StructField("name", DataTypes.StringType),
StructField("age", DataTypes.IntegerType)
))
//構(gòu)建RDD
val listRDD: RDD[Person] = sc.makeRDD(list)
val RowRDD: RDD[Row] = listRDD.map(field => {
Row(field.name, field.age)
})
val perDF: DataFrame = spark.createDataFrame(RowRDD,schema)
//創(chuàng)建DataSet
import spark.implicits._ //這句話一定要加
val perDS: Dataset[Person] = perDF.as[Person]
/**
* 這里主要介紹DF 和 DS的區(qū)別
*/
perDF.foreach(field=>{
val name=field.get(0) //根據(jù)元素的index，取出相應(yīng)的元素的值
val age=field.getInt(1) //根據(jù)元素的index和元素的類型取出元素的值
field.getAs[Int]("age") //根據(jù)元素的類型和元素的名稱取出元素的值
println(s"${age},${name}")
})
perDS.foreach(field=>{
//直接根據(jù)上面定義的元素的名稱取值
val age=field.age
val name=field.name
println(s"${age},${name}")
})
}
}
case class Person(name: String, age: Int)個(gè)人感覺，就是DataFrame雖然集成和很多優(yōu)點(diǎn)，但是，如果想從DataFrame中取出具體的某個(gè)對(duì)象的某個(gè)屬性，是不能確定的，步驟比較繁瑣，而且類型不確定。但是使用DataSet則有效額的避免了所有的問題。
