Spark SQL玩起来
標簽（空格分隔）： Spark
[toc]
前言
Spark SQL的介紹只包含官方文檔的Getting Started、DataSource、Performance Tuning和Distributed SQL Engine部分。不含其他的遷移和PySpark等部分。
Spark SQL介紹
Spark SQL是一個Spark模塊用于結(jié)構(gòu)化數(shù)據(jù)處理。與基本的Spark RDD API不同，Spark SQL提供的接口為Spark提供了有關(guān)數(shù)據(jù)結(jié)構(gòu)和正在執(zhí)行的計算的更多信息。 在內(nèi)部，Spark SQL使用此額外信息來執(zhí)行額外的優(yōu)化。 有幾種與Spark SQL交互的方法，包括SQL和Dataset API。 在使用相同的執(zhí)行引擎計算結(jié)果時，與使用表達計算的API或者語言無關(guān)。 這種統(tǒng)一意味著開發(fā)人員可以輕松地在不同的API之間來回切換，從而提供表達給定轉(zhuǎn)換的最自然的方式。
SQL
Spark SQL的一個用途是執(zhí)行SQL查詢。Spark SQL還可用于從現(xiàn)有Hive中讀取數(shù)據(jù)。 有關(guān)如何配置此功能的更多信息，請參閱Hive Tables部分。 從其他編程語言中運行SQL時，結(jié)果將作為Dataset/DataFrame返回。 還可以使用命令行或JDBC/ODBC與SQL接口進行交互。
Dataset和DataFrame
Dataset數(shù)據(jù)集是分布式數(shù)據(jù)集合。數(shù)據(jù)集是Spark 1.6中添加的一個新接口，它提供了RDD的優(yōu)勢（強類型，使用強大的lambda函數(shù)的能力）和Spark SQL優(yōu)化執(zhí)行引擎的優(yōu)點。 數(shù)據(jù)集可以從JVM對象構(gòu)造，然后使用功能轉(zhuǎn)換（map，flatMap，filter等）進行操作。 數(shù)據(jù)集API在Scala和Java中可用。 Python沒有對Dataset API的支持。 但由于Python的動態(tài)特性，數(shù)據(jù)集API的許多好處已經(jīng)可用（即可以通過名稱自然地訪問行的字段row.columnName）。 R的情況類似。
DataFrame是一個組織成命名列的數(shù)據(jù)集。 它在概念上等同于關(guān)系數(shù)據(jù)庫中的表或R / Python中的數(shù)據(jù)框，但在底層具有更豐富的優(yōu)化。 DataFrame可以從多種來源構(gòu)建，例如：結(jié)構(gòu)化數(shù)據(jù)文件，Hive中的表，外部數(shù)據(jù)庫或現(xiàn)有RDD。 DataFrame API在Scala，Java，Python和R中可用。在Scala和Java中，DataFrame由行數(shù)據(jù)集表示。 在Scala API中，DataFrame只是Dataset[Row]的類型別名。 而在Java API中，用戶需要使用Dataset<Row>來表示DataFrame。
Spark SQL入門知識
SparkSession
Spark中所有功能的入口點是SparkSession類。通過類似下面的代碼來創(chuàng)建：
import org.apache.spark.sql.SparkSessionval spark = SparkSession.builder().appName("Spark SQL basic example").config("spark.some.config.option", "some-value").getOrCreate()// For implicit conversions like converting RDDs to DataFrames import spark.implicits._Spark 2.0中的SparkSession為Hive功能提供內(nèi)置支持，包括使用HiveQL編寫查詢，訪問Hive UDF以及從Hive表讀取數(shù)據(jù)的功能。 要使用這些功能，并不需擁有現(xiàn)有的Hive設(shè)置。
創(chuàng)建DataFrame
使用SparkSession，應(yīng)用程序可以從現(xiàn)有RDD，Hive表或Spark數(shù)據(jù)源創(chuàng)建DataFrame。作為示例，以下內(nèi)容基于JSON文件的內(nèi)容創(chuàng)建DataFrame：
val df = spark.read.json("examples/src/main/resources/people.json")// Displays the content of the DataFrame to stdout df.show() // +----+-------+ // | age| name| // +----+-------+ // |null|Michael| // | 30| Andy| // | 19| Justin| // +----+-------+無類型數(shù)據(jù)集操作（又名DataFrame操作）
DataFrames為Scala，Java，Python和R中的結(jié)構(gòu)化數(shù)據(jù)操作提供一種DSL的語言。如上所述，在Spark 2.0中，DataFrames只是Scala和Java API中的行數(shù)據(jù)集。與“類型轉(zhuǎn)換”相比，這些操作也稱為“無類型轉(zhuǎn)換”，帶有強類型Scala / Java數(shù)據(jù)集。這里展示使用數(shù)據(jù)集進行結(jié)構(gòu)化數(shù)據(jù)處理的一些基本示例：
// This import is needed to use the $-notation import spark.implicits._ // Print the schema in a tree format df.printSchema() // root // |-- age: long (nullable = true) // |-- name: string (nullable = true)// Select only the "name" column df.select("name").show() // +-------+ // | name| // +-------+ // |Michael| // | Andy| // | Justin| // +-------+// Select everybody, but increment the age by 1 df.select($"name", $"age" + 1).show() // +-------+---------+ // | name|(age + 1)| // +-------+---------+ // |Michael| null| // | Andy| 31| // | Justin| 20| // +-------+---------+// Select people older than 21 df.filter($"age" > 21).show() // +---+----+ // |age|name| // +---+----+ // | 30|Andy| // +---+----+// Count people by age df.groupBy("age").count().show() // +----+-----+ // | age|count| // +----+-----+ // | 19| 1| // |null| 1| // | 30| 1| // +----+-----+以編程方式運行SQL查詢
SparkSession上的sql函數(shù)使應(yīng)用程序能夠以編程方式運行SQL查詢并將結(jié)果作為DataFrame返回。
// Register the DataFrame as a SQL temporary view df.createOrReplaceTempView("people")val sqlDF = spark.sql("SELECT * FROM people") sqlDF.show() // +----+-------+ // | age| name| // +----+-------+ // |null|Michael| // | 30| Andy| // | 19| Justin| // +----+-------+全局臨時視圖
Spark SQL中的臨時視圖是會話范圍的，如果創(chuàng)建它的會話終止，它將消失。 如果希望擁有一個在所有會話之間共享的臨時視圖并保持活動狀態(tài)，直到Spark應(yīng)用程序終止，您可以創(chuàng)建一個全局臨時視圖。 全局臨時視圖與系統(tǒng)保留的數(shù)據(jù)庫global_temp綁定，我們必須使用限定名稱來引用它，例如 SELECT * FROM global_temp.view1。
// Register the DataFrame as a global temporary view df.createGlobalTempView("people")// Global temporary view is tied to a system preserved database `global_temp` spark.sql("SELECT * FROM global_temp.people").show() // +----+-------+ // | age| name| // +----+-------+ // |null|Michael| // | 30| Andy| // | 19| Justin| // +----+-------+// Global temporary view is cross-session spark.newSession().sql("SELECT * FROM global_temp.people").show() // +----+-------+ // | age| name| // +----+-------+ // |null|Michael| // | 30| Andy| // | 19| Justin| // +----+-------+創(chuàng)建數(shù)據(jù)集
數(shù)據(jù)集與RDD類似，但是，它們不使用Java序列化或Kryo，而是使用專用的編碼器來序列化對象以便通過網(wǎng)絡(luò)進行處理或傳輸。 雖然編碼器和標準序列化都負責將對象轉(zhuǎn)換為字節(jié)，但編碼器是動態(tài)生成的代碼，并使用一種格式，允許Spark執(zhí)行許多操作，如過濾，排序和散列，而無需將字節(jié)反序列化為對象。
case class Person(name: String, age: Long)// Encoders are created for case classes val caseClassDS = Seq(Person("Andy", 32)).toDS() caseClassDS.show() // +----+---+ // |name|age| // +----+---+ // |Andy| 32| // +----+---+// Encoders for most common types are automatically provided by importing spark.implicits._ val primitiveDS = Seq(1, 2, 3).toDS() primitiveDS.map(_ + 1).collect() // Returns: Array(2, 3, 4)// DataFrames can be converted to a Dataset by providing a class. Mapping will be done by name val path = "examples/src/main/resources/people.json" val peopleDS = spark.read.json(path).as[Person] peopleDS.show() // +----+-------+ // | age| name| // +----+-------+ // |null|Michael| // | 30| Andy| // | 19| Justin| // +----+-------+與RDD交互操作
Spark SQL支持兩種不同的方法將現(xiàn)有RDD轉(zhuǎn)換為數(shù)據(jù)集。 第一種方法使用反射來推斷包含特定類型對象的RDD的schema。 這種基于反射的方法可以提供更簡潔的代碼，并且在您編寫Spark應(yīng)用程序時已經(jīng)了解schema時可以很好地工作。
創(chuàng)建數(shù)據(jù)集的第二種方法是通過編程接口，這種方法允許你構(gòu)建模式，然后將其應(yīng)用于現(xiàn)有RDD。 雖然此方法更繁瑣一些，但它允許在直到運行時才知道列及其類型時構(gòu)造數(shù)據(jù)集。
利用反射推斷的方法
Spark SQL的Scala接口支持自動將包含RDD的case class轉(zhuǎn)換為DataFrame。 case class用來定義表的模式。 case類的參數(shù)名稱是通過反射讀取的，這些名稱會成為列的名稱。 case類也可以被嵌套或包含復(fù)雜類型，如Seqs或Arrays。 此RDD可以隱式轉(zhuǎn)換為DataFrame，然后注冊為表。而這個表可以在后續(xù)SQL語句中使用。
// For implicit conversions from RDDs to DataFrames import spark.implicits._// Create an RDD of Person objects from a text file, convert it to a Dataframe val peopleDF = spark.sparkContext.textFile("examples/src/main/resources/people.txt").map(_.split(",")).map(attributes => Person(attributes(0), attributes(1).trim.toInt)).toDF() // Register the DataFrame as a temporary view peopleDF.createOrReplaceTempView("people")// SQL statements can be run by using the sql methods provided by Spark val teenagersDF = spark.sql("SELECT name, age FROM people WHERE age BETWEEN 13 AND 19")// The columns of a row in the result can be accessed by field index teenagersDF.map(teenager => "Name: " + teenager(0)).show() // +------------+ // | value| // +------------+ // |Name: Justin| // +------------+// or by field name teenagersDF.map(teenager => "Name: " + teenager.getAs[String]("name")).show() // +------------+ // | value| // +------------+ // |Name: Justin| // +------------+// No pre-defined encoders for Dataset[Map[K,V]], define explicitly implicit val mapEncoder = org.apache.spark.sql.Encoders.kryo[Map[String, Any]] // Primitive types and case classes can be also defined as // implicit val stringIntMapEncoder: Encoder[Map[String, Any]] = ExpressionEncoder()// row.getValuesMap[T] retrieves multiple columns at once into a Map[String, T] teenagersDF.map(teenager => teenager.getValuesMap[Any](List("name", "age"))).collect() // Array(Map("name" -> "Justin", "age" -> 19))利用編程接口聲明schema的方法
如果無法提前定義case類（例如，記錄的結(jié)構(gòu)以字符串形式編碼，或者文本數(shù)據(jù)集將被解析，字段將針對不同的用戶進行不同的映射），則可以通過三個步驟以編程方式創(chuàng)建DataFrame。
聚合
內(nèi)置的DataFrames函數(shù)提供常見的聚合，如count()，countDistinct()，avg()，max()，min()等。雖然這些函數(shù)是為DataFrames設(shè)計的，但Spark SQL也有類型安全的版本 其中一些在Scala和Java中使用強類型數(shù)據(jù)集。 此外，用戶不限于使用預(yù)定義的聚合函數(shù)，也可以創(chuàng)建自己的聚合函數(shù)。
無類型的UDAF
用戶必須擴展UserDefinedAggregateFunction抽象類以實現(xiàn)自定義無類型聚合函數(shù)。 例如，用戶定義的平均值可能如下所示：
import org.apache.spark.sql.{Row, SparkSession} import org.apache.spark.sql.expressions.MutableAggregationBuffer import org.apache.spark.sql.expressions.UserDefinedAggregateFunction import org.apache.spark.sql.types._object MyAverage extends UserDefinedAggregateFunction {// Data types of input arguments of this aggregate functiondef inputSchema: StructType = StructType(StructField("inputColumn", LongType) :: Nil)// Data types of values in the aggregation bufferdef bufferSchema: StructType = {StructType(StructField("sum", LongType) :: StructField("count", LongType) :: Nil)}// The data type of the returned valuedef dataType: DataType = DoubleType// Whether this function always returns the same output on the identical inputdef deterministic: Boolean = true// Initializes the given aggregation buffer. The buffer itself is a `Row` that in addition to// standard methods like retrieving a value at an index (e.g., get(), getBoolean()), provides// the opportunity to update its values. Note that arrays and maps inside the buffer are still// immutable.def initialize(buffer: MutableAggregationBuffer): Unit = {buffer(0) = 0Lbuffer(1) = 0L}// Updates the given aggregation buffer `buffer` with new input data from `input`def update(buffer: MutableAggregationBuffer, input: Row): Unit = {if (!input.isNullAt(0)) {buffer(0) = buffer.getLong(0) + input.getLong(0)buffer(1) = buffer.getLong(1) + 1}}// Merges two aggregation buffers and stores the updated buffer values back to `buffer1`def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {buffer1(0) = buffer1.getLong(0) + buffer2.getLong(0)buffer1(1) = buffer1.getLong(1) + buffer2.getLong(1)}// Calculates the final resultdef evaluate(buffer: Row): Double = buffer.getLong(0).toDouble / buffer.getLong(1) }// Register the function to access it spark.udf.register("myAverage", MyAverage)val df = spark.read.json("examples/src/main/resources/employees.json") df.createOrReplaceTempView("employees") df.show() // +-------+------+ // | name|salary| // +-------+------+ // |Michael| 3000| // | Andy| 4500| // | Justin| 3500| // | Berta| 4000| // +-------+------+val result = spark.sql("SELECT myAverage(salary) as average_salary FROM employees") result.show() // +--------------+ // |average_salary| // +--------------+ // | 3750.0| // +--------------+類型安全的用戶定義聚合函數(shù)
強類型數(shù)據(jù)集的用戶定義聚合通過Aggregator抽象類來實現(xiàn)。 例如，類型安全的用戶定義平均值可能如下所示：
import org.apache.spark.sql.{Encoder, Encoders, SparkSession} import org.apache.spark.sql.expressions.Aggregatorcase class Employee(name: String, salary: Long) case class Average(var sum: Long, var count: Long)object MyAverage extends Aggregator[Employee, Average, Double] {// A zero value for this aggregation. Should satisfy the property that any b + zero = bdef zero: Average = Average(0L, 0L)// Combine two values to produce a new value. For performance, the function may modify `buffer`// and return it instead of constructing a new objectdef reduce(buffer: Average, employee: Employee): Average = {buffer.sum += employee.salarybuffer.count += 1buffer}// Merge two intermediate valuesdef merge(b1: Average, b2: Average): Average = {b1.sum += b2.sumb1.count += b2.countb1}// Transform the output of the reductiondef finish(reduction: Average): Double = reduction.sum.toDouble / reduction.count// Specifies the Encoder for the intermediate value typedef bufferEncoder: Encoder[Average] = Encoders.product// Specifies the Encoder for the final output value typedef outputEncoder: Encoder[Double] = Encoders.scalaDouble }val ds = spark.read.json("examples/src/main/resources/employees.json").as[Employee] ds.show() // +-------+------+ // | name|salary| // +-------+------+ // |Michael| 3000| // | Andy| 4500| // | Justin| 3500| // | Berta| 4000| // +-------+------+// Convert the function to a `TypedColumn` and give it a name val averageSalary = MyAverage.toColumn.name("average_salary") val result = ds.select(averageSalary) result.show() // +--------------+ // |average_salary| // +--------------+ // | 3750.0| // +--------------+數(shù)據(jù)源
Spark SQL支持通過DataFrame接口對各種數(shù)據(jù)源進行操作。 DataFrame可以使用關(guān)系型轉(zhuǎn)換操作進行操作，也可以用于創(chuàng)建臨時視圖。 將DataFrame注冊為臨時視圖允許您對其數(shù)據(jù)運行SQL查詢。 下面的部分會介紹使用Spark數(shù)據(jù)源加載和保存數(shù)據(jù)的一般方法，然后介紹可用于內(nèi)置數(shù)據(jù)源的特定配置選項。
通用加載/保存功能
在最簡單的形式中，默認數(shù)據(jù)源（parquet文件，除非另外由spark.sql.sources.default配置指定）將用于所有操作。
val usersDF = spark.read.load("examples/src/main/resources/users.parquet") usersDF.select("name", "favorite_color").write.save("namesAndFavColors.parquet")手動指定選項
你還可以手動指定將要使用的數(shù)據(jù)源以及要傳遞給數(shù)據(jù)源的任何其他選項。 數(shù)據(jù)源由其完全限定名稱（即org.apache.spark.sql.parquet）指定，但對于內(nèi)置源，你還可以使用其短名稱（json，parquet，jdbc，orc，libsvm，csv，text）。 從任何數(shù)據(jù)源類型加載的DataFrame都可以使用此語法轉(zhuǎn)換為其他類型。
加載一個json文件可以用如下方法：
而加載一個csv可以這樣：
val peopleDFCsv = spark.read.format("csv").option("sep", ";").option("inferSchema", "true").option("header", "true").load("examples/src/main/resources/people.csv")在寫操作期間也使用額外選項。 例如，您可以控制ORC數(shù)據(jù)源的bloom過濾器和字典編碼。 以下ORC示例將在favorite_color上創(chuàng)建bloom過濾器，并對name和favorite_color使用字典編碼。 對于Parquet，也存在parquet.enable.dictionary。 要查找有關(guān)額外ORC / Parquet選項的更多詳細信息，請訪問官方Apache ORC / Parquet網(wǎng)站。
usersDF.write.format("orc").option("orc.bloom.filter.columns", "favorite_color").option("orc.dictionary.key.threshold", "1.0").save("users_with_options.orc")直接在文件上運行SQL
您可以直接使用SQL查詢該文件，而不是使用讀取API將文件加載到DataFrame并進行查詢。
val sqlDF = spark.sql("SELECT * FROM parquet.`examples/src/main/resources/users.parquet`")保存模式
保存操作可以有選擇地使用SaveMode，不同選項模式指定如何處理現(xiàn)有數(shù)據(jù)（如果存在）。 重要的是要意識到這些保存模式不使用任何鎖定并且不是原子的。 此外，執(zhí)行覆蓋時，將在寫出新數(shù)據(jù)之前刪除數(shù)據(jù)。
| SaveMode.ErrorIfExists (default) | "error" or "errorifexists" (default) | 將DataFrame保存到數(shù)據(jù)源時，如果數(shù)據(jù)已存在，則會引發(fā)異常。 |
| SaveMode.Append | "append" | 將DataFrame保存到數(shù)據(jù)源時，如果數(shù)據(jù)/表已存在，則DataFrame的內(nèi)容應(yīng)附加到現(xiàn)有數(shù)據(jù)。 |
| SaveMode.Overwrite | "overwrite" | 覆蓋模式意味著在將DataFrame保存到數(shù)據(jù)源時，如果數(shù)據(jù)/表已經(jīng)存在，則預(yù)期現(xiàn)有數(shù)據(jù)將被DataFrame的內(nèi)容覆蓋。 |
| SaveMode.Ignore | "ignore" | 忽略模式意味著在將DataFrame保存到數(shù)據(jù)源時，如果數(shù)據(jù)已存在，則預(yù)期保存操作不會保存DataFrame的內(nèi)容而不會更改現(xiàn)有數(shù)據(jù)。 這類似于SQL中的CREATE TABLE IF NOT EXISTS。 |
保存到持久表
也可以使用saveAsTable命令將DataFrames作為持久表保存到Hive Metastore中。請注意，使用此功能不需要現(xiàn)有的Hive部署。 Spark將為您創(chuàng)建默認的本地Hive Metastore（使用Derby）。 與createOrReplaceTempView命令不同，saveAsTable將實現(xiàn)DataFrame的內(nèi)容并創(chuàng)建指向Hive Metastore中數(shù)據(jù)的指針。 只要您保持與同一Metastore的連接，即使您的Spark程序重新啟動后，持久表仍然存在。 可以通過使用表的名稱調(diào)用SparkSession上的table方法來創(chuàng)建持久表的DataFrame。
對于基于文件的數(shù)據(jù)源，例如 text，parquet，json等，你可以通過路徑選項指定自定義表路徑，例如 df.write.option（“path”，“/ some / path”）.saveAsTable（“t”）。 刪除表時，將不會刪除自定義表路徑，并且表數(shù)據(jù)仍然存在。 如果未指定自定義表路徑，則Spark會將數(shù)據(jù)寫入倉庫目錄下的默認表路徑。 刪除表時，也將刪除默認表路徑。
從Spark 2.1開始，持久數(shù)據(jù)源表將每個分區(qū)元數(shù)據(jù)存儲在Hive Metastore中。 這帶來了幾個好處：
- 由于Metastore只能返回查詢所需的分區(qū)，因此不再需要在表的第一個查詢中發(fā)現(xiàn)所有分區(qū)。
- 現(xiàn)在，對于使用Datasource API創(chuàng)建的表，可以使用ALTER TABLE PARTITION ... SET LOCATION等Hive DDL。
請注意，在創(chuàng)建外部數(shù)據(jù)源表（具有路徑選項的表）時，默認情況下不會收集分區(qū)信息。 要同步Metastore中的分區(qū)信息，可以調(diào)用MSCK REPAIR TABLE。
分桶、排序和分區(qū)
對于基于文件的數(shù)據(jù)源，也可以對輸出進行分桶和排序或分區(qū)。 分桶和排序僅適用于持久表：
peopleDF.write.bucketBy(42, "name").sortBy("age").saveAsTable("people_bucketed")分區(qū)可以在使用數(shù)據(jù)集API時與save和saveAsTable一起使用。
usersDF.write.partitionBy("favorite_color").format("parquet").save("namesPartByColor.parquet")可以對單個表同時使用分區(qū)和分桶：
usersDF.write.partitionBy("favorite_color").bucketBy(42, "name").saveAsTable("users_partitioned_bucketed")partitionBy會創(chuàng)建一個目錄結(jié)構(gòu)，如”分區(qū)發(fā)現(xiàn)“這一章所述。 因此，它對具有高基數(shù)的列的適用性有限。 相比之下，bucketBy可以在固定數(shù)量的桶中分配數(shù)據(jù)，并且可以在出現(xiàn)許多無界的唯一值時使用。
Parquet文件
Parquet是一種面向列的存儲格式，許多數(shù)據(jù)處理系統(tǒng)都支持它。Spark SQL支持讀取和寫入Parquet文件，這些文件自動保留原始數(shù)據(jù)的schema。 在寫Parquet文件時，出于兼容性原因，所有列都會自動轉(zhuǎn)換為可為空（nullable）模式。
以編程方式加載數(shù)據(jù)
使用如下的例子來實現(xiàn)：
// Encoders for most common types are automatically provided by importing spark.implicits._ import spark.implicits._val peopleDF = spark.read.json("examples/src/main/resources/people.json")// DataFrames can be saved as Parquet files, maintaining the schema information peopleDF.write.parquet("people.parquet")// Read in the parquet file created above // Parquet files are self-describing so the schema is preserved // The result of loading a Parquet file is also a DataFrame val parquetFileDF = spark.read.parquet("people.parquet")// Parquet files can also be used to create a temporary view and then used in SQL statements parquetFileDF.createOrReplaceTempView("parquetFile") val namesDF = spark.sql("SELECT name FROM parquetFile WHERE age BETWEEN 13 AND 19") namesDF.map(attributes => "Name: " + attributes(0)).show() // +------------+ // | value| // +------------+ // |Name: Justin| // +------------+分區(qū)發(fā)現(xiàn)
表分區(qū)是Hive等系統(tǒng)中常用的優(yōu)化方法。 在分區(qū)表中，數(shù)據(jù)通常存儲在不同的目錄中，分區(qū)列值被編碼為每個分區(qū)目錄路徑。所有內(nèi)置文件源（包括Text / CSV / JSON / ORC / Parquet）都能夠自動發(fā)現(xiàn)和推斷分區(qū)信息。 例如，我們可以使用以下目錄結(jié)構(gòu)將所有以前使用的人口數(shù)據(jù)存儲到分區(qū)表中，并將兩個額外的列（性別和國家）作為分區(qū)列：
path └── to└── table├── gender=male│ ├── ...│ ││ ├── country=US│ │ └── data.parquet│ ├── country=CN│ │ └── data.parquet│ └── ...└── gender=female├── ...│├── country=US│ └── data.parquet├── country=CN│ └── data.parquet└── ...通過將 path/to/table 傳遞給SparkSession.read.parquet或SparkSession.read.load，Spark SQL將自動從路徑中提取分區(qū)信息。 現(xiàn)在返回的DataFrame的schema變?yōu)?#xff1a;
root |-- name: string (nullable = true) |-- age: long (nullable = true) |-- gender: string (nullable = true) |-- country: string (nullable = true)請注意，分區(qū)列的數(shù)據(jù)類型是自動推斷的。 目前，支持數(shù)字數(shù)據(jù)類型，日期，時間戳和字符串類型。 有時，用戶可能不希望自動推斷分區(qū)列的數(shù)據(jù)類型。 對于這些用例，可以通過spark.sql.sources.partitionColumnTypeInference.enabled配置自動類型推斷，默認為true。 禁用類型推斷時，字符串類型將用于分區(qū)列。
從Spark 1.6.0開始，分區(qū)發(fā)現(xiàn)默認只查找給定路徑下的分區(qū)。 對于上面的示例，如果用戶將path/to/table/gender=male傳遞給SparkSession.read.parquet或SparkSession.read.load，則不會將性別視為分區(qū)列。 如果用戶需要指定分區(qū)發(fā)現(xiàn)應(yīng)該從哪個基本路徑開始，則可以在數(shù)據(jù)源選項中設(shè)置basePath。 例如，當path/to/table/gender=male是數(shù)據(jù)的路徑并且用戶將basePath設(shè)置為path/to/table/時，gender將是分區(qū)列。
模式合并Schema Merging
與Protocol Buffer，Avro和Thrift一樣，Parquet也支持模式演變。 用戶可以從簡單模式開始，并根據(jù)需要逐漸向模式添加更多列。 通過這種方式，用戶可能最終得到具有不同但相互兼容的模式的多個Parquet文件。 Parquet數(shù)據(jù)源現(xiàn)在能夠自動檢測這種情況并合并所有這些文件的模式。
由于模式合并是一項相對昂貴的操作，并且在大多數(shù)情況下不是必需的，因此我們默認從1.5.0開始關(guān)閉它。 您可以通過以下兩種方式啟用它：
Hive Metastore Parquet表轉(zhuǎn)換
在讀取和寫入Hive Metastore Parquet表時，Spark SQL將嘗試使用自己的Parquet支持而不是Hive SerDe來獲得更好的性能。 此行為由spark.sql.hive.convertMetastoreParquet配置控制，默認情況下處于打開狀態(tài)。
Hive/Parquet Schema Reconciliation
從表模式處理的角度來看，Hive和Parquet之間存在兩個主要區(qū)別。
由于這個原因，在將Hive Metastore Parquet表轉(zhuǎn)換為Spark SQL Parquet表時，我們必須將Hive Metastore模式與Parquet模式進行協(xié)調(diào)。 相應(yīng)的規(guī)則是：
- 僅出現(xiàn)在Parquet模式中的任何字段都將在協(xié)調(diào)的模式中被放棄。
- 僅出現(xiàn)在Hive Metastore模式中的任何字段都將在協(xié)調(diào)模式中添加為可空字段。
元數(shù)據(jù)刷新Metadata Refreshing
Spark SQL緩存Parquet元數(shù)據(jù)以獲得更好的性能。 啟用Hive Metastore Parquet表轉(zhuǎn)換后，還會緩存這些轉(zhuǎn)換表的元數(shù)據(jù)。 如果這些表由Hive或其他外部工具更新，則需要手動刷新它們以確保元數(shù)據(jù)一致。
// spark is an existing SparkSession spark.catalog.refreshTable("my_table")配置
可以使用SparkSession上的setConf方法或使用SQL運行SET key = value命令來完成Parquet的配置。
| spark.sql.parquet.binaryAsString | false | 其他一些Parquet生產(chǎn)系統(tǒng)，特別是Impala，Hive和舊版本的Spark SQL在寫出Parquet模式時，不要區(qū)分二進制數(shù)據(jù)和字符串。 這個flag告訴Spark SQL將二進制數(shù)據(jù)解釋為字符串，以提供與這些系統(tǒng)的兼容性。 |
| spark.sql.parquet.int96AsTimestamp | true | 一些Parquet生產(chǎn)系統(tǒng)，特別是Impala和Hive，將時間戳存儲到INT96中。 這個flag告訴Spark SQL將INT96數(shù)據(jù)解釋為時間戳，以提供與這些系統(tǒng)的兼容性。 |
| spark.sql.parquet.compression.codec | snappy | 設(shè)置編寫Parquet文件時使用的壓縮編解碼器。 如果是compression或parquet.compression在聲明表的選項/屬性中指定聲明，優(yōu)先級為compression，parquet.compression，spark.sql.parquet.compression.codec。 可接受的值包括：none，uncompressed，snappy，gzip，lzo，brotli，lz4，zstd。注意zstd需要在Hadoop 2.9.0之前安裝ZStandardCodec，brotli需要要安裝BrotliCodec。 |
| spark.sql.parquet.filterPushdown | true | 設(shè)置為true時啟用Parquet過濾器下推優(yōu)化。 |
| spark.sql.hive.convertMetastoreParquet | true | 設(shè)置為false時，Spark SQL將使用Hive SerDe作為Parquet而不是內(nèi)置支持。 |
| spark.sql.parquet.mergeSchema | false | 如果為true，則Parquet數(shù)據(jù)源合并從所有數(shù)據(jù)文件收集的模式，否則從摘要文件選取模式，如果沒有可用的摘要文件，則從隨機數(shù)據(jù)文件中選取模式。 |
| spark.sql.optimizer.metadataOnly | true | 如果為true，則利用表的元數(shù)據(jù)來做僅元數(shù)據(jù)查詢優(yōu)化生成分區(qū)列而不是表掃描。 它適用于掃描的所有列都是分區(qū)列，并且查詢具有滿足distinct語義的聚合運算符的情況。 |
| spark.sql.parquet.writeLegacyFormat | false | 如果為true，則數(shù)據(jù)將以Spark 1.4及更早版本的方式寫入。 例如，十進制值將以Apache Parquet的固定長度字節(jié)數(shù)組格式編寫，供其他系統(tǒng)如Apache Hive和Apache Impala使用。如果為false，將使用Parquet中的較新格式。例如，十進制數(shù)將以基于int的格式編寫。如果打算使用Parquet輸出的對應(yīng)系統(tǒng)不支持此新格式，請設(shè)置為true。 |
ORC Files
從Spark 2.3開始，Spark使用新ORC文件格式的向量化的ORC reader來支持ORC文件。為此，新添加了以下配置。 當spark.sql.orc.impl設(shè)置為native并且spark.sql.orc.enableVectorizedReader設(shè)置為true時，向量化reader用于原生ORC表（例如，使用USING ORC子句創(chuàng)建的表）。對于Hive ORC serde表（例如，使用USING HIVE OPTIONS（fileFormat'ORC'）子句創(chuàng)建的表），當spark.sql.hive.convertMetastoreOrc也設(shè)置為true時，使用向量化reader。
| spark.sql.orc.impl | native | ORC實現(xiàn)的名稱。 它可以是 native 和 hive 之一。 native 表示在Apache ORC 1.4上構(gòu)建的原生ORC支持。 hive表示Hive 1.2.1中的ORC庫。 |
| spark.sql.orc.enableVectorizedReader | true | 在 native 實現(xiàn)中啟用矢量化orc解碼。如果 false ，則在 native 實現(xiàn)中使用新的非向量化ORC reader。 對于 hive 實現(xiàn)，這將被忽略。 |
JSON Files
Spark SQL可以自動推斷JSON數(shù)據(jù)集的模式，并將其作為Dataset[Row]加載。 可以使用Dataset[String]或JSON文件上的SparkSession.read.json()完成此轉(zhuǎn)換。
請注意，作為json文件提供的文件不是典型的JSON文件。 每行必須包含一個單獨的，自包含的有效JSON對象。 有關(guān)更多信息，請參閱JSON Lines文本格式，也稱為換行符分隔的JSON。
For a regular multi-line JSON file, set the multiLine option to true.
對于一個常規(guī)的多行JSON文件，設(shè)置multiLine選項為true。
Hive表Hive Tables
Spark SQL還支持讀取和寫入存儲在Apache Hive中的數(shù)據(jù)。 但是，由于Hive具有大量依賴項，而這些依賴項不包含在默認的Spark分發(fā)版本中。如果可以在類路徑上找到Hive依賴項，Spark將自動加載它們。 請注意，這些Hive依賴項也必須存在于所有工作節(jié)點上，因為它們需要訪問Hive序列化和反序列化庫（SerDes）才能訪問存儲在Hive中的數(shù)據(jù)。
通過在conf/中放置hive-site.xml，core-site.xml（用于安全性配置）和hdfs-site.xml（用于HDFS配置）文件來完成Hive的配置。
使用Hive時，必須使用Hive支持來實例化SparkSession，包括連接到持久化的Hive Metastore，支持Hive serdes和Hive用戶定義函數(shù)。 沒有現(xiàn)有Hive部署的用戶仍可以啟用Hive支持。 當未由hive-site.xml配置時，上下文會自動在當前目錄中創(chuàng)建metastore_db，并創(chuàng)建一個由spark.sql.warehouse.dir配置的目錄，該目錄默認為當前目錄中的spark-warehouse目錄，Spark應(yīng)用程序從此開始。 請注意，自Spark 2.0.0起，不推薦使用hive-site.xml中的hive.metastore.warehouse.dir屬性。 而是使用spark.sql.warehouse.dir指定倉庫中數(shù)據(jù)庫的默認位置。 您可能需要向啟動Spark應(yīng)用程序的用戶授予寫入權(quán)限。
指定Hive表的存儲格式
創(chuàng)建Hive表時，需要定義此表應(yīng)如何從/向文件系統(tǒng)讀取/寫入數(shù)據(jù)，即“輸入格式”和“輸出格式”。 您還需要定義此表如何將數(shù)據(jù)反序列化為行，或?qū)⑿行蛄谢癁閿?shù)據(jù)，即“serde”。 以下選項可用于指定存儲格式（“serde”，“輸入格式”，“輸出格式”），例如， CREATE TABLE src(id int) USING hive OPTIONS(fileFormat 'parquet')。 默認情況下，我們將表文件作為純文本讀取。 請注意，創(chuàng)建表時尚不支持Hive存儲handler，您可以使用Hive端的存儲handler創(chuàng)建表，并使用Spark SQL讀取它。
| fileFormat | fileFormat是一種存儲格式規(guī)范包，包括“serde”，“input format”和“output format”。 目前我們支持6種fileFormats：'sequencefile'，'rcfile'，'orc'，'parquet'，'textfile'和'avro'。 |
| inputFormat, outputFormat | 這兩個選項將相應(yīng)的InputFormat和OutputFormat類的名稱指定為字符串文字，例如org.apache.hadoop.hive.ql.io.orc.OrcInputFormat。 這兩個選項必須成對出現(xiàn)，如果已經(jīng)指定了fileFormat選項，你不能請再指定它們。 |
| serde | 此選項指定serde類的名稱。 指定fileFormat選項時，如果給定的fileFormat已經(jīng)包含了serde的信息則請勿再指定此選項。 目前“sequencefile”，“textfile”和“rcfile”不包含serde信息，您可以將此選項與這3個fileFormats一起使用。 |
| fieldDelim, escapeDelim, collectionDelim, mapkeyDelim, lineDelim | 這些選項只能與“textfile”文件格式一起使用。 它們定義了如何將文件內(nèi)容分隔為行。 |
與不同版本的Hive Metastore交互
Spark SQL的Hive支持最重要的部分之一是與Hive Metastore的交互，這使得Spark SQL能夠訪問Hive表的元數(shù)據(jù)。從Spark 1.4.0開始，可以使用單個二進制構(gòu)建的Spark SQL來查詢不同版本的Hive Metastores，使用下面描述的配置。 請注意，獨立于用于與Metastore通信的Hive版本，Spark SQL將針對Hive 1.2.1進行編譯作為內(nèi)部實現(xiàn)，并使用這些類進行內(nèi)部執(zhí)行（serdes，UDF，UDAF等）。
下面的選項用來配置Hive的版本，從而檢索元數(shù)據(jù)。
| spark.sql.hive.metastore.version | 1.2.1 | Hive metastore的版本。可選的配置從0.12.0到 2.3.3。 |
| spark.sql.hive.metastore.jars | builtin | 用來實例化HiveMetastoreClient的jar包的地址。可以是一下3個選項之一： |
| spark.sql.hive.metastore.sharedPrefixes | com.mysql.jdbc, org.postgresql, com.microsoft.sqlserver, oracle.jdbc | 以逗號分隔的類前綴列表，應(yīng)使用在Spark SQL和特定版本的Hive之間共享的類加載器加載。 舉個應(yīng)該被共享的類的示例是與Metastore進行通信所需的JDBC驅(qū)動程序。 其他需要共享的類是與已共享的類交互的類。 例如，log4j使用的自定義appender。 |
| spark.sql.hive.metastore.barrierPrefixes | (empty) | 以逗號分隔的類前綴列表，應(yīng)為Spark SQL在與每個Hive版通信時需要顯式重新加載的類。 例如，在前綴中聲明的Hive的UDF就是典型的需要被共享的。（例如 org.apache.spark.* ） |
JDBC To Other Databases
Spark SQL還包括一個可以使用JDBC從其他數(shù)據(jù)庫讀取數(shù)據(jù)的數(shù)據(jù)源。 與使用JdbcRDD相比，此功能應(yīng)該更受歡迎。 這是因為這樣操作的結(jié)果作為DataFrame返回，可以在Spark SQL中輕松處理，也可以與其他數(shù)據(jù)源連接。 JDBC數(shù)據(jù)源也更易于在Java或Python中使用，因為它不需要用戶提供ClassTag。 （請注意，這與Spark SQL JDBC服務(wù)器不同，后者允許其他應(yīng)用程序使用Spark SQL運行查詢）。
首先，您需要在spark類路徑中包含特定數(shù)據(jù)庫的JDBC驅(qū)動程序。 例如，要從Spark Shell連接到postgres，您將運行以下命令：
可以使用Data Sources API將遠程數(shù)據(jù)庫中的表加載為DataFrame或Spark SQL臨時視圖。用戶可以在數(shù)據(jù)源選項中指定JDBC連接屬性。 用戶名和密碼通常作為登錄數(shù)據(jù)源的連接屬性提供。 除連接屬性外，Spark還支持以下不區(qū)分大小寫的選項：
| url | JDBC連接串URL。特定源的連接屬性以URL的形式聲明。比如jdbc:postgresql://localhost/test?user=fred&password=secret |
| dbtable | 應(yīng)該讀取或?qū)懭氲腏DBC表。 請注意，在讀取路徑中使用它時，可以使用SQL查詢的 FROM 子句中有效的任何內(nèi)容。 例如，您也可以在括號中使用子查詢，而不是完整的表。 不允許同時指定dbtable和query選項。 |
| query | 將數(shù)據(jù)讀入Spark的查詢。指定的查詢將被括起來并用作 FROM 子句中的子查詢。 Spark還會為子查詢子句分配別名。 例如，spark將向JDBC Source發(fā)出以下形式的查詢。 SELECT <columns> FROM (<user_specified_query>) spark_gen_alias 使用此選項時，以下是一些限制。 例如： spark.read.format("jdbc") ??.option("dbtable", "(select c1, c2 from t1) as subq") ??.option("partitionColumn", "subq.c1") ??.load() |
| driver | JDBC驅(qū)動的類名。 |
| partitionColumn, lowerBound, upperBound | 如果指定了任何選項，則必須全部指定這些選項。 此外，必須指定 numPartitions 。 它們描述了在從多個工作者并行讀取時如何對表進行分區(qū)。 partitionColumn 必須是相關(guān)表中的數(shù)字、日期或時間戳列。 請注意， lowerBound 和 upperBound 僅用于決定分區(qū)步幅，而不是用于過濾表中的行。 因此，表中的所有行都將被分區(qū)并返回。 此選項僅適用于讀數(shù)據(jù)。 |
| numPartitions | 可用于并行讀取和寫入的表的最大分區(qū)數(shù)。還確定了最大并發(fā)JDBC連接數(shù)。如果要寫入的分區(qū)數(shù)超過此限制，我們通過在寫入之前調(diào)用coalesce(numPartitions)將其減少到此限制。 |
| queryTimeout | 驅(qū)動程序等待Statement對象執(zhí)行到指定秒數(shù)的超時時長。 0意味著沒有限制。在寫入路徑中，此選項取決于JDBC驅(qū)動程序如何實現(xiàn) setQueryTimeout 這個API，例如，h2 JDBC驅(qū)動程序檢查每個查詢的超時而不是整個JDBC批處理。它默認為 0 。 |
| fetchsize | JDBC的fetch大小，用于確定每次讀取回合要獲取的行數(shù)。這有助于JDBC驅(qū)動程序的性能，默認為低fetch大小（例如，Oracle是10行）。 此選項僅適用于讀取。 |
| batchsize | JDBC批處理大小，用于確定每次IO往返要插入的行數(shù)。 這有助于JDBC驅(qū)動程序的性能。此選項僅適用于寫入。默認為 1000 。 |
| isolationLevel | 事務(wù)隔離級別，適用于當前連接。它可以是 NONE ， READ_COMMITTED ， READ_UNCOMMITTED ， REPEATABLE_READ 或 SERIALIZABLE 之一 ，對應(yīng)于JDBC的Connection對象定義的標準事務(wù)隔離級別，默認為 READ_UNCOMMITTED 。此選項僅適用于寫入。 請參閱 java.sql.Connection 中的文檔。 |
| sessionInitStatement | 在向遠程數(shù)據(jù)庫打開每個數(shù)據(jù)庫會話之后，在開始讀取數(shù)據(jù)之前，此選項將執(zhí)行自定義SQL語句（或PL/SQL塊）。使用它來實現(xiàn)會話初始化代碼。 示例：option("sessionInitStatement", """BEGIN execute immediate 'alter session set "_serial_direct_read"=true'; END;""") |
| truncate | 這是JDBC writer相關(guān)選項。啟用 SaveMode.Overwrite code>時，此選項會導致Spark截斷現(xiàn)有表，而不是刪除并重新創(chuàng)建它。 這可以更有效，并且防止刪除表元數(shù)據(jù)（例如，索引）。 但是，在某些情況下，例如新數(shù)據(jù)具有不同的schema時，它將無法工作。 它默認為 false 。 此選項僅適用于寫入。 |
| cascadeTruncate | 這是JDBC writer相關(guān)選項。 如果JDBC數(shù)據(jù)庫（目前是PostgreSQL和Oracle）啟用并支持，則此選項允許執(zhí)行 TRUNCATE TABLE t CASCADE （在PostgreSQL的情況下， TRUNCATE TABLE ONLY t CASCADE 以防止無意中截斷下層的表）。這將影響其他表，因此應(yīng)謹慎使用。 此選項僅適用于寫入。它默認為當前配置的JDBC數(shù)據(jù)庫的默認級聯(lián)截斷行為，在每個JDBCDialect中的 isCascadeTruncate 中指定。 |
| createTableOptions | 這是JDBC writer相關(guān)選項。如果指定，則此選項允許在創(chuàng)建表時設(shè)置特定于數(shù)據(jù)庫的表和分區(qū)選項（例如，CREATE TABLE t (name string) ENGINE=InnoDB）。此選項僅適用于寫入。 |
| createTableColumnTypes | 創(chuàng)建表時要使用的數(shù)據(jù)庫列的數(shù)據(jù)類型而不是默認值。應(yīng)以與CREATE TABLE列語法相同的格式指定數(shù)據(jù)類型信息（例如："name CHAR(64), comments VARCHAR(1024)")。指定的類型應(yīng)該是有效的spark sql數(shù)據(jù)類型。此選項僅適用于寫入。 |
| customSchema | 用于從JDBC連接器讀取數(shù)據(jù)的自定義schema。例如，"id DECIMAL(38, 0), name STRING"。 您還可以只指定部分字段，其他字段使用默認類型映射。 例如，"id DECIMAL(38, 0)"。 列名應(yīng)與JDBC表的相應(yīng)列名相同。用戶可以指定Spark SQL的相應(yīng)數(shù)據(jù)類型，而不是使用默認值。此選項僅適用于讀取。 |
| pushDownPredicate | 這個選項用于在JDBC數(shù)據(jù)源啟用或禁用謂詞下推。默認值為true，在這種情況下，Spark會盡可能地將過濾條件下推到JDBC數(shù)據(jù)源。否則，如果設(shè)置為false，則不會將過濾條件下推到JDBC數(shù)據(jù)源，因此所有過濾條件都將由Spark處理。當Spark能夠比JDBC數(shù)據(jù)源更快地執(zhí)行謂詞過濾時，謂詞下推通常會被關(guān)閉。 |
Apache Avro 數(shù)據(jù)源
從Spark 2.4后，Spark SQL提供對于讀寫Apache Avro數(shù)據(jù)的內(nèi)置支持。
部署
spark-avro模塊是外置的，默認情況下不包含在spark-submit或spark-shell中。
與任何Spark應(yīng)用程序一樣，spark-submit用于啟動您的應(yīng)用程序。 使用--packages可以將spark-avro_2.11及其依賴項直接添加到spark-submit，例如，
./bin/spark-submit --packages org.apache.spark:spark-avro_2.11:2.4.0 ...對于在spark-shell上進行試驗，您也可以使用--packages直接添加org.apache.sparkspark-avro_2.11及其依賴項，
./bin/spark-shell --packages org.apache.spark:spark-avro_2.11:2.4.0 ...Load and Save Functions
由于spark-avro模塊是外部的，因此DataFrameReader或DataFrameWriter中沒有.avro API。
要以Avro格式加載/保存數(shù)據(jù)，您需要將數(shù)據(jù)源選項格式指定為avro（或org.apache.spark.sql.avro）。
val usersDF = spark.read.format("avro").load("examples/src/main/resources/users.avro") usersDF.select("name", "favorite_color").write.format("avro").save("namesAndFavColors.avro")to_avro() and from_avro()
Avro軟件包提供了to_avro函數(shù)，可以將列編碼為Avro格式的二進制文件，from_avro()將Avro二進制數(shù)據(jù)解碼為列。兩個函數(shù)都將一列轉(zhuǎn)換為另一列，輸入/輸出SQL數(shù)據(jù)類型可以是復(fù)雜類型或基本類型。
在讀取或?qū)懭胂馣afka這樣的流式數(shù)據(jù)源時，將Avro記錄作為列非常有用。 每個Kafka鍵值記錄都會增加一些元數(shù)據(jù)，例如Kafka的攝取時間戳，Kafka的偏移量等。
- 如果包含數(shù)據(jù)的“value”字段位于Avro中，則可以使用from_avro()提取數(shù)據(jù)，豐富數(shù)據(jù)，清理數(shù)據(jù)，然后再將其推送到Kafka下游或?qū)⑵鋵懭胛募?/li>
- to_avro()可用于將結(jié)構(gòu)體轉(zhuǎn)換為Avro記錄。 在將數(shù)據(jù)寫入Kafka時，如果要將多個列重新編碼為單個列，此方法特別有用。
這兩個方法目前僅支持Scala和Java。
數(shù)據(jù)源選項
Avro的數(shù)據(jù)源選項可以通過DataFrameReader或者DataFrameWriter的.option方法來設(shè)置。
| avroSchema | None | 用戶以JSON格式提供可選的Avro schema。記錄字段的日期類型和命名應(yīng)匹配輸入的Avro數(shù)據(jù)或Catalyst數(shù)據(jù)，否則讀/寫操作將失敗。 | read and write |
| recordName | topLevelRecord | 在寫入結(jié)果時的頂層記錄名字，這在Avro的spec是需要的 | write |
| recordNamespace | "" | 寫入結(jié)果的記錄命名空間 | write |
| ignoreExtension | true | 該選項控制在讀取時忽略沒有 .avro 擴展名的文件。 如果啟用該選項，則加載所有文件（帶有和不帶 .avro 擴展名）。 | read |
| compression | snappy | compression 選項允許指定write中使用的壓縮編解碼器 目前支持的編解碼器有 uncompressed ， snappy ， deflate ， bzip2 和 xz 。 如果未設(shè)置該選項，則要考慮配置spark.sql.avro.compression.codec | write |
配置
可以使用SparkSession的setConf方法或使用SQL運行SET key = value命令來完成Avro的配置。
| spark.sql.legacy.replaceDatabricksSparkAvro.enabled | true | 如果設(shè)置為true，則數(shù)據(jù)源提供者 com.databricks.spark.avro 將映射到內(nèi)置的外部Avro數(shù)據(jù)源模塊，以實現(xiàn)向后兼容性。 |
| spark.sql.avro.compression.codec | snappy | 用于編寫AVRO文件的壓縮編解碼器。支持的編解碼器：uncompressed，deflate，snappy，bzip2和xz。默認編解碼器是snappy。 |
| spark.sql.avro.deflate.level | -1 | 用于編寫AVRO文件的deflate編解碼器的壓縮級別。 有效值必須介于1到9之間（包括1或9）或-1。 默認值為-1，對應(yīng)于當前實現(xiàn)中的6級。 |
Compatibility with Databricks spark-avro
此Avro數(shù)據(jù)源模塊最初來自Databricks的開源存儲庫spark-avro并與之兼容。
默認情況下，啟用SQL配置spark.sql.legacy.replaceDatabricksSparkAvro.enabled，數(shù)據(jù)源提供者com.databricks.spark.avro將映射到此內(nèi)置Avro模塊。對于在目錄元數(shù)據(jù)庫中使用Provider屬性創(chuàng)建的Spark表作為com.databricks.spark.avro，如果您使用此內(nèi)置Avro模塊，則映射對于加載這些表至關(guān)重要。
請注意，在Databricks的spark-avro中，為快捷函數(shù).avro()創(chuàng)建了隱式類AvroDataFrameWriter和AvroDataFrameReader。在這個內(nèi)置但外部的模塊中，兩個隱式類都被刪除了。請改用DataFrameWriter或DataFrameReader中的.format（“avro”），它應(yīng)該干凈且足夠好。
如果您更喜歡使用自己構(gòu)建的spark-avro jar文件，則只需禁用配置spark.sql.legacy.replaceDatabricksSparkAvro.enabled，并在部署應(yīng)用程序時使用選項--jars。有關(guān)詳細信息，請閱讀“應(yīng)用程序提交指南”中的“高級依賴關(guān)系管理”部分。
Supported types for Avro -> Spark SQL conversion
目前，Spark支持在Avro記錄下讀取所有原始類型和復(fù)雜類型。
| boolean | BooleanType |
| int | IntegerType |
| long | LongType |
| float | FloatType |
| double | DoubleType |
| string | StringType |
| enum | StringType |
| fixed | BinaryType |
| bytes | BinaryType |
| record | StructType |
| array | ArrayType |
| map | MapType |
| union | See below |
除了上面列出的類型，它還支持讀取聯(lián)合類型。 以下三種類型被視為基本聯(lián)合類型：
它還支持讀取以下Avro邏輯類型：
| date | int | DateType |
| timestamp-millis | long | TimestampType |
| timestamp-micros | long | TimestampType |
| decimal | fixed | DecimalType |
| decimal | bytes | DecimalType |
目前，它忽略了Avro文件中存在的文檔，別名和其他屬性。
Supported types for Spark SQL -> Avro conversion
Spark支持將所有Spark SQL類型寫入Avro。 對于大多數(shù)類型，從Spark類型到Avro類型的映射很簡單（例如，IntegerType轉(zhuǎn)換為int）; 但是，下面列出了一些特殊情況：
| ByteType | int | |
| ShortType | int | |
| BinaryType | bytes | |
| DateType | int | date |
| TimestampType | long | timestamp-micros |
| DecimalType | fixed | decimal |
您還可以使用選項avroSchema指定整個輸出Avro schema，以便可以將Spark SQL類型轉(zhuǎn)換為其他Avro類型。 默認情況下不應(yīng)用以下轉(zhuǎn)換，并且需要用戶指定的Avro schema：
| BinaryType | fixed | |
| StringType | enum | |
| TimestampType | long | timestamp-millis |
| DecimalType | bytes | decimal |
故障排除Troubleshooting
- JDBC驅(qū)動程序類必須對客戶端會話和所有執(zhí)行程序上的原始類加載器可見。 這是因為Java的DriverManager類進行了安全檢查，導致它忽略了當打開連接時原始類加載器不可見的所有驅(qū)動程序。 一種方便的方法是修改所有工作節(jié)點上的compute_classpath.sh以包含驅(qū)動程序JAR。
- 某些數(shù)據(jù)庫（如H2）會將所有名稱轉(zhuǎn)換為大寫。您需要使用大寫字母在Spark SQL中引用這些名稱。
- 用戶可以在數(shù)據(jù)源選項中指定特定于供應(yīng)商的JDBC連接屬性以進行特殊處理。例如，spark.read.format("jdbc").option("url", oracleJdbcUrl).option("oracle.jdbc.mapDateToTimestamp", "false")。 oracle.jdbc.mapDateToTimestamp默認為true，用戶通常需要禁用此標志以避免Oracle日期被解析為時間戳。
性能調(diào)優(yōu)
對于某些工作負載，可以通過在內(nèi)存中緩存數(shù)據(jù)或打開一些實驗選項來提高性能。
Caching Data In Memory
Spark SQL可以通過調(diào)用spark.catalog.cacheTable("tableName")或dataFrame.cache()使用內(nèi)存中的列式格式來緩存表。 然后，Spark SQL將僅掃描所需的列，并自動調(diào)整壓縮以最小化內(nèi)存使用和GC壓力。 您可以調(diào)用spark.catalog.uncacheTable("tableName")從內(nèi)存中刪除表。
可以使用SparkSession的setConf方法或使用SQL運行SET key = value命令來完成內(nèi)存中緩存的配置。
| spark.sql.inMemoryColumnarStorage.compressed | true | 設(shè)置為true時，Spark SQL將根據(jù)數(shù)據(jù)統(tǒng)計信息自動為每列選擇壓縮編解碼器。 |
| spark.sql.inMemoryColumnarStorage.batchSize | 10000 | 控制列存緩存的批次大小。較大的批處理大小可以提高內(nèi)存利用率和壓縮率，但在緩存數(shù)據(jù)時存在OOM風險。 |
其他配置項
以下選項也可用于調(diào)整查詢執(zhí)行的性能。由于更多優(yōu)化會自動執(zhí)行，因此在將來的版本中可能會棄用這些選項。
| spark.sql.files.maxPartitionBytes | 134217728 (128 MB) | 讀取文件時打包到單個分區(qū)的最大字節(jié)數(shù)。 |
| spark.sql.files.openCostInBytes | 4194304 (4 MB) | 打開文件的估計成本是通過可以在同一時間掃描的字節(jié)數(shù)測量的。這在將多個文件放入分區(qū)時是有用的。最好是做過度估計，這樣使用較小文件的分區(qū)將比較大文件的分區(qū)（首先安排的分區(qū)）更快。 |
| spark.sql.broadcastTimeout | 300 | 廣播連接中廣播等待時間的超時（以秒為單位） |
| spark.sql.autoBroadcastJoinThreshold | 10485760 (10 MB) | 配置在執(zhí)行join時將廣播到所有工作節(jié)點的表的最大大小（以字節(jié)為單位）。 通過將此值設(shè)置為-1，可以禁用廣播。請注意，當前的統(tǒng)計信息僅支持Hive Metastore表，并且其中命令A(yù)NALYZE TABLE <tableName> COMPUTE STATISTICS noscan已經(jīng)運行。 |
| spark.sql.shuffle.partitions | 200 | 配置在為join或聚合shuffle數(shù)據(jù)時要使用的分區(qū)數(shù)。 |
Broadcast Hint for SQL Queries
BROADCAST hint指導Spark在將其與另一個表或視圖join時廣播每個指定的表。 當Spark決定join方法時，廣播散列連接（broadcast hash join即BHJ）是首選，即使統(tǒng)計信息高于spark.sql.autoBroadcastJoinThreshold配置的。當join的兩端都被指定時，Spark會廣播具有較低統(tǒng)計信息的那一方。 注意Spark并不保證始終選擇BHJ，因為并非所有情況（例如全外連接）都支持BHJ。 當選擇廣播嵌套循環(huán)連接（broadcast nested loop join）時，我們?nèi)匀宦爮膆int的。
import org.apache.spark.sql.functions.broadcast broadcast(spark.table("src")).join(spark.table("records"), "key").show()分布式SQL引擎Distributed SQL Engine
Spark SQL還可以使用其JDBC/ODBC或命令行界面充當分布式查詢引擎。 在此模式下，最終用戶或應(yīng)用程序可以直接與Spark SQL交互以運行SQL查詢，而無需編寫任何代碼。
Running the Thrift JDBC/ODBC server
此處實現(xiàn)的Thrift JDBC/ODBC服務(wù)器對應(yīng)于Hive 1.2.1中的HiveServer2。 您可以使用Spark或Hive 1.2.1附帶的beeline腳本測試JDBC服務(wù)器。
要啟動JDBC / ODBC服務(wù)器，請在Spark目錄中運行以下命令：
./sbin/start-thriftserver.sh此腳本接受所有bin/spark-submit命令行選項，以及--hiveconf選項以指定Hive屬性。 您可以運行./sbin/start-thriftserver.sh --help以獲取所有可用選項的完整列表。默認情況下，服務(wù)器監(jiān)聽localhost:10000。您可以通過任一環(huán)境變量覆蓋此行為，例如：
export HIVE_SERVER2_THRIFT_PORT=<listening-port> export HIVE_SERVER2_THRIFT_BIND_HOST=<listening-host> ./sbin/start-thriftserver.sh \--master <master-uri> \...或者系統(tǒng)屬性system properties
./sbin/start-thriftserver.sh \--hiveconf hive.server2.thrift.port=<listening-port> \--hiveconf hive.server2.thrift.bind.host=<listening-host> \--master <master-uri>...現(xiàn)在您可以使用beeline來測試Thrift JDBC/ODBC服務(wù)器：
./bin/beeline使用以下方式直接連接到JDBC/ODBC服務(wù)器：
beeline> !connect jdbc:hive2://localhost:10000Beeline會詢問您的用戶名和密碼。在非安全模式下，只需在您的計算機上輸入用戶名和空白密碼即可。對于安全模式，請按照beeline文檔中的說明進行操作。
通過將hive-site.xml，core-site.xml和hdfs-site.xml文件放在conf/中來完成Hive的配置。
您也可以使用Hive附帶的beeline腳本。
Thrift JDBC服務(wù)器還支持通過HTTP傳輸發(fā)送thrift RPC消息。使用以下設(shè)置將HTTP模式作為系統(tǒng)屬性或在conf/中的hive-site.xml文件中啟用：
hive.server2.transport.mode - Set this to value: http hive.server2.thrift.http.port - HTTP port number to listen on; default is 10001 hive.server2.http.endpoint - HTTP endpoint; default is cliservice要進行測試，請使用beeline以http模式連接到JDBC/ODBC服務(wù)器：
beeline> !connect jdbc:hive2://<host>:<port>/<database>?hive.server2.transport.mode=http;hive.server2.thrift.http.path=<http_endpoint>Running the Spark SQL CLI
Spark SQL CLI是一種方便的工具，可以在本地模式下運行Hive Metastore服務(wù)，并執(zhí)行從命令行輸入的查詢。 請注意，Spark SQL CLI無法與Thrift JDBC服務(wù)器通信。
要啟動Spark SQL CLI，請在Spark目錄中運行以下命令：
./bin/spark-sql通過將hive-site.xml，core-site.xml和hdfs-site.xml文件放在conf/中來完成Hive的配置。 您可以運行./bin/spark-sql --help以獲取所有可用選項的完整列表。
總結(jié)
以上是生活随笔為你收集整理的Spark SQL玩起来的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 聊聊技术路线的选择
- 下一篇： mysql 面试知识点笔记（二）查询优化