通用load/write方法

手動指定選項

Spark SQL的DataFrame介面支援多種資料來源的操作。一個DataFrame可以進行RDDs方式的操作，也可以被註冊為臨時表。把DataFrame註冊為臨時表之後，就可以對該DataFrame執行SQL查詢。

Spark SQL的預設資料來源為Parquet格式。資料來源為Parquet檔案時，Spark SQL可以方便的執行所有的操作。

修改配置項spark.sql.sources.default，可修改預設資料來源格式。

scala> val df = spark.read.load("hdfs://hadoop001:9000/namesAndAges.parquet")
df: org.apache.spark.sql.DataFrame = [age: bigint,name: string]
scala> df.select("name").write.save("names.parquet")

當資料來源格式不是parquet格式檔案時，需要手動指定資料來源的格式。資料來源格式需要指定全名（例如：org.apache.spark.sql.parquet），如果資料來源格式為內建格式，則只需要指定簡稱json,parquet,jdbc,orc,libsvm,csv,text來指定資料的格式。

可以通過SparkSession提供的read.load方法用於通用載入資料，使用write和save儲存資料。

scala> val peopleDF = spark.read.format("json").load("hdfs://hadoop001:9000/people.json")
peopleDF: org.apache.spark.sql.DataFrame = [age: bigint,name: string]
scala> peopleDF.write.format("parquet").save("hdfs://hadoop001:9000/namesAndAges.parquet")
scala>

除此之外，可以直接執行SQL在檔案上:

val sqlDF = spark.sql("SELECT * FROM parquet.`hdfs://hadoop001:9000/namesAndAges.parquet`")
sqlDF.show()

檔案儲存選項

可以採用SaveMode執行儲存操作，SaveMode定義了對資料的處理模式。需要注意的是，這些儲存模式不使用任何鎖定，不是原子操作。此外，當使用Overwrite方式執行時，在輸出新資料之前原資料就已經被刪除。SaveMode詳細介紹如下表：

Scala/Java	Any Language	Meaning
SaveMode.ErrorIfExists(default)	“error”(default)	如果檔案存在，則報錯
SaveMode.Append	“append”	追加
SaveMode.Overwrite	“overwrite”	覆寫
SaveMode.Ignore	“ignore”	資料存在，則忽略

Parquet檔案

Parquet讀寫

Parquet格式經常在Hadoop生態圈中被使用，它也支援Spark SQL的全部資料型別。Spark SQL 提供了直接讀取和儲存 Parquet 格式檔案的方法。

// 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("hdfs://hadoop001:9000/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("hdfs://hadoop001:9000/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|
// +------------+

解析分割槽資訊

對錶進行分割槽是對資料進行優化的方式之一。在分割槽的表內，資料通過分割槽列將資料儲存在不同的目錄下。Parquet資料來源現在能夠自動發現並解析分割槽資訊。例如，對人口資料進行分割槽儲存，分割槽列為gender和country，使用下面的目錄結構：

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給 SQLContext.read.parque

或SQLContext.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。

如果想關閉該功能，直接將該引數設定為disabled。此時，分割槽列資料格式將被預設設定為string型別，不再進行型別解析。

Schema合併

像ProtocolBuffer、Avro和Thrift那樣，Parquet也支援Schema evolution（Schema演變）。使用者可以先定義一個簡單的Schema，然後逐漸的向Schema中增加列描述。通過這種方式，使用者可以獲取多個有不同Schema但相互相容的Parquet檔案。現在Parquet資料來源能自動檢測這種情況，併合並這些檔案的schemas。

因為Schema合併是一個高消耗的操作，在大多數情況下並不需要，所以Spark SQL從1.5.0開始預設關閉了該功能。可以通過下面兩種方式開啟該功能：

當資料來源為Parquet檔案時，將資料來源選項mergeSchema設定為true。

設定全域性SQL選項：

spark.sql.parquet.mergeSchema為true。

// sqlContext from the previous example is used in this example.
// This is used to implicitly convert an RDD to a DataFrame.
import spark.implicits._
// Create a simple DataFrame,stored into a partition directory
val df1 = sc.makeRDD(1 to 5).map(i => (i,i * 2)).toDF("single","double")
df1.write.parquet("hdfs://hadoop001:9000/data/test_table/key=1")
// Create another DataFrame in a new partition directory,// adding a new column and dropping an existing column
val df2 = sc.makeRDD(6 to 10).map(i => (i,i * 3)).toDF("single","triple")
df2.write.parquet("hdfs://hadoop001:9000/data/test_table/key=2")
// Read the partitioned table
val df3 = spark.read.option("mergeSchema","true").parquet("hdfs://hadoop001:9000/data/test_table")
df3.printSchema()
// The final schema consists of all 3 columns in the Parquet files together
// with the partitioning column appeared in the partition directory paths.
// root
// |-- single: int (nullable = true)
// |-- double: int (nullable = true)
// |-- triple: int (nullable = true)
// |-- key : int (nullable = true)

Hive資料來源

Apache Hive是Hadoop上的SQL引擎，Spark SQL編譯時可以包含Hive支援，也可以不包含。包含Hive支援的Spark SQL可以支援Hive表訪問、UDF(使用者自定義函式)以及 Hive 查詢語言(HiveQL/HQL)等。需要強調的 一點是，如果要在Spark SQL中包含Hive的庫，並不需要事先安裝Hive。一般來說，最好還是在編譯Spark SQL時引入Hive支援，這樣就可以使用這些特性了。如果你下載的是二進位制版本的 Spark，它應該已經在編譯時添加了 Hive 支援。

若要把Spark SQL連線到一個部署好的Hive上，你必須把hive-site.xml複製到 Spark的配置檔案目錄中($SPARK_HOME/conf)。即使沒有部署好Hive，Spark SQL也可以執行。

需要注意的是，如果你沒有部署好Hive，Spark SQL會在當前的工作目錄中創建出自己的Hive 元資料倉庫，叫作 metastore_db。此外，如果你嘗試使用 HiveQL 中的 CREATE TABLE (並非 CREATE EXTERNAL TABLE)語句來建立表，這些表會被放在你預設的檔案系統中的 /user/hive/warehouse 目錄中(如果你的 classpath 中有配好的 hdfs-site.xml，預設的檔案系統就是 HDFS，否則就是本地檔案系統)。

import java.io.File
import org.apache.spark.sql.Row
import org.apache.spark.sql.SparkSession
case class Record(key: Int,value: String)
// warehouseLocation points to the default location for managed databases and tables
val warehouseLocation = new File("spark-warehouse").getAbsolutePath
val spark = SparkSession
.builder()
.appName("Spark Hive Example")
.config("spark.sql.warehouse.dir",warehouseLocation)
.enableHiveSupport()
.getOrCreate()
import spark.implicits._
import spark.sql
sql("CREATE TABLE IF NOT EXISTS src (key INT,value STRING)")
sql("LOAD DATA LOCAL INPATH 'examples/src/main/resources/kv1.txt' INTO TABLE src")
// Queries are expressed in HiveQL
sql("SELECT * FROM src").show()
// +---+-------+
// |key| value|
// +---+-------+
// |238|val_238|
// | 86| val_86|
// |311|val_311|
// ...
// Aggregation queries are also supported.
sql("SELECT COUNT(*) FROM src").show()
// +--------+
// |count(1)|
// +--------+
// | 500 |
// +--------+
// The results of SQL queries are themselves DataFrames and support all normal functions.
val sqlDF = sql("SELECT key,value FROM src WHERE key < 10 ORDER BY key")
// The items in DataFrames are of type Row,which allows you to access each column by ordinal.
val stringsDS = sqlDF.map {
case Row(key: Int,value: String) => s"Key: $key,Value: $value"
}
stringsDS.show()
// +--------------------+
// | value|
// +--------------------+
// |Key: 0,Value: val_0|
// |Key: 0,Value: val_0|
// ...
// You can also use DataFrames to create temporary views within a SparkSession.
val recordsDF = spark.createDataFrame((1 to 100).map(i => Record(i,s"val_$i")))
recordsDF.createOrReplaceTempView("records")
// Queries can then join DataFrame data with data stored in Hive.
sql("SELECT * FROM records r JOIN src s ON r.key = s.key").show()
// +---+------+---+------+
// |key| value|key| value|
// +---+------+---+------+
// | 2| val_2| 2| val_2|
// | 4| val_4| 4| val_4|
// | 5| val_5| 5| val_5|
// ...

內嵌Hive應用

如果要使用內嵌的Hive，什麼都不用做，直接用就可以了。 –conf :

spark.sql.warehouse.dir=

注意：如果你使用的是內部的Hive，在Spark2.0之後，spark.sql.warehouse.dir用於指定資料倉庫的地址，如果你需要是用HDFS作為路徑，那麼需要將core-site.xml和hdfs-site.xml 加入到Spark conf目錄，否則只會建立master節點上的warehouse目錄，查詢時會出現檔案找不到的問題，這是需要向使用HDFS，則需要將metastore刪除，重啟叢集。

外部Hive應用

如果想連線外部已經部署好的Hive，需要通過以下幾個步驟。

a 將Hive中的hive-site.xml拷貝或者軟連線到Spark安裝目錄下的conf目錄下。

b 開啟spark shell，注意帶上訪問Hive元資料庫的JDBC客戶端。

$ bin/spark-shell --master spark://hadoop001:7077 --jars mysql-connector-java-5.1.27-bin.jar

JSON資料集

Spark SQL 能夠自動推測 JSON資料集的結構，並將它載入為一個Dataset[Row]. 可以通過SparkSession.read.json()去載入一個 Dataset[String]或者一個JSON 檔案.注意，這個JSON檔案不是一個傳統的JSON檔案，每一行都得是一個JSON串。

{"name":"Michael"}
{"name":"Andy","age":30}
{"name":"Justin","age":19}
// Primitive types (Int,String,etc) and Product types (case classes) encoders are
// supported by importing this when creating a Dataset.
import spark.implicits._
// A JSON dataset is pointed to by path.
// The path can be either a single text file or a directory storing text files
val path = "examples/src/main/resources/people.json"
val peopleDF = spark.read.json(path)
// The inferred schema can be visualized using the printSchema() method
peopleDF.printSchema()
// root
// |-- age: long (nullable = true)
// |-- name: string (nullable = true)
// Creates a temporary view using the DataFrame
peopleDF.createOrReplaceTempView("people")
// SQL statements can be run by using the sql methods provided by spark
val teenagerNamesDF = spark.sql("SELECT name FROM people WHERE age BETWEEN 13 AND 19")
teenagerNamesDF.show()
// +------+
// | name|
// +------+
// |Justin|
// +------+
// Alternatively,a DataFrame can be created for a JSON dataset represented by
// a Dataset[String] storing one JSON object per string
val otherPeopleDataset = spark.createDataset(
"""{"name":"Yin","address":{"city":"Columbus","state":"Ohio"}}""" :: Nil)
val otherPeople = spark.read.json(otherPeopleDataset)
otherPeople.show()
// +---------------+----+
// | address|name|
// +---------------+----+
// |[Columbus,Ohio]| Yin|
// +---------------+----+

JDBC

Spark SQL可以通過JDBC從關係型資料庫中讀取資料的方式建立DataFrame，通過對DataFrame一系列的計算後，還可以將資料再寫回關係型資料庫中。

注意，需要將相關的資料庫驅動放到spark的類路徑下。

$ bin/spark-shell --master spark://hadoop001:7077 --jars mysql-connector-java-5.1.27-bin.jar
// Note: JDBC loading and saving can be achieved via either the load/save or jdbc methods
// Loading data from a JDBC source
val jdbcDF = spark.read.format("jdbc").option("url","jdbc:mysql://hadoop001:3306/rdd").option("dbtable"," rddtable").option("user","root").option("password","hive").load()
val connectionProperties = new Properties()
connectionProperties.put("user","root")
connectionProperties.put("password","hive")
val jdbcDF2 = spark.read
.jdbc("jdbc:mysql://hadoop001:3306/rdd","rddtable",connectionProperties)
// Saving data to a JDBC source
jdbcDF.write
.format("jdbc")
.option("url","jdbc:mysql://hadoop001:3306/rdd")
.option("dbtable","rddtable2")
.option("user","root")
.option("password","hive")
.save()
jdbcDF2.write
.jdbc("jdbc:mysql://hadoop001:3306/mysql","db",connectionProperties)
// Specifying create table column data types on write
jdbcDF.write
.option("createTableColumnTypes","name CHAR(64),comments VARCHAR(1024)")
.jdbc("jdbc:mysql://hadoop001:3306/mysql",connectionProperties)

以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支援我們。