一、簡介

1.1 多資料來源支援

Spark 支援以下六個核心資料來源，同時 Spark 社群還提供了多達上百種資料來源的讀取方式，能夠滿足絕大部分使用場景。

• CSV
• JSON
• Parquet
• ORC
• JDBC/ODBC connections
• Plain-text files

注：以下所有測試檔案均可從本倉庫的resources 目錄進行下載

1.2 讀資料格式

所有讀取 API 遵循以下呼叫格式：

// 格式
DataFrameReader.format(...).option("key", "value").schema(...).load()

// 示例
spark.read.format("csv")
.option("mode", "FAILFAST")          // 讀取模式
.option("inferSchema", "true")       // 是否自動推斷 schema
.option("path", "path/to/file(s)")   // 檔案路徑
.schema(someSchema)                  // 使用預定義的 schema      
.load()
 

讀取模式有以下三種可選項：

1.3 寫資料格式

// 格式
DataFrameWriter.format(...).option(...).partitionBy(...).bucketBy(...).sortBy(...).save()

//示例
dataframe.write.format("csv")
.option("mode", "OVERWRITE")         //寫模式
.option("dateFormat", "yyyy-MM-dd")  //日期格式
.option("path", "path/to/file(s)")
.save()
 

寫資料模式有以下四種可選項：

二、CSV

CSV 是一種常見的文字檔案格式，其中每一行表示一條記錄，記錄中的每個欄位用逗號分隔。

2.1 讀取CSV檔案

自動推斷型別讀取讀取示例：

spark.read.format("csv")
.option("header", "false")        // 檔案中的第一行是否為列的名稱
.option("mode", "FAILFAST")      // 是否快速失敗
.option("inferSchema", "true")   // 是否自動推斷 schema
.load("/usr/file/csv/dept.csv")
.show()
 

使用預定義型別：

import org.apache.spark.sql.types.{StructField, StructType, StringType,LongType}
//預定義資料格式
val myManualSchema = new StructType(Array(
    StructField("deptno", LongType, nullable = false),
    StructField("dname", StringType,nullable = true),
    StructField("loc", StringType,nullable = true)
))
spark.read.format("csv")
.option("mode", "FAILFAST")
.schema(myManualSchema)
.load("/usr/file/csv/dept.csv")
.show()

2.2 寫入CSV檔案

df.write.format("csv").mode("overwrite").save("/tmp/csv/dept2")

也可以指定具體的分隔符：

df.write.format("csv").mode("overwrite").option("sep", "\t").save("/tmp/csv/dept2")

2.3 可選配置

為節省主文篇幅，所有讀寫配置項見文末 9.1 小節。

三、JSON

3.1 讀取JSON檔案

spark.read.format("json").option("mode", "FAILFAST").load("/usr/file/json/dept.json").show(5)

需要注意的是：預設不支援一條資料記錄跨越多行 (如下)，可以通過配置 multiLine 為 true 來進行更改，其預設值為 false。

// 預設支援單行
{"DEPTNO": 10,"DNAME": "ACCOUNTING","LOC": "NEW YORK"}

//預設不支援多行
{
  "DEPTNO": 10,
  "DNAME": "ACCOUNTING",
  "LOC": "NEW YORK"
}

3.2 寫入JSON檔案

df.write.format("json").mode("overwrite").save("/tmp/spark/json/dept")

3.3 可選配置

為節省主文篇幅，所有讀寫配置項見文末 9.2 小節。

四、Parquet

Parquet 是一個開源的面向列的資料儲存，它提供了多種儲存優化，允許讀取單獨的列非整個檔案，這不僅節省了儲存空間而且提升了讀取效率，它是 Spark 是預設的檔案格式。

4.1 讀取Parquet檔案

spark.read.format("parquet").load("/usr/file/parquet/dept.parquet").show(5)

2.2 寫入Parquet檔案

df.write.format("parquet").mode("overwrite").save("/tmp/spark/parquet/dept")

2.3 可選配置

Parquet 檔案有著自己的儲存規則，因此其可選配置項比較少，常用的有如下兩個：

更多可選配置可以參閱官方文件：https://spark.apache.org/docs/latest/sql-data-sources-parquet.html

五、ORC

ORC 是一種自描述的、型別感知的列檔案格式，它針對大型資料的讀寫進行了優化，也是大資料中常用的檔案格式。

5.1 讀取ORC檔案

spark.read.format("orc").load("/usr/file/orc/dept.orc").show(5)

4.2 寫入ORC檔案

csvFile.write.format("orc").mode("overwrite").save("/tmp/spark/orc/dept")

六、SQL Databases

Spark 同樣支援與傳統的關係型資料庫進行資料讀寫。但是 Spark 程式預設是沒有提供資料庫驅動的，所以在使用前需要將對應的資料庫驅動上傳到安裝目錄下的 jars 目錄中。下面示例使用的是 Mysql 資料庫，使用前需要將對應的 mysql-connector-java-x.x.x.jar 上傳到 jars 目錄下。

6.1 讀取資料

讀取全表資料示例如下，這裡的 help_keyword 是 mysql 內建的字典表，只有 help_keyword_id 和 name 兩個欄位。

spark.read
.format("jdbc")
.option("driver", "com.mysql.jdbc.Driver")            //驅動
.option("url", "jdbc:mysql://127.0.0.1:3306/mysql")   //資料庫地址
.option("dbtable", "help_keyword")                    //表名
.option("user", "root").option("password","root").load().show(10)

從查詢結果讀取資料：

val pushDownQuery = """(SELECT * FROM help_keyword WHERE help_keyword_id <20) AS help_keywords"""
spark.read.format("jdbc")
.option("url", "jdbc:mysql://127.0.0.1:3306/mysql")
.option("driver", "com.mysql.jdbc.Driver")
.option("user", "root").option("password", "root")
.option("dbtable", pushDownQuery)
.load().show()

//輸出
+---------------+-----------+
|help_keyword_id|       name|
+---------------+-----------+
|              0|         <>|
|              1|     ACTION|
|              2|        ADD|
|              3|AES_DECRYPT|
|              4|AES_ENCRYPT|
|              5|      AFTER|
|              6|    AGAINST|
|              7|  AGGREGATE|
|              8|  ALGORITHM|
|              9|        ALL|
|             10|      ALTER|
|             11|    ANALYSE|
|             12|    ANALYZE|
|             13|        AND|
|             14|    ARCHIVE|
|             15|       AREA|
|             16|         AS|
|             17|   ASBINARY|
|             18|        ASC|
|             19|     ASTEXT|
+---------------+-----------+

也可以使用如下的寫法進行資料的過濾：

val props = new java.util.Properties
props.setProperty("driver", "com.mysql.jdbc.Driver")
props.setProperty("user", "root")
props.setProperty("password", "root")
val predicates = Array("help_keyword_id < 10  OR name = 'WHEN'")   //指定資料過濾條件
spark.read.jdbc("jdbc:mysql://127.0.0.1:3306/mysql", "help_keyword", predicates, props).show() 

//輸出：
+---------------+-----------+
|help_keyword_id|       name|
+---------------+-----------+
|              0|         <>|
|              1|     ACTION|
|              2|        ADD|
|              3|AES_DECRYPT|
|              4|AES_ENCRYPT|
|              5|      AFTER|
|              6|    AGAINST|
|              7|  AGGREGATE|
|              8|  ALGORITHM|
|              9|        ALL|
|            604|       WHEN|
+---------------+-----------+

可以使用 numPartitions 指定讀取資料的並行度：

option("numPartitions", 10)

在這裡，除了可以指定分割槽外，還可以設定上界和下界，任何小於下界的值都會被分配在第一個分割槽中，任何大於上界的值都會被分配在最後一個分割槽中。

val colName = "help_keyword_id"   //用於判斷上下界的列
val lowerBound = 300L    //下界
val upperBound = 500L    //上界
val numPartitions = 10   //分割槽綜述
val jdbcDf = spark.read.jdbc("jdbc:mysql://127.0.0.1:3306/mysql","help_keyword",
                             colName,lowerBound,upperBound,numPartitions,props)

想要驗證分割槽內容，可以使用 mapPartitionsWithIndex 這個運算元，程式碼如下：

jdbcDf.rdd.mapPartitionsWithIndex((index, iterator) => {
    val buffer = new ListBuffer[String]
    while (iterator.hasNext) {
        buffer.append(index + "分割槽:" + iterator.next())
    }
    buffer.toIterator
}).foreach(println)

執行結果如下：help_keyword 這張表只有 600 條左右的資料，本來資料應該均勻分佈在 10 個分割槽，但是 0 分割槽裡面卻有 319 條資料，這是因為設定了下限，所有小於 300 的資料都會被限制在第一個分割槽，即 0 分割槽。同理所有大於 500 的資料被分配在 9 分割槽，即最後一個分割槽。

6.2 寫入資料

val df = spark.read.format("json").load("/usr/file/json/emp.json")
df.write
.format("jdbc")
.option("url", "jdbc:mysql://127.0.0.1:3306/mysql")
.option("user", "root").option("password", "root")
.option("dbtable", "emp")
.save()

七、Text

Text 檔案在讀寫效能方面並沒有任何優勢，且不能表達明確的資料結構，所以其使用的比較少，讀寫操作如下：

7.1 讀取Text資料

spark.read.textFile("/usr/file/txt/dept.txt").show()

7.2 寫入Text資料

df.write.text("/tmp/spark/txt/dept")

八、資料讀寫高階特性

8.1 並行讀

多個 Executors 不能同時讀取同一個檔案，但它們可以同時讀取不同的檔案。這意味著當您從一個包含多個檔案的資料夾中讀取資料時，這些檔案中的每一個都將成為 DataFrame 中的一個分割槽，並由可用的 Executors 並行讀取。

8.2 並行寫

寫入的檔案或資料的數量取決於寫入資料時 DataFrame 擁有的分割槽數量。預設情況下，每個資料分割槽寫一個檔案。

8.3 分割槽寫入

分割槽和分桶這兩個概念和 Hive 中分割槽表和分桶表是一致的。都是將資料按照一定規則進行拆分儲存。需要注意的是 partitionBy 指定的分割槽和 RDD 中分割槽不是一個概念：這裡的分割槽表現為輸出目錄的子目錄，資料分別儲存在對應的子目錄中。

val df = spark.read.format("json").load("/usr/file/json/emp.json")
df.write.mode("overwrite").partitionBy("deptno").save("/tmp/spark/partitions")

輸出結果如下：可以看到輸出被按照部門編號分為三個子目錄，子目錄中才是對應的輸出檔案。

8.3 分桶寫入

分桶寫入就是將資料按照指定的列和桶數進行雜湊，目前分桶寫入只支援儲存為表，實際上這就是 Hive 的分桶表。

val numberBuckets = 10
val columnToBucketBy = "empno"
df.write.format("parquet").mode("overwrite")
.bucketBy(numberBuckets, columnToBucketBy).saveAsTable("bucketedFiles")

8.5 檔案大小管理

如果寫入產生小檔案數量過多，這時會產生大量的元資料開銷。Spark 和 HDFS 一樣，都不能很好的處理這個問題，這被稱為“small file problem”。同時資料檔案也不能過大，否則在查詢時會有不必要的效能開銷，因此要把檔案大小控制在一個合理的範圍內。

在上文我們已經介紹過可以通過分割槽數量來控制生成檔案的數量，從而間接控制檔案大小。Spark 2.2 引入了一種新的方法，以更自動化的方式控制檔案大小，這就是 maxRecordsPerFile 引數，它允許你通過控制寫入檔案的記錄數來控制檔案大小。

 // Spark 將確保檔案最多包含 5000 條記錄
df.write.option(“maxRecordsPerFile”, 5000)

九、可選配置附錄

9.1 CSV讀寫可選配置

9.2 JSON讀寫可選配置

9.3 資料庫讀寫可選配置

資料庫讀寫更多配置可以參閱官方文件：https://spark.apache.org/docs/latest/sql-data-sources-jdbc.html

參考資料

1. Matei Zaharia, Bill Chambers . Spark: The Definitive Guide[M] . 2018-02
2. https://spark.apache.org/docs/latest/sql-data-sources.html

更多大資料系列文章可以參見 GitHub 開源專案： 大資料入門指南

相關推薦

Spark 系列（十）—— Spark SQL 外部資料來源

一、簡介 1.1 多資料來源支援 Spark 支援以下六個核心資料來源，同時 Spark 社群還提供了多達上百種資料來源的讀取方式，能夠滿足絕大部分使用場景。 CSV JSON Parquet ORC JDBC/ODBC connections Plain-text files 注：以下所有測試檔案均可

Spark 系列（八）—— Spark SQL 之 DataFrame 和 Dataset

## 一、Spark SQL簡介 Spark SQL 是 Spark 中的一個子模組，主要用於操作結構化資料。它具有以下特點： + 能夠將 SQL 查詢與 Spark 程式無縫混合，允許您使用 SQL 或 DataFrame API 對結構化資料進行查詢； + 支援多種開發語言； + 支援

Spark 系列（九）—— Spark SQL 之 Structured API

一、建立DataFrame和Dataset 1.1 建立DataFrame Spark 中所有功能的入口點是 SparkSession，可以使用 SparkSession.builder() 建立。建立後應用程式就可以從現有 RDD，Hive 表或 Spark 資料來源建立 DataFrame。示例如下： v

Spark 系列（五）—— Spark 執行模式與作業提交

一、作業提交 1.1 spark-submit Spark 所有模式均使用 spark-submit 命令提交作業，其格式如下： ./bin/spark-submit \ --class <main-class> \ # 應用程式主入口類 --master <maste

Spark 系列（十一）—— Spark SQL 聚合函式 Aggregations

一、簡單聚合 1.1 資料準備 // 需要匯入 spark sql 內建的函式包 import org.apache.spark.sql.functions._ val spark = SparkSession.builder().appName("aggregations").mast

Spark原始碼系列（九）Spark SQL初體驗之解析過程詳解

首先宣告一下這個版本的程式碼是1.1的，之前講的都是1.0的。 Spark支援兩種模式，一種是在spark裡面直接寫sql，可以通過sql來查詢物件，類似.net的LINQ一樣，另外一種支援hive的HQL。不管是哪種方式，下面提到的步驟都會有，不同的是具體的執行過程。下面

Spark 系列（十四）—— Spark Streaming 基本操作

一、案例引入 這裡先引入一個基本的案例來演示流的建立：獲取指定埠上的資料並進行詞頻統計。專案依賴和程式碼實現如下： <dependency> <groupId>org.apache.spark</groupId> <artifactId>spar

Spark 系列（十五）—— Spark Streaming 整合 Flume

一、簡介 Apache Flume 是一個分散式，高可用的資料收集系統，可以從不同的資料來源收集資料，經過聚合後傳送到分散式計算框架或者儲存系統中。Spark Straming 提供了以下兩種方式用於 Flume 的整合。 二、推送式方法 在推送式方法 (Flume-style Push-based Appr

Spark 系列（十六）—— Spark Streaming 整合 Kafka

一、版本說明 Spark 針對 Kafka 的不同版本，提供了兩套整合方案：spark-streaming-kafka-0-8 和 spark-streaming-kafka-0-10，其主要區別如下： spark-streaming-kafka-0-8 spark-streaming-kafka-0

深入淺出Mybatis系列（十）---SQL執行流程分析（源碼篇）(轉)

factor demo 讀取配置 gist wrapper load 任性 wrap 深入淺出 轉載自：http://www.cnblogs.com/dongying/p/4142476.html 1. SqlSessionFactory 與 SqlSession. 　　通

Spark採坑系列（三）Spark操作Hive的坑

跟著教學試著用Idea程式設計，實現Spark查詢Hive中的表。結果上來就涼了。 搗鼓好久都不行，在網上查有說將hive-site.xml放到resource目錄就行，還有什麼hadoop針對windows使用者的許可權問題，結果都是扯淡。 其實問題還是處在程式碼上，直接附上程式碼了

Spark學習（柒）- Spark SQL擴充套件和總結

文章目錄 Spark SQL使用場景 Spark SQL載入資料 1) RDD DataFrame/Dataset 2) Local Cloud(HDFS/S3) DataFrame與SQL的對比

Spark學習（陸）- Spark操作外部資料來源

文章目錄 產生背景 概念 目標 操作Parquet檔案資料 操作Hive表資料 操作MySQL表資料 操作MySQL的資料方法一: 操作MySQL的資料方法二: 操作MySQL

大資料乾貨系列（六）--Spark總結

Spark總結一、本質 Spark是一個分散式的計算框架，是下一代的MapReduce，擴充套件了MR的資料處理流程二、mapreduce有什麼問題1.排程慢，啟動map、reduce太耗時2.計算慢，每一步都要儲存中間結果落磁碟3.API抽象簡單，只有map和reduce兩

Spark 系列（三）—— 彈性式資料集RDDs

一、RDD簡介 RDD 全稱為 Resilient Distributed Datasets，是 Spark 最基本的資料抽象，它是隻讀的、分割槽記錄的集合，支援並行操作，可以由外部資料集或其他 RDD 轉換而來，它具有以下特性： 一個 RDD 由一個或者多個分割槽（Partitions）組成。對於 RDD

Spark 系列（四）—— RDD常用運算元詳解

一、Transformation spark 常用的 Transformation 運算元如下表： Transformation 運算元 Meaning（含義） map(func) 對原 RDD 中每個元素運用 func 函式，並生成新的 RDD filter(func) 對原 RDD 中每

Spark 系列（七）—— 基於 ZooKeeper 搭建 Spark 高可用叢集

一、叢集規劃 這裡搭建一個 3 節點的 Spark 叢集，其中三臺主機上均部署 Worker 服務。同時為了保證高可用，除了在 hadoop001 上部署主 Master 服務外，還在 hadoop002 和 hadoop003 上分別部署備用的 Master 服務，Master 服務由 Zookeeper

HBase 系列（十）—— HBase 的 SQL 中間層 Phoenix

一、Phoenix簡介 Phoenix 是 HBase 的開源 SQL 中間層，它允許你使用標準 JDBC 的方式來操作 HBase 上的資料。在 Phoenix 之前，如果你要訪問 HBase，只能呼叫它的 Java API，但相比於使用一行 SQL 就能實現資料查詢，HBase 的 API 還是過於複雜。

Flask系列（十）自定義Form組件

pri 擴展 append BE sam cep default .get pos 一、wtforms源碼流程 1、實例化流程分析 # 源碼流程 1. 執行type的 __call__ 方法，讀取字段到靜態字段 cls._unbound_fields 中； me

Java Thread系列（十）Future 模式

業務 throw 而且 代碼 ack void rri 是不是 決定 Java Thread系列（十）Future 模式 Future 模式適合在處理很耗時的業務邏輯時進行使用，可以有效的減少系統的響應時間，提高系統的吞吐量。 一、Future 模式核心思想 如下的請求