本文將詳細分析和總結(jié)Spark SQL及其DataFrame、Dataset的相關(guān)原理和優(yōu)化過程。

Spark SQL簡介

Spark SQL是Spark中 具有 大規(guī)模關(guān)系查詢的結(jié)構(gòu)化數(shù)據(jù)處理 模塊（spark核心組件：spark sql，spark streaming，spark mllib，spark GraphX）。spark sql支持大規(guī)模的分布式內(nèi)存計算，并且模糊了RDD與 relational table 之間的界限，DataFrame API和Datasets API是與Spark SQL進行交互的方式。

Spark SQL在Spark Core之上運行。它允許開發(fā)人員從Hive表和Parquet文件導(dǎo)入關(guān)系數(shù)據(jù)，對導(dǎo)入的數(shù)據(jù)和現(xiàn)有RDD運行SQL查詢，并輕松將RDD寫出到Hive表或Parquet文件。正如Spark SQL提供的DataFrame API一樣，它可以在外部數(shù)據(jù)源和Sparks內(nèi)置分布式集合上執(zhí)行關(guān)系操作。Spark SQL引入了稱為Catalyst的可擴展優(yōu)化器。

Spark SQL使用結(jié)構(gòu)化和半結(jié)構(gòu)化數(shù)據(jù)的3種主要功能，如：a) 它在Scala，Java和Python中均可使用DataFrame。同樣，簡化了結(jié)構(gòu)化數(shù)據(jù)集的工作，DataFrames與關(guān)系數(shù)據(jù)庫中的表相似。b) Spark SQL以各種結(jié)構(gòu)化格式可以讀取和寫入數(shù)據(jù)。例如，Hive Table，JSON和Parquet。c) 我們可以使用Spark SQL語句 查詢數(shù)據(jù)，在Spark程序內(nèi)部以及從外部工具連接到Spark SQL。

Spark SQL特點如下：a) 數(shù)據(jù)兼容：可從Hive表、外部數(shù)據(jù)庫（JDBC）、RDD、Parquet文件、JSON文件獲取數(shù)據(jù)，可通過Scala方法或SQL方式操作這些數(shù)據(jù)，并把結(jié)果轉(zhuǎn)回RDD。b) 組件擴展：SQL語法解析器、分析器、優(yōu)化器均可重新定義。c) 性能優(yōu)化：內(nèi)存列存儲、動態(tài)字節(jié)碼生成等優(yōu)化技術(shù)，內(nèi)存緩存數(shù)據(jù)。d) 多語言支持：Scala、Java、Python、R。

Spark SQL執(zhí)行計劃

Catalyst Optimizer

Catalyst Optimizer是基于scala中的函數(shù)編寫的。Catalyst Optimizer支持基于規(guī)則和基于成本優(yōu)化。

• 基于規(guī)則的優(yōu)化中，基于規(guī)則的優(yōu)化器使用規(guī)則集來確定如何執(zhí)行查詢。

• 而基于成本的優(yōu)化旨在找到執(zhí)行SQL語句的最合適方法。基于成本的優(yōu)化中，使用規(guī)則生成多個計劃，然后計算其成本。

• Unresolved logical Plan：與編譯器非常類似，spark的優(yōu)化是多階段的，在執(zhí)行任何優(yōu)化之前，需要解析表達式的引用和類型。

• Logical Plan：在Unresolved logical Plan基礎(chǔ)上加上Schema信息，spark直接對Logical Plan進行簡化和優(yōu)化，生成一個優(yōu)化后的logical Plan。這些簡化可以用匹配模式等規(guī)則來編寫。優(yōu)化器不僅限于模式匹配，而且規(guī)則還可以包含任意的Scala代碼。

• 一旦logical plan得到優(yōu)化，spark將生成一個物理計劃。物理計劃階段既有基于規(guī)則的優(yōu)化，也有基于成本的優(yōu)化，以生成最佳的物理計劃。

Spark SQL執(zhí)行計劃

Spark SQL優(yōu)化可提高開發(fā)人員的生產(chǎn)力以及他們編寫的查詢的性能。一個好的查詢優(yōu)化器會自動重寫關(guān)系查詢，以使用諸如早期過濾數(shù)據(jù)，利用可用索引，甚至確保以最有效的順序連接不同的數(shù)據(jù)源之類的技術(shù)來更有效地執(zhí)行。

通過執(zhí)行這些轉(zhuǎn)換，優(yōu)化器改善了關(guān)系查詢的執(zhí)行時間，并使開發(fā)人員從專注于其應(yīng)用程序的語義而非性能上解放出來。

Catalyst利用Scala的強大功能（例如模式匹配和運行時元編程），使開發(fā)人員可以簡明地指定復(fù)雜的關(guān)系優(yōu)化。

SparkSession

在spark2.0版本之前

• sparkContext：在2.0之前，SparkContext作為spark應(yīng)用程序的入口，spark驅(qū)動程序利用spark context連接集群。

• sparkConf ：用來指定配置參數(shù)，例如APPName、或指定spark驅(qū)動，應(yīng)用，core數(shù)量等等。

• 為了使用SQL、HIVE和Streaming的api，需要創(chuàng)建單獨的Context

val conf=newSparkConf() val sc = new SparkContext(conf) val hc = new hiveContext(sc) val ssc = new streamingContext(sc).

在spark2.0及其后續(xù)版本

• SparkSession提供了與底層Spark功能交互的單一入口點，并允許使用Dataframe和DataSet API編寫Spark程序。sparkContext提供的所有功能也都可以在sparkSession中使用。

• 為了使用SQL、HIVE和Streaming的api，不需要創(chuàng)建單獨的Context，因為sparkSession包含所有api。

• 一旦SparkSession被實例化，我們就可以配置Spark的運行時配置屬性

val session = SparkSession.builder() .enableHiveSupport() //提供了與hive metastore的連接。 .getOrCreate() // Import the implicits, unlike in core Spark the implicits are defined // on the context. import session.implicits._

Schemas

和RDD一樣，Spark SQL下的DataFrame和Dataset也是一個分布式的集合。但是相對于RDD，DataFrame和Dataset多了一個額外的Schema信息。如上面所述，Schemas可在Catalyst優(yōu)化器中使用。

通過case class指定schema

case class RawPanda(id: Long, zip: String, pt: String, happy: Boolean, attributes: Array[Double])case class PandaPlace(name: String, pandas: Array[RawPanda]) def createAndPrintSchema() = {val damao = RawPanda(1, "M1B 5K7", "giant", true, Array(0.1, 0.1))val pandaPlace = PandaPlace("toronto", Array(damao))val df = session.createDataFrame(Seq(pandaPlace))df.printSchema() }

通過StructType直接指定Schema

import org.apache.spark.{SparkContext, SparkConf}val personRDD = sc.textFile(args(0)).map(_.split(" ")) //通過StructType直接指定每個字段的schema val schema = StructType(List(StructField("id", IntegerType, true),StructField("name", StringType, true),StructField("age", IntegerType, true))) //將RDD映射到rowRDD val rowRDD = personRDD.map(p => Row(p(0).toInt, p(1).trim, p(2).toInt)) //將schema信息應(yīng)用到rowRDD上 val personDataFrame = sqlContext.createDataFrame(rowRDD, schema)

DataFrame

DataFrame與RDD類似，同樣擁有 不變性，彈性，分布式計算的特性，也有惰性設(shè)計，有transform（轉(zhuǎn)換）與action（執(zhí)行）操作之分。相對于RDD，它能處理大量結(jié)構(gòu)化數(shù)據(jù)，DataFrame包含帶有Schema的行，類似于pandas的DataFrame的 header行。

注意：相對于RDD的lazy設(shè)計，DataFrame只是部分的lazy，例如schema是立即執(zhí)行的。

為什么使用DataFrame

相對于RDD，DataFrame提供了內(nèi)存管理和優(yōu)化的執(zhí)行計劃。

• 自定義內(nèi)存管理：這也被稱為Tungsten，Spark是由scala開發(fā)的，JVM的實現(xiàn)帶來了一些性能上的限制和弊端（例如GC上的overhead，java序列化耗時），使得Spark在性能上無法和一些更加底層的語言（例如c，可以對memory進行高效管理，從而利用hardware的特性）相媲美，Tungsten設(shè)計了一套內(nèi)存管理機制，而不再是交給JVM托管，Spark的operation直接使用分配的binary data而不是JVM objects。

• 優(yōu)化執(zhí)行計劃：這也被稱為query optimizer（例如Catalyst Optimizer）。使用此選項，將為查詢的執(zhí)行創(chuàng)建一個優(yōu)化的執(zhí)行計劃。一旦優(yōu)化的計劃被創(chuàng)建，最終將在Spark的rdd上執(zhí)行。

關(guān)于jvm內(nèi)存，可查看 JVM中的堆外內(nèi)存（off-heap memory）與堆內(nèi)內(nèi)存（on-heap memory）^[1]更多關(guān)于 Tungsten，可查看 Tungsten-github^[2]

DataFrame 經(jīng)驗 tips：

• $ 可以用來隱式的指定DataFrame中的列。

• Spark DataFrame 中 === 用 =!= 來過濾特定列的行數(shù)據(jù)。

• 相對filter操作，distinct和dropDuplicates可能會引起shuffle過程，因此可能會比較慢。

• 與RDD中g(shù)roupby效率低下不同，DataFrame中Groupby已經(jīng)經(jīng)過本地聚合再全局聚合（DataFrame / Dataset groupBy behaviour/optimization^[3]）

• 如果你需要計算各種復(fù)雜的統(tǒng)計運算，建議在GroupData（groupby后）執(zhí)行：

def minMeanSizePerZip(pandas: DataFrame): DataFrame = {// Compute the min and meanpandas.groupBy(pandas("zip")).agg(min(pandas("pandaSize")), mean(pandas("pandaSize")))}

• 在hive Data上，有時使用sql表達式操作比直接在DataFrame上操作，效率更高

def registerTable(df: DataFrame): Unit = {df.registerTempTable("pandas") //將pandas注冊為一個臨時tabledf.write.saveAsTable("perm_pandas")}def querySQL(): DataFrame = {sqlContext.sql("SELECT * FROM pandas WHERE size > 0") //即可利用sql表達式對臨時表進行查詢操作，返回的也是一個DataFrame}

Dataset

Dataset是SparkSQL中的一種數(shù)據(jù)結(jié)構(gòu)，它是強類型的，包含指定的schema（指定了變量的類型）。Dataset是對DataFrame API的擴展。Spark Dataset 提供了類型安全和面向?qū)ο蟮木幊探涌凇ｊP(guān)于強類型和類型安全的定義可參考 Magic lies here - Statically vs Dynamically Typed Languages^[4]

Dataset有以下特點：

• 在編寫代碼時，有如RDD般的方便和靈活。

• 有如DataFrame優(yōu)化表現(xiàn)（同樣使用Tungsten 和 Catalyst Query Optimizer）

• 具有scala語言的靜態(tài)、類型安全的特點

• 使用spark Dataset，可以在編譯時檢查語法和分析，而Dataframe、rdd或常規(guī)SQL查詢不能做到。

RDD，DataFrame，Dataset區(qū)別

數(shù)據(jù)格式上差別

• RDD：它可以方便有效地處理結(jié)構(gòu)化和非結(jié)構(gòu)化數(shù)據(jù)。但和DataFrame和DataSets不一樣，RDD并不能推斷schema信息，而是要求用戶指定它。

• DataFrame：它只適用于結(jié)構(gòu)化和半結(jié)構(gòu)化數(shù)據(jù)，可以推斷schema信息。DataFrames允許Spark管理schema。

• Dataset：它還可以有效地處理結(jié)構(gòu)化和非結(jié)構(gòu)化數(shù)據(jù)，它以行的JVM對象或行對象的集合的形式表示數(shù)據(jù)，一行就是一個通用的無類型的 JVM 對象。

三者互轉(zhuǎn)

在這里插入圖片描述

• RDD轉(zhuǎn)DataFrame（行動操作，立即執(zhí)行）時，需要指定schema信息，有如下三種方法：

  def createFromCaseClassRDD(input: RDD[PandaPlace]) = {// Create DataFrame explicitly using session and schema inferenceval df1 = session.createDataFrame(input)// Create DataFrame using session implicits and schema inferenceval df2 = input.toDF()// Create a Row RDD from our RDD of case classesval rowRDD = input.map(pm => Row(pm.name,pm.pandas.map(pi => Row(pi.id, pi.zip, pi.happy, pi.attributes))))val pandasType = ArrayType(StructType(List(StructField("id", LongType, true),StructField("zip", StringType, true),StructField("happy", BooleanType, true),StructField("attributes", ArrayType(FloatType), true))))// Create DataFrame explicitly with specified schemaval schema = StructType(List(StructField("name", StringType, true),StructField("pandas", pandasType)))val df3 = session.createDataFrame(rowRDD, schema)}

• DataFrame轉(zhuǎn)RDD（轉(zhuǎn)換操作，行動操作再執(zhí)行），簡單的df.rdd得到的是個Row Object，因為每行可以包含任意內(nèi)容，你需要指定特別的類型，這樣你才能獲取每列的內(nèi)容：

  def toRDD(input: DataFrame): RDD[RawPanda] = {val rdd: RDD[Row] = input.rddrdd.map(row => RawPanda(row.getAs[Long](0 "Long"), row.getAs[String](1 "String"),row.getAs[String](2 "String"), row.getAs[Boolean](3 "Boolean"), row.getAs[Array[Double]](4 "Array[Double]")))}

• 轉(zhuǎn)Dataset

  def fromDF(df: DataFrame): Dataset[RawPanda] = {df.as[RawPanda]//RawPanda為一個case class } // rdd轉(zhuǎn) Dataset，可以先轉(zhuǎn) DataFrame再轉(zhuǎn)Dataset /** * Illustrate converting a Dataset to an RDD */ def toRDD(ds: Dataset[RawPanda]): RDD[RawPanda] = {ds.rdd } /** * Illustrate converting a Dataset to a DataFrame */ def toDF(ds: Dataset[RawPanda]): DataFrame = {ds.toDF() }

靜態(tài)類型與運行時類型安全

如果你用的是 Spark SQL 的查詢語句，要直到運行時你才會發(fā)現(xiàn)有語法錯誤（這樣做代價很大），而如果你用的是 DataFrame 和 Dataset，你在編譯時就可以捕獲syntax errors（這樣就節(jié)省了開發(fā)者的時間和整體代價）。也就是說，當(dāng)你在 DataFrame 中調(diào)用了 API 之外的函數(shù)時，編譯器就可以發(fā)現(xiàn)這個錯。不過，如果你使用了一個不存在的字段名字，那就要到運行時才能發(fā)現(xiàn)錯誤了。

Dataset API 都是用 lambda 函數(shù)和 JVM 類型對象表示的，所有不匹配的類型參數(shù)都可以在編譯時發(fā)現(xiàn)。而且在使用 Dataset 時，你的Analysis errors 也會在編譯時被發(fā)現(xiàn)，這樣就節(jié)省了開發(fā)者的時間和代價。例如DataFrame編譯時不檢查列信息（例如無論你寫df.select("name") 還是 df.select("naame") 編譯時均不會報錯，而實際運行時才會報錯），而Dataset在編譯時就會檢查到該類錯誤。

分區(qū)方式

如何針對數(shù)據(jù)分布自定義分區(qū)方式，這對于避免令人頭痛的數(shù)據(jù)傾斜非常重要。

分階段分區(qū)聚合

• 使用map-side預(yù)聚合的shuffle操作。所謂的map-side預(yù)聚合，說的是在每個節(jié)點本地對相同的key進行一次聚合操作，類似于MapReduce中的本地combiner。map-side預(yù)聚合之后，每個節(jié)點本地就只會有一條相同的key，因為多條相同的key都被聚合起來了。其他節(jié)點在拉取所有節(jié)點上的相同key時，就會大大減少需要拉取的數(shù)據(jù)數(shù)量，從而也就減少了磁盤IO以及網(wǎng)絡(luò)傳輸開銷。通常來說，在可能的情況下，建議使用reduceByKey或者aggregateByKey算子來替代掉groupByKey算子。因為reduceByKey和aggregateByKey算子都會使用用戶自定義的函數(shù)對每個節(jié)點本地的相同key進行預(yù)聚合。而groupByKey算子是不會進行預(yù)聚合的，全量的數(shù)據(jù)會在集群的各個節(jié)點之間分發(fā)和傳輸，性能相對來說比較差。與RDD中g(shù)roupby效率低下不同，DataFrame中Groupby已經(jīng)經(jīng)過本地聚合再全局聚合

• 加隨機數(shù)分區(qū)分階段聚合。這個方案的核心實現(xiàn)思路就是進行兩階段聚合。第一次是局部聚合，先給每個key都打上一個隨機數(shù)，比如10以內(nèi)的隨機數(shù)，此時原先一樣的key就變成不一樣的了，比如(hello, 1) (hello, 1) (hello, 1) (hello, 1)，就會變成(1_hello, 1) (1_hello, 1) (2_hello, 1) (2_hello, 1)。接著對打上隨機數(shù)后的數(shù)據(jù)，執(zhí)行reduceByKey等聚合操作，進行局部聚合，那么局部聚合結(jié)果，就會變成了(1_hello, 2) (2_hello, 2)。然后將各個key的前綴給去掉，就會變成(hello,2)(hello,2)，再次進行全局聚合操作，就可以得到最終結(jié)果了，比如(hello, 4)。--- 將原本相同的key通過附加隨機前綴的方式，變成多個不同的key，就可以讓原本被一個task處理的數(shù)據(jù)分散到多個task上去做局部聚合，進而解決單個task處理數(shù)據(jù)量過多的問題。接著去除掉隨機前綴，再次進行全局聚合，就可以得到最終的結(jié)果。加隨機數(shù)參考代碼如下：

val repartRdd = originRdd// 切割.flatMap(_.split(" "))// 映射為元組.map((_, 1))// 給key加上隨機數(shù)，聚合時候具有隨機性.map(t => {val rnum = Random.nextInt(partitionNum)(t._1 + "_" + rnum, 1)})// 初次聚合.reduceByKey(_ + _)// 去除key隨機后綴.map(e => {val word = e._1.toString().substring(0, e._1.toString().indexOf("_"))val count = e._2(word, count)})// 再次聚合.reduceByKey(_ + _)// 排序(false->降序， true->升序).map(e => (e._2, e._1)).sortByKey(false).map(e => (e._2, e._1))

Hash Partitioning in Spark 與 Range Partitioning in Spark

可參考 Spark中的分區(qū)方法詳解（https://www.cnblogs.com/tongxupeng/p/10435976.html），個人感覺這篇博客已經(jīng)寫得非常詳細完整。

• DataFrame.repartition(col)：可由 指定的列 表達式來進行分區(qū)，默認hash分區(qū) (隨機key) 方式

• DataFrame.repartitionByRange(col): 可由 指定的列 表達式來進行分區(qū)，默認Range分區(qū) (隨機key) 方式

Spark.DataFrame 與 DataSet 無自定義分區(qū)方式，可先將rdd自定分區(qū)完成，再轉(zhuǎn)成DataFrame。

sqlContext.createDataFrame(df.rdd.map(r => (r.getInt(1), r)).partitionBy(partitioner).values,df.schema )

UDFs & UDAFs 使用

User-defined functions(udfs) 和 user-defined aggregate functions(udafs) 提供了使用自己的自定義代碼擴展DataFrame和SQL API的方法，同時保留了Catalyst優(yōu)化器。這對性能的提高非常有用，否則您需要將數(shù)據(jù)轉(zhuǎn)換為RDD（并可能再次轉(zhuǎn)換）來執(zhí)行任意函數(shù)，這非常昂貴。udf和udaf也可以使用SQL查詢表達式進行 內(nèi)部訪問。

注：使用python編寫udf和udaf函數(shù)，會丟失性能優(yōu)勢。

UDFs

spark 2.x:

def get_max(x: Double, y: Double): Double={if ( x > y )xelsey}val udf_get_max = udf(get_max _) df = df.withColumn("max_fea", udf_get_max(df("fea1"), df("fea2")))

UDAFs

相對于udfs，udafs編寫較為復(fù)雜，需要繼承 UserDefinedAggregateFunction 并實現(xiàn)里面的函數(shù)，但UDAFs的性能相當(dāng)好。可以直接在列上使用UDAF，也可以像對非聚合UDF那樣將其添加到函數(shù)注冊表中。

計算平均值的UDAF例子代碼：

import org.apache.spark.sql.Row import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction} import org.apache.spark.sql.types._object AverageUserDefinedAggregateFunction extends UserDefinedAggregateFunction {// 聚合函數(shù)的輸入數(shù)據(jù)結(jié)構(gòu)override def inputSchema: StructType = StructType(StructField("input", LongType) :: Nil)// 緩存區(qū)數(shù)據(jù)結(jié)構(gòu)override def bufferSchema: StructType = StructType(StructField("sum", LongType) :: StructField("count", LongType) :: Nil)// 聚合函數(shù)返回值數(shù)據(jù)結(jié)構(gòu)override def dataType: DataType = DoubleType// 聚合函數(shù)是否是冪等的，即相同輸入是否總是能得到相同輸出override def deterministic: Boolean = true// 初始化緩沖區(qū)override def initialize(buffer: MutableAggregationBuffer): Unit = {buffer(0) = 0Lbuffer(1) = 0L}// 給聚合函數(shù)傳入一條新數(shù)據(jù)進行處理override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {if (input.isNullAt(0)) returnbuffer(0) = buffer.getLong(0) + input.getLong(0)buffer(1) = buffer.getLong(1) + 1}// 合并聚合函數(shù)緩沖區(qū)override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {buffer1(0) = buffer1.getLong(0) + buffer2.getLong(0)buffer1(1) = buffer1.getLong(1) + buffer2.getLong(1)}// 計算最終結(jié)果override def evaluate(buffer: Row): Any = buffer.getLong(0).toDouble / buffer.getLong(1)}

然后在主函數(shù)里注冊并使用該函數(shù)：

spark.read.json("data/user").createOrReplaceTempView("v_user") spark.udf.register("u_avg", AverageUserDefinedAggregateFunction)// 將整張表看做是一個分組對求所有人的平均年齡 spark.sql("select count(1) as count, u_avg(age) as avg_age from v_user").show()// 按照性別分組求平均年齡 spark.sql("select sex, count(1) as count, u_avg(age) as avg_age from v_user group by sex").show()

參考

• High Performance Spark

• https://data-flair.training/blogs/spark-sql-optimization/

• https://www.quora.com/What-is-the-difference-between-spark-context-and-spark-session

• https://www.cnblogs.com/netoxi/p/7223413.html

• https://www.infoq.cn/article/three-apache-spark-apis-rdds-dataframes-and-datasets/

• https://tech.meituan.com/2016/05/12/spark-tuning-pro.html

• https://www.cnblogs.com/tongxupeng/p/10435976.html

參考資料

[1]

JVM中的堆外內(nèi)存（off-heap memory）與堆內(nèi)內(nèi)存（on-heap memory）: https://blog.csdn.net/khxu666/article/details/80775635

[2]

Tungsten-github: https://github.com/hustnn/TungstenSecret

[3]

DataFrame / Dataset groupBy behaviour/optimization: https://stackoverflow.com/questions/32902982/dataframe-dataset-groupby-behaviour-optimization

[4]

Magic lies here - Statically vs Dynamically Typed Languages: https://android.jlelse.eu/magic-lies-here-statically-typed-vs-dynamically-typed-languages-d151c7f95e2b

往期精彩回顧適合初學(xué)者入門人工智能的路線及資料下載機器學(xué)習(xí)在線手冊深度學(xué)習(xí)在線手冊AI基礎(chǔ)下載（pdf更新到25集）備注：加入本站微信群或者qq群，請回復(fù)“加群”獲取一折本站知識星球優(yōu)惠券，請回復(fù)“知識星球”

喜歡文章，點個在看

總結(jié)

以上是生活随笔為你收集整理的Spark性能优化 -- Spark SQL、DataFrame、Dataset的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。