Spark SQL中常見4種數(shù)據(jù)源是什么

144次閱讀

共計 10170 個字符，預計需要花費 26 分鐘才能閱讀完成。

這篇文章主要介紹 Spark SQL 中常見 4 種數(shù)據(jù)源是什么，文中介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們一定要看完！

通用 load/write 方法

手動指定選項

Spark SQL 的 DataFrame 接口支持多種數(shù)據(jù)源的操作。一個 DataFrame 可以進行 RDDs 方式的操作，也可以被注冊為臨時表。把 DataFrame 注冊為臨時表之后，就可以對該 DataFrame 執(zhí)行 SQL 查詢。

Spark SQL 的默認數(shù)據(jù)源為 Parquet 格式。數(shù)據(jù)源為 Parquet 文件時，Spark SQL 可以方便的執(zhí)行所有的操作。

修改配置項 spark.sql.sources.default，可修改默認數(shù)據(jù)源格式。

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)

當數(shù)據(jù)源格式不是 parquet 格式文件時，需要手動指定數(shù)據(jù)源的格式。數(shù)據(jù)源格式需要指定全名（例如：org.apache.spark.sql.parquet），如果數(shù)據(jù)源格式為內置格式，則只需要指定簡稱 json, parquet, jdbc, orc, libsvm, csv, text 來指定數(shù)據(jù)的格式。

可以通過 SparkSession 提供的 read.load 方法用于通用加載數(shù)據(jù)，使用 write 和 save 保存數(shù)據(jù)。

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 執(zhí)行存儲操作，SaveMode 定義了對數(shù)據(jù)的處理模式。需要注意的是，這些保存模式不使用任何鎖定，不是原子操作。此外，當使用 Overwrite 方式執(zhí)行時，在輸出新數(shù)據(jù)之前原數(shù)據(jù)就已經被刪除。SaveMode 詳細介紹如下表：

Scala/JavaAny LanguageMeaningSaveMode.ErrorIfExists(default)“error”(default)如果文件存在，則報錯 SaveMode.Append“append”追加 SaveMode.Overwrite“overwrite”覆寫 SaveMode.Ignore“ignore”數(shù)據(jù)存在，則忽略

Parquet 文件

Parquet 讀寫

Parquet 格式經常在 Hadoop 生態(tài)圈中被使用，它也支持 Spark SQL 的全部數(shù)據(jù)類型。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|
// +------------+

解析分區(qū)信息

對表進行分區(qū)是對數(shù)據(jù)進行優(yōu)化的方式之一。在分區(qū)的表內，數(shù)據(jù)通過分區(qū)列將數(shù)據(jù)存儲在不同的目錄下。Parquet 數(shù)據(jù)源現(xiàn)在能夠自動發(fā)現(xiàn)并解析分區(qū)信息。例如，對人口數(shù)據(jù)進行分區(qū)存儲，分區(qū)列為 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 將自動解析分區(qū)信息。

返回的 DataFrame 的 Schema 如下：

root
|-- name: string (nullable = true)
|-- age: long (nullable = true)
|-- gender: string (nullable = true)
|-- country: string (nullable = true)

需要注意的是，數(shù)據(jù)的分區(qū)列的數(shù)據(jù)類型是自動解析的。當前，支持數(shù)值類型和字符串類型。自動解析分區(qū)類型的參數(shù)為：

spark.sql.sources.partitionColumnTypeInference.enabled

默認值為 true。

如果想關閉該功能，直接將該參數(shù)設置為 disabled。此時，分區(qū)列數(shù)據(jù)格式將被默認設置為 string 類型，不再進行類型解析。

Schema 合并

像 ProtocolBuffer、Avro 和 Thrift 那樣，Parquet 也支持 Schema evolution（Schema 演變）。用戶可以先定義一個簡單的 Schema，然后逐漸的向 Schema 中增加列描述。通過這種方式，用戶可以獲取多個有不同 Schema 但相互兼容的 Parquet 文件?，F(xiàn)在 Parquet 數(shù)據(jù)源能自動檢測這種情況，并合并這些文件的 schemas。

因為 Schema 合并是一個高消耗的操作，在大多數(shù)情況下并不需要，所以 Spark SQL 從 1.5.0 開始默認關閉了該功能?？梢酝ㄟ^下面兩種方式開啟該功能：

當數(shù)據(jù)源為 Parquet 文件時，將數(shù)據(jù)源選項 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 數(shù)據(jù)源

Apache Hive 是 Hadoop 上的 SQL 引擎，Spark SQL 編譯時可以包含 Hive 支持，也可以不包含。包含 Hive 支持的 Spark SQL 可以支持 Hive 表訪問、UDF(用戶自定義函數(shù))以及 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 會在當前的工作目錄中創(chuàng)建出自己的 Hive 元數(shù)據(jù)倉庫，叫作 metastore_db。此外，如果你嘗試使用 HiveQL 中的 CREATE TABLE (并非 CREATE EXTERNAL TABLE)語句來創(chuàng)建表，這些表會被放在你默認的文件系統(tǒng)中的 /user/hive/warehouse 目錄中(如果你的 classpath 中有配好的 hdfs-site.xml，默認的文件系統(tǒng)就是 HDFS，否則就是本地文件系統(tǒng))。

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|
// |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 用于指定數(shù)據(jù)倉庫的地址，如果你需要是用 HDFS 作為路徑，那么需要將 core-site.xml 和 hdfs-site.xml 加入到 Spark conf 目錄，否則只會創(chuàng)建 master 節(jié)點上的 warehouse 目錄，查詢時會出現(xiàn)文件找不到的問題，這是需要向使用 HDFS，則需要將 metastore 刪除，重啟集群。

外部 Hive 應用

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

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

b 打開 spark shell，注意帶上訪問 Hive 元數(shù)據(jù)庫的 JDBC 客戶端。

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

JSON 數(shù)據(jù)集

Spark SQL 能夠自動推測 JSON 數(shù)據(jù)集的結構，并將它加載為一個 Dataset[Row]. 可以通過 SparkSession.read.json()去加載一個 Dataset[String]或者一個 JSON 文件. 注意，這個 JSON 文件不是一個傳統(tǒng)的 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 從關系型數(shù)據(jù)庫中讀取數(shù)據(jù)的方式創(chuàng)建 DataFrame，通過對 DataFrame 一系列的計算后，還可以將數(shù)據(jù)再寫回關系型數(shù)據(jù)庫中。

注意，需要將相關的數(shù)據(jù)庫驅動放到 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 ,  db , connectionProperties)

以上是“Spark SQL 中常見 4 種數(shù)據(jù)源是什么”這篇文章的所有內容，感謝各位的閱讀！希望分享的內容對大家有幫助，更多相關知識，歡迎關注丸趣 TV 行業(yè)資訊頻道！

正文完