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如下：

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

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|

// |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", "db", connectionProperties)

以上就是本文的全部內容，希望對大家的學習有所幫助，有喜歡小編的文章可以關注小編並幫小編轉發，小編後期會持續更新的丫。

---

分享到:

---

閱讀更多 架構師公社 的文章

關鍵字: 數據源 JSON Hadoop

---