Spark-RDD

刺骨的言语ヽ痛彻心扉 2022-05-17 06:50 390阅读 0赞

# Resilient Distributed Dataset（RDD：弹性分布式数据集） #

## 概览： ##

在较高的层次上，每个Spark应用程序都包含一个驱动程序，该程序运行用户的主要功能并在集群上执行各种并行操作。 Spark提供的**主要抽象**是**弹性分布式数据集（****RDD****）**，它是跨群集节点分区的元素集合，可以并行操作。 RDD是通过从Hadoop文件系统（或任何其他Hadoop支持的文件系统）中的文件或驱动程序中的现有Scala集合开始并对其进行转换来创建的。用户还可以要求Spark在内存中保留RDD，允许它在并行操作中有效地重用。最后，RDD会自动从节点故障中恢复。

<table> 
 <tbody> 
  <tr> 
   <td style="background-color:#FFFFFF;border-color:#000000;vertical-align:top;"> 
    <table> 
     <tbody> 
      <tr> 
       <td> <p style="margin-left:0cm;">groupId= org.apache.hadoop</p> <p style="margin-left:0cm;">artifactId = hadoop-client</p> <p style="margin-left:0cm;">version = &lt;your-hdfs-version&gt;</p> </td> 
      </tr> 
     </tbody> 
    </table></td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <td style="background-color:#FFFFFF;border-color:#000000;vertical-align:top;"> 
    <table> 
     <tbody> 
      <tr> 
       <td> <p style="margin-left:0cm;">groupId = org.apache.spark</p> <p style="margin-left:0cm;">artifactId = spark-core_2.11</p> <p style="margin-left:0cm;">version = 2.3.1</p> </td> 
      </tr> 
     </tbody> 
    </table></td> 
  </tr> 
 </tbody> 
</table>

Spark中的第二个抽象是**可以在并行操作中使用的共享变量**。默认情况下，当Spark并行运行一个函数作为不同节点上的一组任务时，它会将函数中使用的每个变量的副本发送给每个任务。有时，变量需要跨任务共享，或者在任务和驱动程序之间共享。 Spark支持两种类型的共享变量：广播变量，可用于缓存所有节点的内存中的值;累加器，它们是仅“添加”到的变量，例如计数器和总和。

## 连接spark ##

添加依赖：

此外，如果您希望访问HDFS群集，则需要为您的HDFS版本添加对hadoop-client的依赖性。

import org.apache.spark.api.java.JavaSparkContext;

import org.apache.spark.api.java.JavaRDD;

import org.apache.spark.SparkConf;

## 初始化Spark ##

**SparkConf conf = new SparkConf().setAppName(appName).setMaster(master);**

**JavaSparkContext sc = new JavaSparkContext(conf);**

appName参数是应用程序在集群UI上显示的名称。 master是Spark，Mesos或YARN群集URL，或者是以本地模式运行的特殊“本地”字符串。 实际上，当在群集上运行时，您不希望在程序中对master进行硬编码，而是使用spark-submit启动应用程序并在那里接收它。 但是，对于本地测试和单元测试，您可以传递“local”来运行Spark in-process。

## RDDs ##

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

### Parallelized Collections ###

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

**List<Integer> data = Arrays.asList(1, 2, 3, 4, 5);**

**JavaRDD<Integer> distData = sc.parallelize(data);**

一旦创建，分布式数据集（distData）可以并行操作。 例如，我们可以调用distData.reduce（（a，b） - > a + b）来添加列表的元素。 我们稍后将描述对分布式数据集的操作。

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

### External Datasets ###

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

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

**JavaRDD****<****String****>** **distFile** **=** **sc****.****textFile****(****"data.txt"****);**

创建后，distFile可以由数据集操作执行。例如，我们可以使用map添加所有行的大小，并按如下方式减少操作：distFile.map（s - > s.length（））。reduce（（a，b） - > a + b）。

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

1.  如果在本地文件系统上使用路径，则还必须可以在工作节点上的相同路径上访问该文件。将文件复制到所有工作者或使用网络安装的共享文件系统。

1.  Spark的所有基于文件的输入方法（包括textFile）都支持在目录，压缩文件和通配符上运行。例如，您可以使用textFile（“/ my / directory”），textFile（“/ my / directory / \* .txt”）和textFile（“/ my / directory / \* .gz”）。

1.  textFile方法还采用可选的第二个参数来控制文件的分区数。默认情况下，Spark为文件的每个块创建一个分区（HDFS中默认为128MB），但您也可以通过传递更大的值来请求更多的分区。请注意，您不能拥有比块少的分区。

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

1.  JavaSparkContext.wholeTextFiles允许您读取包含多个小文本文件的目录，并将每个文件作为（文件名，内容）对返回。这与textFile形成对比，textFile将在每个文件中每行返回一条记录。

1.  对于SequenceFiles，使用SparkContext的sequenceFile \[K，V\]方法，其中K和V是文件中键和值的类型。这些应该是Hadoop的Writable接口的子类，如IntWritable和Text。

1.  对于其他Hadoop InputFormats，您可以使用JavaSparkContext.hadoopRDD方法，该方法采用任意JobConf和输入格式类，键类和值类。设置这些与使用输入源的Hadoop作业相同。您还可以使用基于“新”MapReduce API（org.apache.hadoop.mapreduce）的InputFormats的JavaSparkContext.newAPIHadoopRDD。

1.  JavaRDD.saveAsObjectFile和JavaSparkContext.objectFile支持以包含序列化Java对象的简单格式保存RDD。虽然这不像Avro这样的专用格式有效，但它提供了一种保存任何RDD的简便方法。

### RDD Oerations ###

RDD支持两种类型的操作：**转换**（从现有数据集**创建新**数据集）和**操作**（在数据集上运行计算后**将值返回到驱动程序**）。例如，map是一个转换，它通过一个函数传递每个数据集元素，并返回一个表示结果的新RDD。另一方面，reduce是一个使用某个函数聚合RDD的所有元素的操作，并将最终结果返回给驱动程序（尽管还有一个返回分布式数据集的并行reduceByKey）。

Spark中的所有转换都是懒惰的，因为它们不会立即计算结果。相反，他们只记得应用于某些基础数据集（例如文件）的转换。**仅当操作需要将结果返回到驱动程序时才会计算转换**。这种设计使Spark能够更有效地运行。例如，我们可以意识到通过map创建的数据集将用于reduce，并仅将reduce的结果返回给驱动程序，而不是更大的映射数据集。

默认情况下，每次对其执行操作时，都可以重新计算每个转换后的RDD。但是，您也可以使用持久化（或缓存）方法在内存中保留RDD，在这种情况下，Spark会在群集上保留元素，以便在下次查询时更快地访问。还支持在磁盘上保留RDD，或在多个节点之间复制。

**JavaRDD<String> lines = sc.textFile("data.txt");**

**JavaRDD<Integer> lineLengths = lines.map(s -> s.length());**

**int totalLength = lineLengths.reduce((a, b) -> a + b);**

第一行从外部文件定义基础RDD。 此数据集未加载到内存中或以其他方式操作：行仅仅是指向文件的指针。 第二行将lineLengths定义为地图转换的结果。 同样，由于懒惰，lineLengths不会立即计算。 最后，我们运行reduce，这是一个动作。 此时，Spark将计算分解为在不同机器上运行的任务，并且每台机器都运行其部分映射和本地缩减，仅返回其对驱动程序的答案。

如果我们以后想再次使用lineLengths，我们可以添加：

**lineLengths****.****persist****(****StorageLevel****.****MEMORY\_ONLY****());**

Passing Functions to Spark

Spark的API在很大程度上依赖于在驱动程序中传递函数以在集群上运行。 在Java中，函数由实现**org.apache.spark.api.java.function**包中的接口的类表示。 有两种方法可以创建这样的函数：

1.  在您自己的类中实现Function接口，可以是匿名内部类，也可以是命名类，并将其实例传递给Spark。
2.  使用lambda表达式简明地定义实现。

虽然本指南的大部分内容使用lambda语法来简洁，但很容易以长格式使用所有相同的API。 例如，我们可以按如下方式编写上面的代码：

**JavaRDD****<****String****>** **lines** **=** **sc****.****textFile****(****"data.txt"****);**

**JavaRDD****<****Integer****>** **lineLengths** **=** **lines****.****map****(****new** **Function****<****String****,** **Integer****>()****\{**

** ** **public** **Integer** **call****(****String s****)****\{** **return** **s****.****length****();****\}**

**\});**

**int** **totalLength** **=** **lineLengths****.****reduce****(****new** **Function2****<****Integer****,** **Integer****,** **Integer****>()****\{**

** ** **public** **Integer** **call****(****Integer a****,** **Integer b****)****\{** **return** **a** **+** **b****;****\}**

**\});**

或者，如果内联函数编写不实用：

**class GetLength implements Function<String, Integer> \{**

**  public Integer call(String s) \{ return s.length(); \}**

**\}**

**class Sum implements Function2<Integer, Integer, Integer> \{**

**  public Integer call(Integer a, Integer b) \{ return a + b; \}**

**\}**

**JavaRDD<String> lines = sc.textFile("data.txt");**

**JavaRDD<Integer> lineLengths = lines.map(new GetLength());**

**int totalLength = lineLengths.reduce(new Sum());**

请注意，Java中的匿名内部类也可以访问封闭范围中的变量，只要它们标记为final即可。 Spark会将这些变量的副本发送到每个工作节点，就像其他语言一样。

Understanding closures（了解闭包）

Spark的一个难点是在跨集群执行代码时理解变量和方法的范围和生命周期。 修改其范围之外的变量的RDD操作可能经常引起混淆。 在下面的示例中，我们将查看使用foreach（）递增计数器的代码，但其他操作也可能出现类似问题。

例如：

考虑下面的天真RDD元素总和，根据执行是否在同一JVM中发生，它可能表现不同。 一个常见的例子是在本地模式下运行Spark（--master = local \[n\]）而不是将Spark应用程序部署到集群（例如通过spark-submit to YARN）：

i**nt counter = 0;**

**JavaRDD<Integer> rdd = sc.parallelize(data);**

**// Wrong: Don't do this!!**

**rdd.foreach(x -> counter += x);**

**println("Counter value: " + counter);**

local and cluster model

上述代码的行为未定义，可能无法按预期工作。为了执行作业，Spark将RDD操作的处理分解为任务，每个任务由执行程序执行。在执行之前，Spark计算任务的关闭。闭包是那些变量和方法，它们必须是可见的，以便执行程序在RDD上执行其计算（在本例中为foreach（））。该闭包被序列化并发送给每个执行者。

发送给每个执行程序的闭包内的变量现在是副本，因此，当在foreach函数中引用计数器时，它不再是驱动程序节点上的计数器。驱动程序节点的内存中仍然有一个计数器，但执行程序不再可见！执行程序只能看到序列化闭包中的副本。因此，计数器的最终值仍然为零，因为计数器上的所有操作都引用了序列化闭包内的值。

在本地模式下，在某些情况下，foreach函数实际上将在与驱动程序相同的JVM中执行，并将引用相同的原始计数器，并且可能实际更新它。

为了确保在这些场景中定义良好的行为，应该使用累加器。 Spark中的累加器专门用于提供一种机制，用于在跨集群中的工作节点拆分执行时安全地更新变量。本指南的“累加器”部分更详细地讨论了这些内容。

通常，闭包 - 类似循环或本地定义的方法的构造不应该用于改变某些全局状态。 Spark没有定义或保证从闭包外部引用的对象的突变行为。执行此操作的某些代码可能在本地模式下工作，但这只是偶然的，并且此类代码在分布式模式下不会按预期运行。如果需要某些全局聚合，请使用累加器。

Printing elements of an RDD

另一个常见的习惯用法是尝试使用rdd.foreach（println）或rdd.map（println）打印出RDD的元素。 在一台机器上，这将生成预期的输出并打印所有RDD的元素。 但是，在集群模式下，执行程序调用的stdout输出现在写入执行程序的stdout，而不是驱动程序上的那个，因此驱动程序上的stdout不会显示这些！ 要打印驱动程序上的所有元素，可以使用collect（）方法首先将RDD带到驱动程序节点：**rdd.collect****（）.foreach****（println****）**。 但是，这会导致驱动程序内存不足，因为collect（）会将整个RDD提取到一台机器上; 如果你只需要打印RDD的一些元素，更安全的方法是使用**take****（）：rdd.take****（100****）.foreach****（println****）**。

Working with Key-Value Pairs

虽然大多数Spark操作都适用于包含任何类型对象的RDD，但一些特殊操作仅适用于键值对的RDD。 最常见的是分布式“shuffle”操作，例如通过密钥对元素进行分组或聚合。

在Java中，键值对使用Scala标准库中的scala.Tuple2类表示。 您可以简单地调用**new Tuple2****（a****，b****）**来创建元组，稍后使用**tuple.\_1****（）**和tuple.\_2（）访问其字段。

键值对的RDD由JavaPairRDD类表示。 您可以使用特殊版本的map操作从JavaRDD构建JavaPairRDD，例如mapToPair和flatMapToPair。 JavaPairRDD将具有标准RDD功能和特殊键值功能。

例如，以下代码对键值对使用reduceByKey操作来计算文件中每行文本出现的次数：

**JavaRDD****<****String****>** **lines** **=** **sc****.****textFile****(****"data.txt"****);**

**JavaPairRDD****<****String****,****Integer****>** **pairs** **=** **lines****.****mapToPair****(****s** **->****new****Tuple2****(****s****,****1****));**

**JavaPairRDD****<****String****,****Integer****>** **counts** **=** **pairs****.****reduceByKey****((****a****,** **b****)****->** **a** **+** **b****);**

我们也可以使用**counts.sortByKey****（）**来按字母顺序对这些对进行排序，最后使用**counts.collect****（）**将它们作为对象数组返回到驱动程序。

注意：在键值对操作中使用自定义对象作为键时，必须确保自定义equals（）方法伴随着匹配的hashCode（）方法。 有关完整的详细信息，请参阅Object.hashCode（）文档中概述的合同。

Transformations

**map**（func）返回通过函数func传递源的每个元素形成的新分布式数据集。

**filter**（func）返回通过选择func返回true的源元素形成的新数据集。

**flatMap**（func）与map类似，但每个输入项可以映射到0个或更多输出项（因此func应返回Seq而不是单个项）。

**mapPartitions**（func）与map类似，但在RDD的每个分区（块）上单独运行，因此当在类型T的RDD上运行时，func必须是Iterator <T> => Iterator <U>类型。

**mapPartitionsWithIndex**（func）与mapPartitions类似，但也为func提供了一个表示分区索引的整数值，因此当在类型的RDD上运行时，func必须是类型（Int，Iterator <T>）=> Iterator <U> T.

**sample**（withReplacement，fraction，seed）使用给定的随机数生成器种子，使用或不使用替换，对数据的一小部分进行采样。

**union**（otherDataset）返回一个新数据集，其中包含源数据集和参数中元素的并集。

**intersection**（otherDataset）返回一个新的RDD，它包含源数据集和参数中的元素的交集。

**distinct**（\[numPartitions\]））返回包含源数据集的不同元素的新数据集。

**groupByKey**（\[numPartitions\]）在（K，V）对的数据集上调用时，返回（K，Iterable <V>）对的数据集。

注意：如果要对每个键执行聚合（例如总和或平均值）进行分组，则使用reduceByKey或aggregateByKey将产生更好的性能。

注意：默认情况下，输出中的并行级别取决于父RDD的分区数。您可以传递可选的numPartitions参数来设置不同数量的任务。

**reduceByKey**（func，\[numPartitions\]）当调用（K，V）对的数据集时，返回（K，V）对的数据集，其中使用给定的reduce函数func聚合每个键的值，该函数必须是type（V，V）=> V.与groupByKey类似，reduce任务的数量可通过可选的第二个参数进行配置。

**aggregateByKey**（zeroValue）（seqOp，combOp，\[numPartitions\]）在（K，V）对的数据集上调用时，返回（K，U）对的数据集，其中使用给定的组合函数聚合每个键的值，中性的“零”值。允许与输入值类型不同的聚合值类型，同时避免不必要的分配。与groupByKey类似，reduce任务的数量可通过可选的第二个参数进行配置。

**sortByKey****（\[ascending\]****，\[numPartitions\]****）**在K实现Ordered的（K，V）对的数据集上调用时，返回按键按升序或降序排序的（K，V）对的数据集，如布尔升序参数。

**join****（otherDataset****，\[numPartitions\]****）**当调用类型为（K，V）和（K，W）的数据集时，返回（K，（V，W））对的数据集以及每个键的所有元素对。通过leftOuterJoin，rightOuterJoin和fullOuterJoin支持外连接。

**cogroup****（otherDataset****，\[numPartitions\]****）**当调用类型为（K，V）和（K，W）的数据集时，返回（K，（Iterable <V>，Iterable <W>））元组的数据集。此操作也称为groupWith。

**cartesian**（otherDataset）当调用类型为T和U的数据集时，返回（T，U）对的数据集（所有元素对）。

**pipe****（command****，\[envVars\]****）**通过shell命令管道RDD的每个分区，例如：一个Perl或bash脚本。 RDD元素被写入进程的stdin，并且输出到其stdout的行将作为字符串的RDD返回。

**coalesce**（numPartitions）将RDD中的分区数减少为numPartitions。过滤大型数据集后，可以更有效地运行操作。

**repartition**（numPartitions）随机重新调整RDD中的数据以创建更多或更少的分区并在它们之间进行平衡。这总是随机播放网络上的所有数据。

**repartitionAndSortWithinPartitions**（partitioner）根据给定的分区重新分区RDD，并在每个生成的分区中按键对记录进行排序。这比调用重新分区然后在每个分区内排序更有效，因为它可以将排序推送到shuffle机器中。

Actions

**reduce**（func）使用函数func（它接受两个参数并返回一个）来聚合数据集的元素。该函数应该是可交换的和关联的，以便可以并行正确计算。

**collect**（）在驱动程序中将数据集的所有元素作为数组返回。在过滤器或其他返回足够小的数据子集的操作之后，这通常很有用。

**count**（）返回数据集中的元素数。

**first**（）返回数据集的第一个元素（类似于take（1））。

**take****（n****）**返回包含数据集的前n个元素的数组。

**takeSample****（withReplacement****，num****，\[seed\]****）**返回一个数组，其中包含数据集的num个元素的随机样本，有或没有替换，可选择预先指定随机数生成器种子。

**takeOrdered****（n****，\[ordering\]****）**使用自然顺序或自定义比较器返回RDD的前n个元素。

**saveAsTextFile****（path**）将数据集的元素作为文本文件（或文本文件集）写入本地文件系统，HDFS或任何其他Hadoop支持的文件系统的给定目录中。 Spark将在每个元素上调用toString，将其转换为文件中的一行文本。

**saveAsSequenceFile****（路径）**

（Java和Scala）将数据集的元素作为Hadoop SequenceFile写入本地文件系统，HDFS或任何其他Hadoop支持的文件系统中的给定路径中。这可以在实现Hadoop的Writable接口的键值对的RDD上使用。在Scala中，它也可以在可隐式转换为Writable的类型上使用（Spark包括基本类型的转换，如Int，Double，String等）。

**saveAsObjectFile****（路径）**

（Java和Scala）使用Java序列化以简单格式编写数据集的元素，然后可以使用SparkContext.objectFile（）加载它。

**countByKey**（）仅适用于类型为（K，V）的RDD。返回（K，Int）对的散列映射以及每个键的计数。

**foreach**（func）对数据集的每个元素运行函数func。这通常用于副作用，例如更新累加器或与外部存储系统交互。

注意：在foreach（）之外修改除累加器之外的变量可能会导致未定义的行为。有关详细信息，请参阅了解闭包。

Spark RDD API还公开了某些操作的异步版本，例如foreach的foreachAsync，它会立即将一个FutureAction返回给调用者，而不是在完成操作时阻塞。 这可用于管理或等待操作的异步执行。

Shuffle operations

Spark中的某些操作会触发称为shuffle的事件。 随机播放是Spark的重新分配数据的机制，因此它可以跨分区进行不同的分组。 这通常涉及跨执行程序和机器复制数据，使得混洗成为复杂且昂贵的操作。

Background

为了理解在shuffle期间发生的事情，我们可以考虑reduceByKey操作的示例。 reduceByKey操作生成一个新的RDD，其中单个键的所有值都组合成一个元组 - 键和对与该键关联的所有值执行reduce函数的结果。挑战在于，并非单个密钥的所有值都必须位于同一个分区，甚至是同一个机器上，但它们必须位于同一位置才能计算结果。

在Spark中，数据通常不跨分区分布，以便在特定操作的必要位置。在计算过程中，单个任务将在单个分区上运行 - 因此，要组织单个reduceByKey reduce任务执行的所有数据，Spark需要执行全部操作。它必须从所有分区读取以查找所有键的所有值，然后将各个值组合在一起以计算每个键的最终结果 - 这称为shuffle。

尽管新洗牌数据的每个分区中的元素集将是确定性的，并且分区本身的排序也是如此，但这些元素的排序不是。如果在随机播放后需要可预测的有序数据，则可以使用：

*  mapPartitions to sort each partition using, for example, .sorted
 *  repartitionAndSortWithinPartitions to efficiently sort partitions while simultaneously repartitioning
 *  sortBy to make a globally ordered RDD

可以导致混洗的操作包括重新分区操作，例如重新分区和合并，“ByKey操作（计数除外）”，如groupByKey和reduceByKey，以及联合操作，如cogroup和join。

Performance Impact

Shuffle是一项昂贵的操作，因为它涉及磁盘I / O，数据序列化和网络I / O.为了组织shuffle的数据，Spark生成多组任务 - 映射任务以组织数据，以及一组reduce任务来聚合它。这个术语来自MapReduce，并不直接与Spark的地图和减少操作有关。

在内部，各个地图任务的结果会保留在内存中，直到它们无法适应。然后，这些基于目标分区进行排序并写入单个文件。在reduce方面，任务读取相关的排序块。

某些shuffle操作会消耗大量的堆内存，因为它们使用内存中的数据结构来在传输记录之前或之后组织记录。具体来说，reduceByKey和aggregateByKey在地图侧创建这些结构，并且'ByKey操作在reduce侧生成这些结构。当数据不适合内存时，Spark会将这些表溢出到磁盘，从而导致磁盘I / O的额外开销和垃圾收集增加。

Shuffle还会在磁盘上生成大量中间文件。从Spark 1.3开始，这些文件将被保留，直到不再使用相应的RDD并进行垃圾回收。这样做是为了在重新计算谱系时不需要重新创建shuffle文件。如果应用程序保留对这些RDD的引用或GC不经常启动，则垃圾收集可能仅在很长一段时间后才会发生。这意味着长时间运行的Spark作业可能会占用大量磁盘空间。配置Spark上下文时，spark.local.dir配置参数指定临时存储目录。

可以通过调整各种配置参数来调整随机行为。请参阅“Spark配置指南”中的“随机行为”部分。

### RDD Persistence ###

Spark中最重要的功能之一是跨操作在内存中持久化（或缓存）数据集。当您持久保存RDD时，每个节点都会存储它在内存中计算的任何分区，并在该数据集（或从中派生的数据集）的其他操作中重用它们。这使得未来的行动更快（通常超过10倍）。缓存是迭代算法和快速交互式使用的关键工具。

您可以使用persist（）或cache（）方法标记要保留的RDD。第一次在动作中计算它时，它将保留在节点的内存中。 Spark的缓存是容错的 - 如果RDD的任何分区丢失，它将使用最初创建它的转换自动重新计算。

此外，每个持久化RDD可以使用不同的存储级别进行存储，例如，允许您将数据集保留在磁盘上，将其保留在内存中，但作为序列化Java对象（以节省空间），跨节点复制它。通过将StorageLevel对象（Scala，Java，Python）传递给persist（）来设置这些级别。 cache（）方法是使用默认存储级别的简写，即StorageLevel.MEMORY\_ONLY（在内存中存储反序列化的对象）。完整的存储级别是：

**MEMORY\_ONLY**将RDD存储为JVM中的反序列化Java对象。如果RDD不适合内存，则某些分区将不会被缓存，并且每次需要时都会重新计算。这是默认级别。

**MEMORY\_AND\_DISK**将RDD存储为JVM中的反序列化Java对象。如果RDD不适合内存，请存储不适合磁盘的分区，并在需要时从那里读取它们。

**MEMORY\_ONLY\_SER**

（Java和Scala）将RDD存储为序列化Java对象（每个分区一个字节数组）。这通常比反序列化对象更节省空间，特别是在使用快速序列化器时，但读取CPU密集程度更高。

**MEMORY\_AND\_DISK\_SER**

（Java和Scala）与MEMORY\_ONLY\_SER类似，但是将内存中不适合的分区溢出到磁盘，而不是每次需要时即时重新计算它们。

**DISK\_ONLY**仅将RDD分区存储在磁盘上。

**MEMORY\_ONLY\_2****，MEMORY\_AND\_DISK\_2**等。与上面的级别相同，但复制两个群集节点上的每个分区。

**OFF\_HEAP**（实验）与MEMORY\_ONLY\_SER类似，但将数据存储在堆外内存中。这需要启用堆外内存。

注意：在Python中，存储对象将始终使用Pickle库进行序列化，因此选择序列化级别无关紧要。 Python中的可用存储级别包括MEMORY\_ONLY，MEMORY\_ONLY\_2，MEMORY\_AND\_DISK，MEMORY\_AND\_DISK\_2，DISK\_ONLY和DISK\_ONLY\_2。

即使没有用户调用持久性，Spark也会在随机操作（例如reduceByKey）中自动保留一些中间数据。 这样做是为了避免在shuffle期间节点出现故障时重新计算整个输入。 我们仍然建议用户在生成的RDD上调用持久性，如果他们计划重用它。

Which Storage Level to Choose?

Spark的存储级别旨在提供内存使用和CPU效率之间的不同折衷。我们建议您通过以下流程选择一个：

1.  如果您的RDD与默认存储级别（MEMORY\_ONLY）很舒适，请保持这种状态。这是CPU效率最高的选项，允许RDD上的操作尽可能快地运行。

1.  如果没有，请尝试使用MEMORY\_ONLY\_SER并选择快速序列化库，以使对象更加节省空间，但仍然可以快速访问。 （Java和Scala）

1.  除非计算数据集的函数很昂贵，否则它们不会溢出到磁盘，或者它们会过滤大量数据。否则，重新计算分区可能与从磁盘读取分区一样快。

1.  如果要快速故障恢复，请使用复制的存储级别（例如，如果使用Spark来处理来自Web应用程序的请求）。所有存储级别通过重新计算丢失的数据提供完全容错，但复制的存储级别允许您继续在RDD上运行任务，而无需等待重新计算丢失的分区。

Removing Data

Spark会自动监视每个节点上的缓存使用情况，并以最近最少使用（LRU）的方式删除旧数据分区。 如果您**想手动删除****RDD**而不是等待它从缓存中删除，请使用**RDD.unpersist****（）**方法。

## Shared Variables ##

通常，当在远程集群节点上执行传递给Spark操作（例如map或reduce）的函数时，它将在函数中使用的所有变量的单独副本上工作。 这些变量将复制到每台计算机，并且远程计算机上的变量更新不会传播回驱动程序。 支持跨任务的通用，读写共享变量效率低下。 但是，Spark确实为两种常见的使用模式提供了两种有限类型的**共享变量**：**广播变量**和**累加器**。

### Broadcast Variables ###

广播变量允许程序员在每台机器上保留一个只读变量，而不是随副本一起发送它的副本。例如，它们可用于以有效的方式为每个节点提供大输入数据集的副本。 Spark还尝试使用有效的广播算法来分发广播变量，以降低通信成本。

Spark动作通过一组阶段执行，由分布式“shuffle”操作分隔。 Spark自动广播每个阶段中任务所需的公共数据。以这种方式广播的数据以序列化形式缓存并在运行每个任务之前反序列化。这意味着显式创建广播变量仅在跨多个阶段的任务需要相同数据或以反序列化形式缓存数据很重要时才有用。

通过调用SparkContext.broadcast（v）从变量v创建广播变量。广播变量是v的包装器，可以通过调用value方法访问其值。下面的代码显示了这个：

**Broadcast****<****int****\[\]>** **broadcastVar** **=** **sc****.****broadcast****(****new****int****\[\]****\{** **1****,****2****,****3****\});**

**broadcastVar****.****value****();**

***// returns \[1, 2, 3\]***

创建广播变量后，应该在群集上运行的任何函数中使用它而不是值v，这样v不会多次传送到节点。 另外，在广播之后不应修改对象v，以确保所有节点获得广播变量的相同值（例如，如果稍后将变量发送到新节点）。

### Accumulators ###

累加器是仅通过关联和交换操作“添加”的变量，因此可以并行地有效支持。 它们可用于实现计数器（如MapReduce）或总和。 Spark本身支持数值类型的累加器，程序员可以添加对新类型的支持。

作为用户，您可以创建命名或未命名的累加器。 如下图所示，命名累加器（在此实例计数器中）将在Web UI中显示修改该累加器的阶段。 Spark显示“任务”表中任务修改的每个累加器的值。

可以通过调用**SparkContext.longAccumulator****（）**或**SparkContext.doubleAccumulator****（）**来创建数字累加器，以分别累积Long或Double类型的值。 然后，可以使用**add****方法**将群集上运行的任务添加到群集中。 但是，他们无法读懂它的价值。 **只有驱动程序可以使用其value****方法读取累加器的值**。

下面的代码显示了一个累加器用于添加数组的元素：

**LongAccumulator accum = jsc.sc().longAccumulator();**

**sc.parallelize(Arrays.asList(1, 2, 3, 4)).foreach(x -> accum.add(x));**

**// ...**

**// 10/09/29 18:41:08 INFO SparkContext: Tasks finished in 0.317106 s**

**accum.value();**

**// returns 10**

虽然此代码使用了对Long类型累加器的内置支持，但程序员也可以通过继承AccumulatorV2来创建自己的类型。 AccumulatorV2抽象类有几个方法，必须覆盖：reset用于将累加器重置为零，添加用于向累加器中添加另一个值，用于合并另一个相同类型的累加器到这个中。 其他必须覆盖的方法包含在API文档中。 例如，假设我们有一个表示数学向量的MyVector类，我们可以写：

**class VectorAccumulatorV2 implements AccumulatorV2<MyVector, MyVector> \{**

**  private MyVector myVector = MyVector.createZeroVector();**

**  public void reset() \{**

**    myVector.reset();**

**  \}**

**  public void add(MyVector v) \{**

**    myVector.add(v);**

**  \}**

**  ...**

**\}**

**// Then, create an Accumulator of this type:**

**VectorAccumulatorV2 myVectorAcc = new VectorAccumulatorV2();**

**// Then, register it into spark context:**

**jsc.sc().register(myVectorAcc, "MyVectorAcc1");**

请注意，当程序员定义自己的AccumulatorV2类型时，结果类型可能与添加的元素类型不同。

对于仅在操作内执行的累加器更新，Spark保证每个任务对累加器的更新仅应用一次，即重新启动的任务不会更新该值。 在转换中，用户应该知道，如果重新执行任务或作业阶段，则可以多次应用每个任务的更新。

累加器不会改变Spark的惰性评估模型。 如果在RDD上的操作中更新它们，则只有在RDD作为操作的一部分计算时才更新它们的值。 因此，在像map（）这样的惰性转换中进行累积器更新时，不能保证执行累加器更新。 以下代码片段演示了此属性：

**LongAccumulator accum** **=** **jsc****.****sc****().****longAccumulator****();**

**data****.****map****(****x** **->****\{**  **accum****.****add****(****x****);****return** **f****(****x****);****\});**

***// Here, accum is still 0 because no actions have caused the \`map\` to be computed.***

## Deploying to a Cluster ##

应用程序提交指南介绍了如何将应用程序提交到群集。 简而言之，一旦将应用程序打包到JAR（用于Java / Scala）或一组.py或.zip文件（用于Python），bin / spark-submit脚本允许您将其提交给任何支持的集群管理器。

## Launching Spark jobs from Java / Scala ##

org.apache.spark.launcher包提供了使用简单Java API将Spark作业作为子进程启动的类。

## Unit Testing ##

Spark对任何流行的单元测试框架进行单元测试都很友好。 只需在测试中创建一个SparkContext，将主URL设置为local，运行您的操作，然后调用SparkContext.stop（）将其拆除。 确保在finally块或测试框架的tearDown方法中停止上下文，因为Spark不支持在同一程序中同时运行的两个上下文。