Spark 外部數(shù)據(jù)集

外部數(shù)據(jù)集

Spark 可以從任何一個(gè) Hadoop 支持的存儲(chǔ)源創(chuàng)建分布式數(shù)據(jù)集，包括你的本地文件系統(tǒng)，HDFS，Cassandra，HBase，Amazon S3等。 Spark 支持文本文件(text files)，SequenceFiles 和其他 Hadoop InputFormat。

文本文件 RDDs 可以使用 SparkContext 的 textFile 方法創(chuàng)建。 在這個(gè)方法里傳入文件的 URI (機(jī)器上的本地路徑或 hdfs://，s3n:// 等)，然后它會(huì)將文件讀取成一個(gè)行集合。這里是一個(gè)調(diào)用例子：

scala> val distFile = sc.textFile("data.txt")
distFile: RDD[String] = MappedRDD@1d4cee08

一旦創(chuàng)建完成，distFile 就能做數(shù)據(jù)集操作。例如，我們可以用下面的方式使用 map 和 reduce 操作將所有行的長度相加：distFile.map(s => s.length).reduce((a, b) => a + b)。

注意，Spark 讀文件時(shí)：

• 如果使用本地文件系統(tǒng)路徑，文件必須能在 work 節(jié)點(diǎn)上用相同的路徑訪問到。要么復(fù)制文件到所有的 workers，要么使用網(wǎng)絡(luò)的方式共享文件系統(tǒng)。
• 所有 Spark 的基于文件的方法，包括 textFile，能很好地支持文件目錄，壓縮過的文件和通配符。例如，你可以使用 textFile("/my/文件目錄")，textFile("/my/文件目錄/*.txt") 和 textFile("/my/文件目錄/*.gz")。
• textFile 方法也可以選擇第二個(gè)可選參數(shù)來控制切片(slices)的數(shù)目。默認(rèn)情況下，Spark 為每一個(gè)文件塊(HDFS 默認(rèn)文件塊大小是 64M)創(chuàng)建一個(gè)切片(slice)。但是你也可以通過一個(gè)更大的值來設(shè)置一個(gè)更高的切片數(shù)目。注意，你不能設(shè)置一個(gè)小于文件塊數(shù)目的切片值。

除了文本文件，Spark 的 Scala API 支持其他幾種數(shù)據(jù)格式：

• SparkContext.sholeTextFiles 讓你讀取一個(gè)包含多個(gè)小文本文件的文件目錄并且返回每一個(gè)(filename, content)對(duì)。與 textFile 的差異是：它記錄的是每個(gè)文件中的每一行。
• 對(duì)于 SequenceFiles，可以使用 SparkContext 的 sequenceFile[K, V] 方法創(chuàng)建，K 和 V 分別對(duì)應(yīng)的是 key 和 values 的類型。像 IntWritable 與 Text 一樣，它們必須是 Hadoop 的 Writable 接口的子類。另外，對(duì)于幾種通用的 Writables，Spark 允許你指定原聲類型來替代。例如： sequenceFile[Int, String] 將會(huì)自動(dòng)讀取 IntWritables 和 Text。
• 對(duì)于其他的 Hadoop InputFormats，你可以使用 SparkContext.hadoopRDD 方法，它可以指定任意的 JobConf，輸入格式(InputFormat)，key 類型，values 類型。你可以跟設(shè)置 Hadoop job 一樣的方法設(shè)置輸入源。你還可以在新的 MapReduce 接口(org.apache.hadoop.mapreduce)基礎(chǔ)上使用 SparkContext.newAPIHadoopRDD(譯者注：老的接口是 SparkContext.newHadoopRDD)。
• RDD.saveAsObjectFile 和 SparkContext.objectFile 支持保存一個(gè)RDD，保存格式是一個(gè)簡單的 Java 對(duì)象序列化格式。這是一種效率不高的專有格式，如 Avro，它提供了簡單的方法來保存任何一個(gè) RDD。