4.2 创建RDD
4.2 创建RDD
由于Spark一切都是基于RDD的，如何创建RDD就变得非常重要，除了可以直接从父RDD转换，还支持两种方式来创建RDD：
1）并行化一个程序中已经存在的集合（例如，数组）；
2）引用一个外部文件存储系统（HDFS、HBase、Tachyon或是任何一个支持Hadoop输入格式的数据源）中的数据集。
4.2.1 集合（数组）创建RDD
通过并行集合（数组）创建RDD，主要是调用SparkContext的parallelize方法，在Driver（驱动程序）中一个已经存在的集合（数组）上创建，SparkContext对象代表到Spark集群的连接，可以用来创建RDD、广播变量和累加器。可以复制集合的对象创建一个支持并行操作的分布式数据集（ParallelCollectionRDD）。一旦该RDD创建完成，分布数据集可以支持并行操作，比如在该集合上调用Reduce将数组的元素相加。
parallelize方法的定义如下：
def parallelize[T](seq: Seq[T], numSlices: Int = defaultParallelism): RDD[T]
其中，第一个参数为对象集合，第二个参数为设定的分片数，默认值为2，返回指定对象类型的RDD。
下面以Scala语言进行操作，展示如何从一个数组创建一个并行集合，并进行数组元素相加操作。
scala> val data = Array(1, 2, 3, 4, 5)
data: Array[Int] = Array(1, 2, 3, 4, 5)
scala> val distData = sc.parallelize(data)
distData: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0]…
scala> distData.reduce((a, b) => a + b)
res2: Int = 15
并行集合创建的一个重要参数是slices的数目，它指定了将数据集切分为几个分区。在集群模式中，Spark将会在每份slice上运行一个Task。当然，也可以通过parallelize方法的第二个参数进行手动设置（如sc.parallelize(data, 10)），可以为集群中的每个CPU分配2～4个slices（也就是每个CPU分配2～4个Task）。
4.2.2 存储创建RDD
Spark可以从本地文件创建，也可以由Hadoop支持的文件系统（HDFS、KFS、Amazon S3、Hypertable、HBase等），以及Hadoop支持的输入格式创建分布式数据集。
Spark支持textFile、SequenceFiles及任何Hadoop支持的输入格式。
1. 从各种分布式文件系统创建
RDD可以通过SparkContext的textFile（文本文件）方法创建，其定义如下：
def textFile(path: String, minPartitions: Int = defaultMinPartitions): RDD[String]
其中，第一个参数指定文件的URI地址（本地文件路径，或者hdfs://、sdn://、kfs://……），并且以“行”的集合形式读取，第二个参数和parallelize方法一样也是用来指定分片数。下面以Scala语言进行操作为例，展示如何从一个数组创建一个并行集合。
scala> val distFile = sc.textFile(“dfs://data.txt”)
distFile: org.apache.spark.rdd.RDD[String] =spark.HadoopRDD@1d4cee08
一旦创建了并行集合，distFile变量实质上转变成新的RDD，可以使用Map和Reduce操作将所有行数的长度相加：
distFile.map(s => s.length).reduce((a, b) => a + b).
注意
如果使用本地文件系统中的路径，那么该文件在工作节点必须可以被相同的路径访问。这可以通过将文件复制到所有的工作节点或使用网络挂载的共享文件系统实现。所有Spark基于的文件输入方法（包括textFile方法），都支持路径、压缩文件和通配符。可以使用textFile("/path")、textFile("/path/*.txt")和textFile("/path /*.gz")。
HDFS数据块大小为64的MB的倍数，Spark默认为每一个数据块创建一个分片。如果需要一个分片包含多个数据块，可以通过传入参数来指定更多的分片。
wholeTextFiles方法可以读取一个包含多个小的文本文件的目录，并通过键-值对（其中key为文件路径，value为文件内容）的方式返回每一个目录。而textFile函数为每个文件中的每一行返回一个记录。
2.从支持Hadoop输入格式数据源创建
对于其他类型的Hadoop输入格式，可以使用SparkContext.hadoopRDD方法来加载数据，也可以使用SparkContext.newHadoopRDD来处理这些基于新Mapreduce API的输入格式。RDD.saveAsObjectFile和SparkContext.objectFile支持以序列化的Java对象组成简单的格式来保存RDD，并提供了一个简单的方法来保存任何RDD。