RDD编程–创建RDD

概述

Spark 围绕弹性分布式数据集(RDD)的概念展开，RDD 是可以并行操作的元素的容错集合。创建 RDD 有两种方法：在驱动程序中并行化现有集合，或引用外部存储系统中的数据集，例如共享文件系统、HDFS、HBase 或任何提供 Hadoop InputFormat 的数据源。

并行集合

并行化集合是通过在驱动程序（Scala ）中的现有集合上调用SparkContext的parallelize方法来创建的Seq。集合的元素被复制以形成可以并行操作的分布式数据集。例如，这里是如何创建一个包含数字 1 到 5 的并行化集合：

val data = Array(1, 2, 3, 4, 5)
val distData = sc.parallelize(data)

一旦创建，分布式数据集 ( distData) 就可以并行操作。例如，我们可能会调用distData.reduce((a, b) => a + b)将数组的元素相加。我们稍后将描述分布式数据集上的操作。

并行集合的一个重要参数是将数据集切割成的分区数。Spark 将为集群的每个分区运行一个任务。通常，您需要为集群中的每个 CPU 分配 2-4 个分区。通常，Spark 会尝试根据您的集群自动设置分区数。但是，您也可以通过将其作为第二个参数传递给parallelize（例如sc.parallelize(data, 10)）来手动设置它。注意：代码中的某些地方使用术语切片（分区的同义词）来保持向后兼容性。

外部数据集

Spark 可以从 Hadoop 支持的任何存储源创建分布式数据集，包括本地文件系统、HDFS、Cassandra、HBase、Amazon S3等。Spark 支持文本文件、SequenceFiles和任何其他 Hadoop InputFormat。

可以使用SparkContext的textFile方法创建文本文件 RDD 。此方法需要一个URI的文件（本地路径的机器上，或一个hdfs://，s3a://等URI），并读取其作为行的集合。这是一个示例调用：

scala> val distFile = sc.textFile("data.txt")
distFile: org.apache.spark.rdd.RDD[String] = data.txt MapPartitionsRDD[10] at textFile at <console>:26

创建后，distFile可以通过数据集操作对其进行操作。例如，我们可以添加最多使用的所有行的大小map和reduce操作如下：distFile.map(s => s.length).reduce((a, b) => a + b)。

注意事项

使用 Spark 读取文件的一些注意事项：

如果使用本地文件系统上的路径，则该文件也必须可以在工作节点上的相同路径上访问。要么将文件复制到所有工作人员，要么使用网络安装的共享文件系统。
Spark的所有基于文件的输入方法，包括textFile，都支持在目录、压缩文件和通配符上运行。例如，你可以使用textFile("/my/directory")，textFile("/my/directory/.txt")和textFile("/my/directory/.gz")。当读取多个文件时，分区的顺序取决于文件从文件系统返回的顺序。例如，它可能会也可能不会遵循文件的字典顺序（按路径）。在分区内，元素根据它们在底层文件中的顺序进行排序。
该textFile方法还采用可选的第二个参数来控制文件的分区数。默认情况下，Spark 为文件的每个块创建一个分区（在 HDFS 中默认块为 128MB），但您也可以通过传递更大的值来请求更多的分区。请注意，分区数不能少于块数。

除了文本文件，Spark 的 Scala API 还支持其他几种数据格式：

SparkContext.wholeTextFiles允许您读取包含多个小文本文件的目录，并将每个文件作为（文件名、内容）对返回。这与
形成对比textFile，后者将在每个文件中每行返回一条记录。分区由数据局部性决定，在某些情况下，这可能导致分区太少。对于这些情况，wholeTextFiles提供可选的第二个参数来控制最小分区数。
对于SequenceFiles，使用 SparkContext 的sequenceFile[K,V]方法，其中K和V是文件中键和值的类型。这些应该是 Hadoop的Writable接口的子类，如IntWritable和Text。此外，Spark 允许您为一些常见的 Writable指定本机类型；例如，sequenceFile[Int, String]会自动读取 IntWritables 和 Texts。
对于其他 Hadoop InputFormats，您可以使用该SparkContext.hadoopRDD方法，该方法采用任意JobConf输入格式类、键类和值类。以与使用输入源的Hadoop 作业相同的方式设置这些。您还可以使用SparkContext.newAPIHadoopRDD基于“新”MapReduceAPI ( org.apache.hadoop.mapreduce) 的InputFormats 。
RDD.saveAsObjectFile并SparkContext.objectFile支持以由序列化的 Java 对象组成的简单格式保存
RDD。虽然这不如 Avro 这样的专用格式有效，但它提供了一种保存任何 RDD 的简单方法。