1、動機
有時候資料量會大到本機可能無法儲存，這時就需要探索別的讀取和儲存方法了。
Spark支援很多種輸入源和輸出源。一部分原因是Spark本身是基於Hadoop生態圈二構建的，so spark可以通過Hadoop MapReduce 所使用的InputFormat 和 OutPutFormat 介面訪問資料，而大部分常見的檔案格式與儲存系統（S3,HDFS,Cassandra,HBase等）都支援這種介面。
Spark所支援的三種常見資料來源：

• 文字格式與檔案系統
• SparkSQL
• 資料庫與鍵值儲存

2、檔案格式：
結構化
- 文字檔案：不是結構化，普通的文字檔案，每一行一條記錄
- JSON： 半結構化，常見的基於文字的格式，半結構化；大多數庫都要求每行一條記錄。
- CSV：結構化，非常常見的基於文字的格式，通常在電子表格應用中使用。
- SequenceFile：結構化，一種用於鍵值對資料的常見Hadoop檔案格式。
- Protocol buffers：結構化，一種快速的，節約空間的跨語言格式。
- 物件檔案：用來將Spark作業中的資料儲存下來以讓共享的程式碼讀取。改變類的時候它會失效，因為他依賴於java序列化。

（1） 文字檔案
當我們將一個文字檔案讀取為RDD時，輸入的每一行都會成為RDD的每一個元素。也可以將多個文字檔案讀取為一個pair RDD，其中鍵是檔名，值是檔案內容。

scala> val input = sc.textFile("/Users/mac/Documents/javascala/VNCluster/500points.txt ")
input: org.apache.spark.rdd.RDD[String] = /Users/mac/Documents/javascala/VNCluster/500points.txt  MapPartitionsRDD[1] at textFile at <console>:24
 

同樣可以指定分割槽數

scala> val input = sc.textFile("/Users/mac/Documents/javascala/VNCluster/500points.txt ",4)
input: org.apache.spark.rdd.RDD[String] = /Users/mac/Documents/javascala/VNCluster/500points.txt  MapPartitionsRDD[3] at textFile at <console>:24

我是讀入一個數據點文字檔案，然後對資料做一些操作儲存在源目錄中：

scala> val data = input.map
 
(x => (x.split(" ")(0),x.split(" ")(1)))
data: org.apache.spark.rdd.RDD[(String, String)] = MapPartitionsRDD[6] at map at <console>:26

scala> data.first
res3: (String, String) = (0.2489,0.7091)

scala> data.filter{case(x,y) => y.toDouble > 0.5}
res5: org.apache.spark.rdd.RDD[(String, String)] = MapPartitionsRDD[10] at filter at <console>:29

scala> val afterDeal = data.filter{case(x,y) => y.toDouble > 0.5}
afterDeal: org.apache.spark.rdd.RDD[(String, String)] = MapPartitionsRDD[11] at filter at <console>:28

scala> afterDeal.first
res7: (String, String) = (0.2489,0.7091)

scala> afterDeal.saveAsTextFile("/Users/mac/Documents/javascala/VNCluster/rdd.txt")

（2）JSON

• 讀取JSON檔案：將資料作為文字檔案，然後對JSON資料進行解析，這樣的方法可以在所有支援的程式語言中使用。
• 儲存JSON檔案：不需要考慮格式錯誤的資料，並且也知道要寫出的資料型別。

（3）逗號分隔值與製表符分隔值
逗號分隔值（CSV）檔案每行都有固定數目的欄位，欄位間用逗號隔開。記錄通常是一行一條，有時也可以跨行。

• 讀取CSV：

 val conf = new SparkConf().setMaster("local").setAppName("My app")
    val sc = new SparkContext(conf)

    val input = sc.textFile("inputFile")
    val result = input.map{ line => 
      val reader = new CSVReader(new StringBuilder(line))
      reader.readNext()

• 儲存CSV：和儲存儲存文字檔案基本一樣

scala> val input = sc.textFile("/Users/mac/Desktop/500points.csv")
input: org.apache.spark.rdd.RDD[String] = /Users/mac/Desktop/500points.csv MapPartitionsRDD[1] at textFile at <console>:24

scala> input.first
res0: String = 237.09,712.1

scala> val data = input.map(x => (x.split(",")(0),x.split(",")(1)))
data: org.apache.spark.rdd.RDD[(String, String)] = MapPartitionsRDD[2] at map at <console>:26

scala> data.first
res1: (String, String) = (237.09,712.1)

scala> val deal = data.mapValues(x => x.toDouble / x.max.toDouble)
deal: org.apache.spark.rdd.RDD[(String, Double)] = MapPartitionsRDD[3] at mapValues at <console>:28

scala> deal.saveAsTextFile("/Users/mac/Desktop/500pointsresult.csv" )