Spark SQL玩起来
標簽（空格分隔）： Spark
[toc]
前言
Spark SQL的介紹只包含官方文檔的Getting Started、DataSource、Performance Tuning和Distributed SQL Engine部分。不含其他的遷移和PySpark等部分。
Spark SQL介紹
Spark SQL是一個Spark模塊用于結構化數據處理。與基本的Spark RDD API不同，Spark SQL提供的接口為Spark提供了有關數據結構和正在執行的計算的更多信息。 在內部，Spark SQL使用此額外信息來執行額外的優化。 有幾種與Spark SQL交互的方法，包括SQL和Dataset API。 在使用相同的執行引擎計算結果時，與使用表達計算的API或者語言無關。 這種統一意味著開發人員可以輕松地在不同的API之間來回切換，從而提供表達給定轉換的最自然的方式。
SQL
Spark SQL的一個用途是執行SQL查詢。Spark SQL還可用于從現有Hive中讀取數據。 有關如何配置此功能的更多信息，請參閱Hive Tables部分。 從其他編程語言中運行SQL時，結果將作為Dataset/DataFrame返回。 還可以使用命令行或JDBC/ODBC與SQL接口進行交互。
Dataset和DataFrame
Dataset數據集是分布式數據集合。數據集是Spark 1.6中添加的一個新接口，它提供了RDD的優勢（強類型，使用強大的lambda函數的能力）和Spark SQL優化執行引擎的優點。 數據集可以從JVM對象構造，然后使用功能轉換（map，flatMap，filter等）進行操作。 數據集API在Scala和Java中可用。 Python沒有對Dataset API的支持。 但由于Python的動態特性，數據集API的許多好處已經可用（即可以通過名稱自然地訪問行的字段row.columnName）。 R的情況類似。
DataFrame是一個組織成命名列的數據集。 它在概念上等同于關系數據庫中的表或R / Python中的數據框，但在底層具有更豐富的優化。 DataFrame可以從多種來源構建，例如：結構化數據文件，Hive中的表，外部數據庫或現有RDD。 DataFrame API在Scala，Java，Python和R中可用。在Scala和Java中，DataFrame由行數據集表示。 在Scala API中，DataFrame只是Dataset[Row]的類型別名。 而在Java API中，用戶需要使用Dataset<Row>來表示DataFrame。
Spark SQL入門知識
SparkSession
Spark中所有功能的入口點是SparkSession類。通過類似下面的代碼來創建：
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功能提供內置支持，包括使用HiveQL編寫查詢，訪問Hive UDF以及從Hive表讀取數據的功能。 要使用這些功能，并不需擁有現有的Hive設置。
創建DataFrame
使用SparkSession，應用程序可以從現有RDD，Hive表或Spark數據源創建DataFrame。作為示例，以下內容基于JSON文件的內容創建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| // +----+-------+無類型數據集操作（又名DataFrame操作）
DataFrames為Scala，Java，Python和R中的結構化數據操作提供一種DSL的語言。如上所述，在Spark 2.0中，DataFrames只是Scala和Java API中的行數據集。與“類型轉換”相比，這些操作也稱為“無類型轉換”，帶有強類型Scala / Java數據集。這里展示使用數據集進行結構化數據處理的一些基本示例：
// 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函數使應用程序能夠以編程方式運行SQL查詢并將結果作為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中的臨時視圖是會話范圍的，如果創建它的會話終止，它將消失。 如果希望擁有一個在所有會話之間共享的臨時視圖并保持活動狀態，直到Spark應用程序終止，您可以創建一個全局臨時視圖。 全局臨時視圖與系統保留的數據庫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| // +----+-------+創建數據集
數據集與RDD類似，但是，它們不使用Java序列化或Kryo，而是使用專用的編碼器來序列化對象以便通過網絡進行處理或傳輸。 雖然編碼器和標準序列化都負責將對象轉換為字節，但編碼器是動態生成的代碼，并使用一種格式，允許Spark執行許多操作，如過濾，排序和散列，而無需將字節反序列化為對象。
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支持兩種不同的方法將現有RDD轉換為數據集。 第一種方法使用反射來推斷包含特定類型對象的RDD的schema。 這種基于反射的方法可以提供更簡潔的代碼，并且在您編寫Spark應用程序時已經了解schema時可以很好地工作。
創建數據集的第二種方法是通過編程接口，這種方法允許你構建模式，然后將其應用于現有RDD。 雖然此方法更繁瑣一些，但它允許在直到運行時才知道列及其類型時構造數據集。
利用反射推斷的方法
Spark SQL的Scala接口支持自動將包含RDD的case class轉換為DataFrame。 case class用來定義表的模式。 case類的參數名稱是通過反射讀取的，這些名稱會成為列的名稱。 case類也可以被嵌套或包含復雜類型，如Seqs或Arrays。 此RDD可以隱式轉換為DataFrame，然后注冊為表。而這個表可以在后續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類（例如，記錄的結構以字符串形式編碼，或者文本數據集將被解析，字段將針對不同的用戶進行不同的映射），則可以通過三個步驟以編程方式創建DataFrame。
聚合
內置的DataFrames函數提供常見的聚合，如count()，countDistinct()，avg()，max()，min()等。雖然這些函數是為DataFrames設計的，但Spark SQL也有類型安全的版本 其中一些在Scala和Java中使用強類型數據集。 此外，用戶不限于使用預定義的聚合函數，也可以創建自己的聚合函數。
無類型的UDAF
用戶必須擴展UserDefinedAggregateFunction抽象類以實現自定義無類型聚合函數。 例如，用戶定義的平均值可能如下所示：
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| // +--------------+類型安全的用戶定義聚合函數
強類型數據集的用戶定義聚合通過Aggregator抽象類來實現。 例如，類型安全的用戶定義平均值可能如下所示：
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| // +--------------+數據源
Spark SQL支持通過DataFrame接口對各種數據源進行操作。 DataFrame可以使用關系型轉換操作進行操作，也可以用于創建臨時視圖。 將DataFrame注冊為臨時視圖允許您對其數據運行SQL查詢。 下面的部分會介紹使用Spark數據源加載和保存數據的一般方法，然后介紹可用于內置數據源的特定配置選項。
通用加載/保存功能
在最簡單的形式中，默認數據源（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")手動指定選項
你還可以手動指定將要使用的數據源以及要傳遞給數據源的任何其他選項。 數據源由其完全限定名稱（即org.apache.spark.sql.parquet）指定，但對于內置源，你還可以使用其短名稱（json，parquet，jdbc，orc，libsvm，csv，text）。 從任何數據源類型加載的DataFrame都可以使用此語法轉換為其他類型。
加載一個json文件可以用如下方法：
而加載一個csv可以這樣：
val peopleDFCsv = spark.read.format("csv").option("sep", ";").option("inferSchema", "true").option("header", "true").load("examples/src/main/resources/people.csv")在寫操作期間也使用額外選項。 例如，您可以控制ORC數據源的bloom過濾器和字典編碼。 以下ORC示例將在favorite_color上創建bloom過濾器，并對name和favorite_color使用字典編碼。 對于Parquet，也存在parquet.enable.dictionary。 要查找有關額外ORC / Parquet選項的更多詳細信息，請訪問官方Apache ORC / Parquet網站。
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，不同選項模式指定如何處理現有數據（如果存在）。 重要的是要意識到這些保存模式不使用任何鎖定并且不是原子的。 此外，執行覆蓋時，將在寫出新數據之前刪除數據。
| SaveMode.ErrorIfExists (default) | "error" or "errorifexists" (default) | 將DataFrame保存到數據源時，如果數據已存在，則會引發異常。 |
| SaveMode.Append | "append" | 將DataFrame保存到數據源時，如果數據/表已存在，則DataFrame的內容應附加到現有數據。 |
| SaveMode.Overwrite | "overwrite" | 覆蓋模式意味著在將DataFrame保存到數據源時，如果數據/表已經存在，則預期現有數據將被DataFrame的內容覆蓋。 |
| SaveMode.Ignore | "ignore" | 忽略模式意味著在將DataFrame保存到數據源時，如果數據已存在，則預期保存操作不會保存DataFrame的內容而不會更改現有數據。 這類似于SQL中的CREATE TABLE IF NOT EXISTS。 |
保存到持久表
也可以使用saveAsTable命令將DataFrames作為持久表保存到Hive Metastore中。請注意，使用此功能不需要現有的Hive部署。 Spark將為您創建默認的本地Hive Metastore（使用Derby）。 與createOrReplaceTempView命令不同，saveAsTable將實現DataFrame的內容并創建指向Hive Metastore中數據的指針。 只要您保持與同一Metastore的連接，即使您的Spark程序重新啟動后，持久表仍然存在。 可以通過使用表的名稱調用SparkSession上的table方法來創建持久表的DataFrame。
對于基于文件的數據源，例如 text，parquet，json等，你可以通過路徑選項指定自定義表路徑，例如 df.write.option（“path”，“/ some / path”）.saveAsTable（“t”）。 刪除表時，將不會刪除自定義表路徑，并且表數據仍然存在。 如果未指定自定義表路徑，則Spark會將數據寫入倉庫目錄下的默認表路徑。 刪除表時，也將刪除默認表路徑。
從Spark 2.1開始，持久數據源表將每個分區元數據存儲在Hive Metastore中。 這帶來了幾個好處：
- 由于Metastore只能返回查詢所需的分區，因此不再需要在表的第一個查詢中發現所有分區。
- 現在，對于使用Datasource API創建的表，可以使用ALTER TABLE PARTITION ... SET LOCATION等Hive DDL。
請注意，在創建外部數據源表（具有路徑選項的表）時，默認情況下不會收集分區信息。 要同步Metastore中的分區信息，可以調用MSCK REPAIR TABLE。
分桶、排序和分區
對于基于文件的數據源，也可以對輸出進行分桶和排序或分區。 分桶和排序僅適用于持久表：
peopleDF.write.bucketBy(42, "name").sortBy("age").saveAsTable("people_bucketed")分區可以在使用數據集API時與save和saveAsTable一起使用。
usersDF.write.partitionBy("favorite_color").format("parquet").save("namesPartByColor.parquet")可以對單個表同時使用分區和分桶：
usersDF.write.partitionBy("favorite_color").bucketBy(42, "name").saveAsTable("users_partitioned_bucketed")partitionBy會創建一個目錄結構，如”分區發現“這一章所述。 因此，它對具有高基數的列的適用性有限。 相比之下，bucketBy可以在固定數量的桶中分配數據，并且可以在出現許多無界的唯一值時使用。
Parquet文件
Parquet是一種面向列的存儲格式，許多數據處理系統都支持它。Spark SQL支持讀取和寫入Parquet文件，這些文件自動保留原始數據的schema。 在寫Parquet文件時，出于兼容性原因，所有列都會自動轉換為可為空（nullable）模式。
以編程方式加載數據
使用如下的例子來實現：
// 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| // +------------+分區發現
表分區是Hive等系統中常用的優化方法。 在分區表中，數據通常存儲在不同的目錄中，分區列值被編碼為每個分區目錄路徑。所有內置文件源（包括Text / CSV / JSON / ORC / Parquet）都能夠自動發現和推斷分區信息。 例如，我們可以使用以下目錄結構將所有以前使用的人口數據存儲到分區表中，并將兩個額外的列（性別和國家）作為分區列：
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將自動從路徑中提取分區信息。 現在返回的DataFrame的schema變為：
root |-- name: string (nullable = true) |-- age: long (nullable = true) |-- gender: string (nullable = true) |-- country: string (nullable = true)請注意，分區列的數據類型是自動推斷的。 目前，支持數字數據類型，日期，時間戳和字符串類型。 有時，用戶可能不希望自動推斷分區列的數據類型。 對于這些用例，可以通過spark.sql.sources.partitionColumnTypeInference.enabled配置自動類型推斷，默認為true。 禁用類型推斷時，字符串類型將用于分區列。
從Spark 1.6.0開始，分區發現默認只查找給定路徑下的分區。 對于上面的示例，如果用戶將path/to/table/gender=male傳遞給SparkSession.read.parquet或SparkSession.read.load，則不會將性別視為分區列。 如果用戶需要指定分區發現應該從哪個基本路徑開始，則可以在數據源選項中設置basePath。 例如，當path/to/table/gender=male是數據的路徑并且用戶將basePath設置為path/to/table/時，gender將是分區列。
模式合并Schema Merging
與Protocol Buffer，Avro和Thrift一樣，Parquet也支持模式演變。 用戶可以從簡單模式開始，并根據需要逐漸向模式添加更多列。 通過這種方式，用戶可能最終得到具有不同但相互兼容的模式的多個Parquet文件。 Parquet數據源現在能夠自動檢測這種情況并合并所有這些文件的模式。
由于模式合并是一項相對昂貴的操作，并且在大多數情況下不是必需的，因此我們默認從1.5.0開始關閉它。 您可以通過以下兩種方式啟用它：
Hive Metastore Parquet表轉換
在讀取和寫入Hive Metastore Parquet表時，Spark SQL將嘗試使用自己的Parquet支持而不是Hive SerDe來獲得更好的性能。 此行為由spark.sql.hive.convertMetastoreParquet配置控制，默認情況下處于打開狀態。
Hive/Parquet Schema Reconciliation
從表模式處理的角度來看，Hive和Parquet之間存在兩個主要區別。
由于這個原因，在將Hive Metastore Parquet表轉換為Spark SQL Parquet表時，我們必須將Hive Metastore模式與Parquet模式進行協調。 相應的規則是：
- 僅出現在Parquet模式中的任何字段都將在協調的模式中被放棄。
- 僅出現在Hive Metastore模式中的任何字段都將在協調模式中添加為可空字段。
元數據刷新Metadata Refreshing
Spark SQL緩存Parquet元數據以獲得更好的性能。 啟用Hive Metastore Parquet表轉換后，還會緩存這些轉換表的元數據。 如果這些表由Hive或其他外部工具更新，則需要手動刷新它們以確保元數據一致。
// spark is an existing SparkSession spark.catalog.refreshTable("my_table")配置
可以使用SparkSession上的setConf方法或使用SQL運行SET key = value命令來完成Parquet的配置。
| spark.sql.parquet.binaryAsString | false | 其他一些Parquet生產系統，特別是Impala，Hive和舊版本的Spark SQL在寫出Parquet模式時，不要區分二進制數據和字符串。 這個flag告訴Spark SQL將二進制數據解釋為字符串，以提供與這些系統的兼容性。 |
| spark.sql.parquet.int96AsTimestamp | true | 一些Parquet生產系統，特別是Impala和Hive，將時間戳存儲到INT96中。 這個flag告訴Spark SQL將INT96數據解釋為時間戳，以提供與這些系統的兼容性。 |
| spark.sql.parquet.compression.codec | snappy | 設置編寫Parquet文件時使用的壓縮編解碼器。 如果是compression或parquet.compression在聲明表的選項/屬性中指定聲明，優先級為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 | 設置為true時啟用Parquet過濾器下推優化。 |
| spark.sql.hive.convertMetastoreParquet | true | 設置為false時，Spark SQL將使用Hive SerDe作為Parquet而不是內置支持。 |
| spark.sql.parquet.mergeSchema | false | 如果為true，則Parquet數據源合并從所有數據文件收集的模式，否則從摘要文件選取模式，如果沒有可用的摘要文件，則從隨機數據文件中選取模式。 |
| spark.sql.optimizer.metadataOnly | true | 如果為true，則利用表的元數據來做僅元數據查詢優化生成分區列而不是表掃描。 它適用于掃描的所有列都是分區列，并且查詢具有滿足distinct語義的聚合運算符的情況。 |
| spark.sql.parquet.writeLegacyFormat | false | 如果為true，則數據將以Spark 1.4及更早版本的方式寫入。 例如，十進制值將以Apache Parquet的固定長度字節數組格式編寫，供其他系統如Apache Hive和Apache Impala使用。如果為false，將使用Parquet中的較新格式。例如，十進制數將以基于int的格式編寫。如果打算使用Parquet輸出的對應系統不支持此新格式，請設置為true。 |
ORC Files
從Spark 2.3開始，Spark使用新ORC文件格式的向量化的ORC reader來支持ORC文件。為此，新添加了以下配置。 當spark.sql.orc.impl設置為native并且spark.sql.orc.enableVectorizedReader設置為true時，向量化reader用于原生ORC表（例如，使用USING ORC子句創建的表）。對于Hive ORC serde表（例如，使用USING HIVE OPTIONS（fileFormat'ORC'）子句創建的表），當spark.sql.hive.convertMetastoreOrc也設置為true時，使用向量化reader。
| spark.sql.orc.impl | native | ORC實現的名稱。 它可以是 native 和 hive 之一。 native 表示在Apache ORC 1.4上構建的原生ORC支持。 hive表示Hive 1.2.1中的ORC庫。 |
| spark.sql.orc.enableVectorizedReader | true | 在 native 實現中啟用矢量化orc解碼。如果 false ，則在 native 實現中使用新的非向量化ORC reader。 對于 hive 實現，這將被忽略。 |
JSON Files
Spark SQL可以自動推斷JSON數據集的模式，并將其作為Dataset[Row]加載。 可以使用Dataset[String]或JSON文件上的SparkSession.read.json()完成此轉換。
請注意，作為json文件提供的文件不是典型的JSON文件。 每行必須包含一個單獨的，自包含的有效JSON對象。 有關更多信息，請參閱JSON Lines文本格式，也稱為換行符分隔的JSON。
For a regular multi-line JSON file, set the multiLine option to true.
對于一個常規的多行JSON文件，設置multiLine選項為true。
Hive表Hive Tables
Spark SQL還支持讀取和寫入存儲在Apache Hive中的數據。 但是，由于Hive具有大量依賴項，而這些依賴項不包含在默認的Spark分發版本中。如果可以在類路徑上找到Hive依賴項，Spark將自動加載它們。 請注意，這些Hive依賴項也必須存在于所有工作節點上，因為它們需要訪問Hive序列化和反序列化庫（SerDes）才能訪問存儲在Hive中的數據。
通過在conf/中放置hive-site.xml，core-site.xml（用于安全性配置）和hdfs-site.xml（用于HDFS配置）文件來完成Hive的配置。
使用Hive時，必須使用Hive支持來實例化SparkSession，包括連接到持久化的Hive Metastore，支持Hive serdes和Hive用戶定義函數。 沒有現有Hive部署的用戶仍可以啟用Hive支持。 當未由hive-site.xml配置時，上下文會自動在當前目錄中創建metastore_db，并創建一個由spark.sql.warehouse.dir配置的目錄，該目錄默認為當前目錄中的spark-warehouse目錄，Spark應用程序從此開始。 請注意，自Spark 2.0.0起，不推薦使用hive-site.xml中的hive.metastore.warehouse.dir屬性。 而是使用spark.sql.warehouse.dir指定倉庫中數據庫的默認位置。 您可能需要向啟動Spark應用程序的用戶授予寫入權限。
指定Hive表的存儲格式
創建Hive表時，需要定義此表應如何從/向文件系統讀取/寫入數據，即“輸入格式”和“輸出格式”。 您還需要定義此表如何將數據反序列化為行，或將行序列化為數據，即“serde”。 以下選項可用于指定存儲格式（“serde”，“輸入格式”，“輸出格式”），例如， CREATE TABLE src(id int) USING hive OPTIONS(fileFormat 'parquet')。 默認情況下，我們將表文件作為純文本讀取。 請注意，創建表時尚不支持Hive存儲handler，您可以使用Hive端的存儲handler創建表，并使用Spark SQL讀取它。
| fileFormat | fileFormat是一種存儲格式規范包，包括“serde”，“input format”和“output format”。 目前我們支持6種fileFormats：'sequencefile'，'rcfile'，'orc'，'parquet'，'textfile'和'avro'。 |
| inputFormat, outputFormat | 這兩個選項將相應的InputFormat和OutputFormat類的名稱指定為字符串文字，例如org.apache.hadoop.hive.ql.io.orc.OrcInputFormat。 這兩個選項必須成對出現，如果已經指定了fileFormat選項，你不能請再指定它們。 |
| serde | 此選項指定serde類的名稱。 指定fileFormat選項時，如果給定的fileFormat已經包含了serde的信息則請勿再指定此選項。 目前“sequencefile”，“textfile”和“rcfile”不包含serde信息，您可以將此選項與這3個fileFormats一起使用。 |
| fieldDelim, escapeDelim, collectionDelim, mapkeyDelim, lineDelim | 這些選項只能與“textfile”文件格式一起使用。 它們定義了如何將文件內容分隔為行。 |
與不同版本的Hive Metastore交互
Spark SQL的Hive支持最重要的部分之一是與Hive Metastore的交互，這使得Spark SQL能夠訪問Hive表的元數據。從Spark 1.4.0開始，可以使用單個二進制構建的Spark SQL來查詢不同版本的Hive Metastores，使用下面描述的配置。 請注意，獨立于用于與Metastore通信的Hive版本，Spark SQL將針對Hive 1.2.1進行編譯作為內部實現，并使用這些類進行內部執行（serdes，UDF，UDAF等）。
下面的選項用來配置Hive的版本，從而檢索元數據。
| spark.sql.hive.metastore.version | 1.2.1 | Hive metastore的版本?？蛇x的配置從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 | 以逗號分隔的類前綴列表，應使用在Spark SQL和特定版本的Hive之間共享的類加載器加載。 舉個應該被共享的類的示例是與Metastore進行通信所需的JDBC驅動程序。 其他需要共享的類是與已共享的類交互的類。 例如，log4j使用的自定義appender。 |
| spark.sql.hive.metastore.barrierPrefixes | (empty) | 以逗號分隔的類前綴列表，應為Spark SQL在與每個Hive版通信時需要顯式重新加載的類。 例如，在前綴中聲明的Hive的UDF就是典型的需要被共享的。（例如 org.apache.spark.* ） |
JDBC To Other Databases
Spark SQL還包括一個可以使用JDBC從其他數據庫讀取數據的數據源。 與使用JdbcRDD相比，此功能應該更受歡迎。 這是因為這樣操作的結果作為DataFrame返回，可以在Spark SQL中輕松處理，也可以與其他數據源連接。 JDBC數據源也更易于在Java或Python中使用，因為它不需要用戶提供ClassTag。 （請注意，這與Spark SQL JDBC服務器不同，后者允許其他應用程序使用Spark SQL運行查詢）。
首先，您需要在spark類路徑中包含特定數據庫的JDBC驅動程序。 例如，要從Spark Shell連接到postgres，您將運行以下命令：
可以使用Data Sources API將遠程數據庫中的表加載為DataFrame或Spark SQL臨時視圖。用戶可以在數據源選項中指定JDBC連接屬性。 用戶名和密碼通常作為登錄數據源的連接屬性提供。 除連接屬性外，Spark還支持以下不區分大小寫的選項：
| url | JDBC連接串URL。特定源的連接屬性以URL的形式聲明。比如jdbc:postgresql://localhost/test?user=fred&password=secret |
| dbtable | 應該讀取或寫入的JDBC表。 請注意，在讀取路徑中使用它時，可以使用SQL查詢的 FROM 子句中有效的任何內容。 例如，您也可以在括號中使用子查詢，而不是完整的表。 不允許同時指定dbtable和query選項。 |
| query | 將數據讀入Spark的查詢。指定的查詢將被括起來并用作 FROM 子句中的子查詢。 Spark還會為子查詢子句分配別名。 例如，spark將向JDBC Source發出以下形式的查詢。 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驅動的類名。 |
| partitionColumn, lowerBound, upperBound | 如果指定了任何選項，則必須全部指定這些選項。 此外，必須指定 numPartitions 。 它們描述了在從多個工作者并行讀取時如何對表進行分區。 partitionColumn 必須是相關表中的數字、日期或時間戳列。 請注意， lowerBound 和 upperBound 僅用于決定分區步幅，而不是用于過濾表中的行。 因此，表中的所有行都將被分區并返回。 此選項僅適用于讀數據。 |
| numPartitions | 可用于并行讀取和寫入的表的最大分區數。還確定了最大并發JDBC連接數。如果要寫入的分區數超過此限制，我們通過在寫入之前調用coalesce(numPartitions)將其減少到此限制。 |
| queryTimeout | 驅動程序等待Statement對象執行到指定秒數的超時時長。 0意味著沒有限制。在寫入路徑中，此選項取決于JDBC驅動程序如何實現 setQueryTimeout 這個API，例如，h2 JDBC驅動程序檢查每個查詢的超時而不是整個JDBC批處理。它默認為 0 。 |
| fetchsize | JDBC的fetch大小，用于確定每次讀取回合要獲取的行數。這有助于JDBC驅動程序的性能，默認為低fetch大小（例如，Oracle是10行）。 此選項僅適用于讀取。 |
| batchsize | JDBC批處理大小，用于確定每次IO往返要插入的行數。 這有助于JDBC驅動程序的性能。此選項僅適用于寫入。默認為 1000 。 |
| isolationLevel | 事務隔離級別，適用于當前連接。它可以是 NONE ， READ_COMMITTED ， READ_UNCOMMITTED ， REPEATABLE_READ 或 SERIALIZABLE 之一 ，對應于JDBC的Connection對象定義的標準事務隔離級別，默認為 READ_UNCOMMITTED 。此選項僅適用于寫入。 請參閱 java.sql.Connection 中的文檔。 |
| sessionInitStatement | 在向遠程數據庫打開每個數據庫會話之后，在開始讀取數據之前，此選項將執行自定義SQL語句（或PL/SQL塊）。使用它來實現會話初始化代碼。 示例：option("sessionInitStatement", """BEGIN execute immediate 'alter session set "_serial_direct_read"=true'; END;""") |
| truncate | 這是JDBC writer相關選項。啟用 SaveMode.Overwrite code>時，此選項會導致Spark截斷現有表，而不是刪除并重新創建它。 這可以更有效，并且防止刪除表元數據（例如，索引）。 但是，在某些情況下，例如新數據具有不同的schema時，它將無法工作。 它默認為 false 。 此選項僅適用于寫入。 |
| cascadeTruncate | 這是JDBC writer相關選項。 如果JDBC數據庫（目前是PostgreSQL和Oracle）啟用并支持，則此選項允許執行 TRUNCATE TABLE t CASCADE （在PostgreSQL的情況下， TRUNCATE TABLE ONLY t CASCADE 以防止無意中截斷下層的表）。這將影響其他表，因此應謹慎使用。 此選項僅適用于寫入。它默認為當前配置的JDBC數據庫的默認級聯截斷行為，在每個JDBCDialect中的 isCascadeTruncate 中指定。 |
| createTableOptions | 這是JDBC writer相關選項。如果指定，則此選項允許在創建表時設置特定于數據庫的表和分區選項（例如，CREATE TABLE t (name string) ENGINE=InnoDB）。此選項僅適用于寫入。 |
| createTableColumnTypes | 創建表時要使用的數據庫列的數據類型而不是默認值。應以與CREATE TABLE列語法相同的格式指定數據類型信息（例如："name CHAR(64), comments VARCHAR(1024)")。指定的類型應該是有效的spark sql數據類型。此選項僅適用于寫入。 |
| customSchema | 用于從JDBC連接器讀取數據的自定義schema。例如，"id DECIMAL(38, 0), name STRING"。 您還可以只指定部分字段，其他字段使用默認類型映射。 例如，"id DECIMAL(38, 0)"。 列名應與JDBC表的相應列名相同。用戶可以指定Spark SQL的相應數據類型，而不是使用默認值。此選項僅適用于讀取。 |
| pushDownPredicate | 這個選項用于在JDBC數據源啟用或禁用謂詞下推。默認值為true，在這種情況下，Spark會盡可能地將過濾條件下推到JDBC數據源。否則，如果設置為false，則不會將過濾條件下推到JDBC數據源，因此所有過濾條件都將由Spark處理。當Spark能夠比JDBC數據源更快地執行謂詞過濾時，謂詞下推通常會被關閉。 |
Apache Avro 數據源
從Spark 2.4后，Spark SQL提供對于讀寫Apache Avro數據的內置支持。
部署
spark-avro模塊是外置的，默認情況下不包含在spark-submit或spark-shell中。
與任何Spark應用程序一樣，spark-submit用于啟動您的應用程序。 使用--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格式加載/保存數據，您需要將數據源選項格式指定為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函數，可以將列編碼為Avro格式的二進制文件，from_avro()將Avro二進制數據解碼為列。兩個函數都將一列轉換為另一列，輸入/輸出SQL數據類型可以是復雜類型或基本類型。
在讀取或寫入像Kafka這樣的流式數據源時，將Avro記錄作為列非常有用。 每個Kafka鍵值記錄都會增加一些元數據，例如Kafka的攝取時間戳，Kafka的偏移量等。
- 如果包含數據的“value”字段位于Avro中，則可以使用from_avro()提取數據，豐富數據，清理數據，然后再將其推送到Kafka下游或將其寫入文件。
- to_avro()可用于將結構體轉換為Avro記錄。 在將數據寫入Kafka時，如果要將多個列重新編碼為單個列，此方法特別有用。
這兩個方法目前僅支持Scala和Java。
數據源選項
Avro的數據源選項可以通過DataFrameReader或者DataFrameWriter的.option方法來設置。
| avroSchema | None | 用戶以JSON格式提供可選的Avro schema。記錄字段的日期類型和命名應匹配輸入的Avro數據或Catalyst數據，否則讀/寫操作將失敗。 | read and write |
| recordName | topLevelRecord | 在寫入結果時的頂層記錄名字，這在Avro的spec是需要的 | write |
| recordNamespace | "" | 寫入結果的記錄命名空間 | write |
| ignoreExtension | true | 該選項控制在讀取時忽略沒有 .avro 擴展名的文件。 如果啟用該選項，則加載所有文件（帶有和不帶 .avro 擴展名）。 | read |
| compression | snappy | compression 選項允許指定write中使用的壓縮編解碼器 目前支持的編解碼器有 uncompressed ， snappy ， deflate ， bzip2 和 xz 。 如果未設置該選項，則要考慮配置spark.sql.avro.compression.codec | write |
配置
可以使用SparkSession的setConf方法或使用SQL運行SET key = value命令來完成Avro的配置。
| spark.sql.legacy.replaceDatabricksSparkAvro.enabled | true | 如果設置為true，則數據源提供者 com.databricks.spark.avro 將映射到內置的外部Avro數據源模塊，以實現向后兼容性。 |
| 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，對應于當前實現中的6級。 |
Compatibility with Databricks spark-avro
此Avro數據源模塊最初來自Databricks的開源存儲庫spark-avro并與之兼容。
默認情況下，啟用SQL配置spark.sql.legacy.replaceDatabricksSparkAvro.enabled，數據源提供者com.databricks.spark.avro將映射到此內置Avro模塊。對于在目錄元數據庫中使用Provider屬性創建的Spark表作為com.databricks.spark.avro，如果您使用此內置Avro模塊，則映射對于加載這些表至關重要。
請注意，在Databricks的spark-avro中，為快捷函數.avro()創建了隱式類AvroDataFrameWriter和AvroDataFrameReader。在這個內置但外部的模塊中，兩個隱式類都被刪除了。請改用DataFrameWriter或DataFrameReader中的.format（“avro”），它應該干凈且足夠好。
如果您更喜歡使用自己構建的spark-avro jar文件，則只需禁用配置spark.sql.legacy.replaceDatabricksSparkAvro.enabled，并在部署應用程序時使用選項--jars。有關詳細信息，請閱讀“應用程序提交指南”中的“高級依賴關系管理”部分。
Supported types for Avro -> Spark SQL conversion
目前，Spark支持在Avro記錄下讀取所有原始類型和復雜類型。
| 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 |
除了上面列出的類型，它還支持讀取聯合類型。 以下三種類型被視為基本聯合類型：
它還支持讀取以下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。 對于大多數類型，從Spark類型到Avro類型的映射很簡單（例如，IntegerType轉換為int）; 但是，下面列出了一些特殊情況：
| ByteType | int | |
| ShortType | int | |
| BinaryType | bytes | |
| DateType | int | date |
| TimestampType | long | timestamp-micros |
| DecimalType | fixed | decimal |
您還可以使用選項avroSchema指定整個輸出Avro schema，以便可以將Spark SQL類型轉換為其他Avro類型。 默認情況下不應用以下轉換，并且需要用戶指定的Avro schema：
| BinaryType | fixed | |
| StringType | enum | |
| TimestampType | long | timestamp-millis |
| DecimalType | bytes | decimal |
故障排除Troubleshooting
- JDBC驅動程序類必須對客戶端會話和所有執行程序上的原始類加載器可見。 這是因為Java的DriverManager類進行了安全檢查，導致它忽略了當打開連接時原始類加載器不可見的所有驅動程序。 一種方便的方法是修改所有工作節點上的compute_classpath.sh以包含驅動程序JAR。
- 某些數據庫（如H2）會將所有名稱轉換為大寫。您需要使用大寫字母在Spark SQL中引用這些名稱。
- 用戶可以在數據源選項中指定特定于供應商的JDBC連接屬性以進行特殊處理。例如，spark.read.format("jdbc").option("url", oracleJdbcUrl).option("oracle.jdbc.mapDateToTimestamp", "false")。 oracle.jdbc.mapDateToTimestamp默認為true，用戶通常需要禁用此標志以避免Oracle日期被解析為時間戳。
性能調優
對于某些工作負載，可以通過在內存中緩存數據或打開一些實驗選項來提高性能。
Caching Data In Memory
Spark SQL可以通過調用spark.catalog.cacheTable("tableName")或dataFrame.cache()使用內存中的列式格式來緩存表。 然后，Spark SQL將僅掃描所需的列，并自動調整壓縮以最小化內存使用和GC壓力。 您可以調用spark.catalog.uncacheTable("tableName")從內存中刪除表。
可以使用SparkSession的setConf方法或使用SQL運行SET key = value命令來完成內存中緩存的配置。
| spark.sql.inMemoryColumnarStorage.compressed | true | 設置為true時，Spark SQL將根據數據統計信息自動為每列選擇壓縮編解碼器。 |
| spark.sql.inMemoryColumnarStorage.batchSize | 10000 | 控制列存緩存的批次大小。較大的批處理大小可以提高內存利用率和壓縮率，但在緩存數據時存在OOM風險。 |
其他配置項
以下選項也可用于調整查詢執行的性能。由于更多優化會自動執行，因此在將來的版本中可能會棄用這些選項。
| spark.sql.files.maxPartitionBytes | 134217728 (128 MB) | 讀取文件時打包到單個分區的最大字節數。 |
| spark.sql.files.openCostInBytes | 4194304 (4 MB) | 打開文件的估計成本是通過可以在同一時間掃描的字節數測量的。這在將多個文件放入分區時是有用的。最好是做過度估計，這樣使用較小文件的分區將比較大文件的分區（首先安排的分區）更快。 |
| spark.sql.broadcastTimeout | 300 | 廣播連接中廣播等待時間的超時（以秒為單位） |
| spark.sql.autoBroadcastJoinThreshold | 10485760 (10 MB) | 配置在執行join時將廣播到所有工作節點的表的最大大小（以字節為單位）。 通過將此值設置為-1，可以禁用廣播。請注意，當前的統計信息僅支持Hive Metastore表，并且其中命令ANALYZE TABLE <tableName> COMPUTE STATISTICS noscan已經運行。 |
| spark.sql.shuffle.partitions | 200 | 配置在為join或聚合shuffle數據時要使用的分區數。 |
Broadcast Hint for SQL Queries
BROADCAST hint指導Spark在將其與另一個表或視圖join時廣播每個指定的表。 當Spark決定join方法時，廣播散列連接（broadcast hash join即BHJ）是首選，即使統計信息高于spark.sql.autoBroadcastJoinThreshold配置的。當join的兩端都被指定時，Spark會廣播具有較低統計信息的那一方。 注意Spark并不保證始終選擇BHJ，因為并非所有情況（例如全外連接）都支持BHJ。 當選擇廣播嵌套循環連接（broadcast nested loop join）時，我們仍然聽從hint的。
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或命令行界面充當分布式查詢引擎。 在此模式下，最終用戶或應用程序可以直接與Spark SQL交互以運行SQL查詢，而無需編寫任何代碼。
Running the Thrift JDBC/ODBC server
此處實現的Thrift JDBC/ODBC服務器對應于Hive 1.2.1中的HiveServer2。 您可以使用Spark或Hive 1.2.1附帶的beeline腳本測試JDBC服務器。
要啟動JDBC / ODBC服務器，請在Spark目錄中運行以下命令：
./sbin/start-thriftserver.sh此腳本接受所有bin/spark-submit命令行選項，以及--hiveconf選項以指定Hive屬性。 您可以運行./sbin/start-thriftserver.sh --help以獲取所有可用選項的完整列表。默認情況下，服務器監聽localhost:10000。您可以通過任一環境變量覆蓋此行為，例如：
export HIVE_SERVER2_THRIFT_PORT=<listening-port> export HIVE_SERVER2_THRIFT_BIND_HOST=<listening-host> ./sbin/start-thriftserver.sh \--master <master-uri> \...或者系統屬性system properties
./sbin/start-thriftserver.sh \--hiveconf hive.server2.thrift.port=<listening-port> \--hiveconf hive.server2.thrift.bind.host=<listening-host> \--master <master-uri>...現在您可以使用beeline來測試Thrift JDBC/ODBC服務器：
./bin/beeline使用以下方式直接連接到JDBC/ODBC服務器：
beeline> !connect jdbc:hive2://localhost:10000Beeline會詢問您的用戶名和密碼。在非安全模式下，只需在您的計算機上輸入用戶名和空白密碼即可。對于安全模式，請按照beeline文檔中的說明進行操作。
通過將hive-site.xml，core-site.xml和hdfs-site.xml文件放在conf/中來完成Hive的配置。
您也可以使用Hive附帶的beeline腳本。
Thrift JDBC服務器還支持通過HTTP傳輸發送thrift RPC消息。使用以下設置將HTTP模式作為系統屬性或在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服務器：
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服務，并執行從命令行輸入的查詢。 請注意，Spark SQL CLI無法與Thrift JDBC服務器通信。
要啟動Spark SQL CLI，請在Spark目錄中運行以下命令：
./bin/spark-sql通過將hive-site.xml，core-site.xml和hdfs-site.xml文件放在conf/中來完成Hive的配置。 您可以運行./bin/spark-sql --help以獲取所有可用選項的完整列表。
總結
以上是生活随笔為你收集整理的Spark SQL玩起来的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 聊聊技术路线的选择
- 下一篇： Leetcode PHP题解--D16