Spark RDD编程

朴灿烈づ我的快乐病毒、 2022-08-21 05:42 188阅读 0赞

# Spark RDD编程 #

## 概述 ##

从高层次上来看，每一个Spark应用都包含一个驱动程序，用于执行用户的main函数以及在集群上运行各种并行操作。Spark提供的主要抽象是弹性分布式数据集（RDD），这是一个包含诸多元素、被划分到不同节点上进行并行处理的数据集合。RDD通过打开HDFS（或其他hadoop支持的文件系统）上的一个文件、在驱动程序中打开一个已有的Scala集合或由其他RDD转换操作得到。用户可以要求Spark将RDD持久化到内存中（presist()操作），这样就可以有效地在并行操作中复用。另外，在节点发生错误时RDD可以自动恢复。

Spark提供的另一个抽象是可以在并行操作中使用的共享变量。在默认情况下，当Spark将一个函数转化成许多任务在不同的节点上运行的时候，对于所有在函数中使用的变量，每一个任务都会得到一个副本。有时，某一个变量需要在任务之间或任务与驱动程序之间共享。Spark支持两种共享变量：广播变量，用来将一个值缓存到所有节点的内存中；累加器，只能用于累加，比如计数器和求和。

## 运行Python脚本 ##

spark运行python脚本，需要使用Spark自带的bin/spark-submit脚本来运行。spark-sumbit脚本会帮助我们引入Python程序的Spark依赖。这个脚本为Spark的PythonAPI配置好了运行环境。

bin/spark-submit my_script.py

## 初始化Spark ##

在一个Spark程序中要做的第一件事就是创建一个SparkContext对象来告诉Spark如何连接一个集群。为了创建SparkContext，你首先需要创建一个SparkConf对象，这个对象会包含你的应用的一些相关信息。

from pyspark import SparkConf, SparkContext
    
    conf = SparkConf().setMaster("local").setAppName("My App")
    sc = SparkContext(conf = conf)

应该是可以由文件来进行配置的，在哪里做呢？？？？

appName参数是在集群UI上显示的此引用的名称。master是一个Spark、Mesos或YARN集群的URL，如果在本地运行那么这个参数应该是特殊的”local”字符串。在实际使用中，当你在集群中运行你的程序，你一般不会把master参数写死在代码中，而是通过用spark-submit运行程序来获得这个参数。但是，在本地测试以及单元测试时，你仍需要自行传入”local”来运行Spark程序。

## 弹性分布式数据集（RDD） ##

Spark是以RDD概念为中心运行的。RDD是一个容错的、可以被并行操作的元素集合。创建一个RDD有两个方法：在你的驱动程序中并行化一个已经存在的集合；从外部存储系统中引用一个数据集，这个存储系统可以是一个共享文件系统，比如HDFS、HBase或任意提供了Hadoop输入格式的数据来源。

### 并行化集合 ###

并行化集合是通过在驱动程序中一个现有的迭代器或集合上调用SparkContext的parallelize方法建立的。为了创建一个能够并行操作的分布数据集，集合中的元素都会被拷贝。比如，以下语句创建了一个包含1到5的并行化集合：

data = [1, 2, 3, 4, 5]
    distData = sc.parallelize(data)

分布数据集（distData）被建立起来之后，就可以进行并行操作了。比如，我们可以调用disData.reduce(lambda a, b: a+b)来对元素进行叠加。

并行集合的一个重要参数是将数据集划分成分片的数量。对每一个分片，Spark会在集群中运行一个对应的任务。典型情况下，集群中的每一个CPU将对应运行2-4个分片。一般情况下，Spark会根据当前集群的情况自行设定分片数量。但是，你也可以通过将第二个参数传递给parallelize方法(比如sc.parallelize(data, 10))来手动确定分片数量。

注意：有些代码中会使用切片（slice，分片的同义词）这个术语来保持向下兼容性。

## 外部数据集 ##

PySpark可以通过Hadoop支持的外部数据源（包括本地文件系统、HDFS、 Cassandra、HBase、亚马逊S3等等）建立分布数据集。Spark支持文本文件、序列文件以及其他任何Hadoop输入格式文件。

通过文本文件创建RDD要使用SparkContext的textFile方法。这个方法会使用一个文件的URI（或本地文件路径，hdfs://、s3n://这样的URI等等）然后读入这个文件建立一个文本行的集合。以下是一个例子：

distFile = sc.textFile("data.txt")

建立完成后distFile上就可以调用数据集操作了。比如，我们可以调用map和reduce操作来叠加所有文本行的长度，代码如下：

distFile.map(lambda s: len(s)).reduce(lambda a, b: a + b)

在Spark中读入文件时有几点要注意：

1.  如果使用了本地文件路径时，要保证在worker节点上这个文件也能够通过这个路径访问。这点可以通过将这个文件拷贝到所有worker上或者使用网络挂载的共享文件系统来解决。
2.  包括textFile在内的所有基于文件的Spark读入方法，都支持将文件夹、压缩文件、包含通配符的路径作为参数。比如，以下代码都是合法的：

textFile("/my/directory")
    textFile("/my/directory/*.txt")
    textFile("/my/directory/*.gz")

textFile方法也可以传入第二个可选参数来控制文件的分片数量。  
默认情况下，Spark会为文件的每一个块（在HDFS中块的大小默认是64MB）创建一个分片。但是你也可以通过传入一个更大的值来要求Spark建立更多的分片。注意，分片的数量绝不能小于文件块的数量。

除了文本文件之外，Spark的Python API还支持多种其他数据格式：  
SparkContext.wholeTextFiles能够读入包含多个小文本文件的目录，然后为每一个文件返回一个（文件名，内容)对。这是与textFile方法为每一个文本行返回一条记录相对应的。  
RDD.saveAsPickleFile和SparkContext.pickleFile支持将RDD以串行化的Python对象格式存储起来。串行化的过程中会以默认10个一批的数量批量处理。

>>> rdd = sc.parallelize(range(1, 4)).map(lambda x: (x, "a" * x ))
    >>> rdd.saveAsSequenceFile("path/to/file")
    >>> sorted(sc.sequenceFile("path/to/file").collect())
    [(1, u'a'), (2, u'aa'), (3, u'aaa')]

### 保存和读取其他Hadoop输入输出格式 ###

PySpark同样支持写入和读出其他Hadoop输入输出格式，包括’新’和’旧’两种Hadoop MapReduce API。如果有必要，一个Hadoop配置可以以Python字典的形式传入。以下是一个例子，使用了Elasticsearch ESInputFormat：

$ SPARK_CLASSPATH=/path/to/elasticsearch-hadoop.jar ./bin/pyspark
    >>> conf = {
       "es.resource" : "index/type"}   # assume Elasticsearch is running on localhost defaults
    >>> rdd = sc.newAPIHadoopRDD("org.elasticsearch.hadoop.mr.EsInputFormat",\
        "org.apache.hadoop.io.NullWritable", "org.elasticsearch.hadoop.mr.LinkedMapWritable", conf=conf)
    >>> rdd.first()         # the result is a MapWritable that is converted to a Python dict
    (u'Elasticsearch ID',{
       u'field1': True, u'field2': u'Some Text', u'field3': 12345})

## RDD操作 ##

RDD支持两类操作：转化操作，用于从已有的数据集转化产生新的数据集；行动操作，用于在计算结束后向驱动程序返回结果。举个例子，map是一个转化操作，可以将数据集中每一个元素传给一个函数，同时将计算结果作为一个新的RDD返回。另一方面，reduce操作是一个行动操作，能够使用某些函数来聚集计算RDD中所有的元素，并且向驱动程序返回最终结果（同时还有一个并行的reduceByKey操作可以返回一个分布数据集）。

在Spark所有的转化操作都是惰性求值的，就是说它们并不会立刻真的计算出结果。相反，它们仅仅是记录下了转换操作的操作对象（比如：一个文件）。只有当一个行动操作被执行，要向驱动程序返回结果时，转化操作才会真的开始计算。这样的设计使得Spark运行更加高效——比如，我们会发觉由map操作产生的数据集将会在reduce操作中用到，之后仅仅是返回了reduce的最终的结果而不是map产生的庞大数据集。  
在默认情况下，每一个由转化操作得到的RDD都会在每次执行行动操作时重新计算生成。但是，你也可以通过调用persist(或cache)方法来将RDD持久化到内存中，这样Spark就可以在下次使用这个数据集时快速获得。

Spark同样提供了对将RDD持久化到硬盘上或在多个节点间复制的支持。

## 基本操作 ##

为了演示RDD的基本操作，请看以下的简单程序:

lines = sc.textFile("data.txt")
    lineLengths = lines.map(lambda s: len(s))
    totalLength = lineLengths.reduce(lambda a, b: a + b)

第一行定义了一个由外部文件产生的基本RDD。这个数据集不是从内存中载入的也不是由其他操作产生的；lines仅仅是一个指向文件的指针。第二行将lineLengths定义为map操作的结果。再强调一次，由于惰性求值的缘故，lineLengths并不会被立即计算得到。最后，我们运行了reduce操作，这是一个行动操作。从这个操作开始，Spark将计算过程划分成许多任务并在多机上运行，每台机器运行自己部分的map操作和reduce操作，最终将自己部分的运算结果返回给驱动程序。

如果我们希望以后重复使用lineLengths，只需在reduce前加入下面这行代码：

lineLengths.persist()

这条代码将使得lineLengths在第一次计算生成之后保存在内存中。

### 向Spark传递函数 ###

在Python之中有三种方式将函数传递给Spark。传递比较短的函数时，可以使用lambda表达式来传递。另外还可以使用顶层函数和局部函数。

比如：

word = rdd.filter(lambda s: "error" in s)
    
    def containsError(s):
         return "error" in s
    
    word = rdd.filter(containsError)

在传递函数时需要小心的一点是，Python会在你不经意间把函数所在的对象也序列化传出去。当你传递的对象是某一个对象的成员，或者包含了对某个对象的一个字段的引用时（例如self.field），Spark就会把整个对象发送到工作节点上去，这可能比你想传递的东西大得多。例如：

class SearchFunction(object):
        def __init__(self, query):
            self.query = query
        def isMatch(self, s):
            return self.query in s
        def getMatchesFunctionReference(self, rdd):
            #问题：在self.isMatch中引用了整个self
            return rdd.filter(self.isMatch)
        def getMatchesMemberReference(self, rdd):
            #问题:在self.query中引用了整个self
            return rdd.filter(lambda x: self.query in x)

此类问题最简单的避免方法就是，使用一个本地变量缓存一份这个数据域的拷贝，直接访问这个数据域：

class WordFunction(object):
        ...
        def getMatchesNoReference(self, rdd):
        #安全：只把需要的数据提取到局部变量中
            query = self.query
            return rdd.filter(lambda x: query in x)

### 使用键值对 ###

虽然大部分Spark的RDD操作都支持所有种类的对象，但是有少部分特殊的操作只能作用于键值对类型的RDD。这类操作中最常见的就是分布的shuffle操作，比如将元素通过键来分组或聚集计算。在Python中，这类操作一般都会使用Python内建的元组类型，比如(1, 2)。它们会先简单地创建类似这样的元组，然后调用你想要的操作。比如，一下代码对键值对调用了reduceByKey操作,来统计每一文本行在文本文件中出现的次数：

lines = sc.textFile("data.txt")
    pairs = lines.map(lambda s: (s, 1)) #先生成内建元祖类型
    counts = pairs.reduceByKey(lambda a, b: a + b)

我们还可以使用counts.sortByKey()，比如，当我们想将这些键值对按照字母表顺序排序，然后调用counts.collect()方法来将结果以对象列表的形式返回。

### 转化&启动操作 ###

下面的表格列出了Spark支持的常用转化操作。欲知细节，请查阅RDD API文档（Scala, Java, Python）和键值对RDD函数文档（Scala, Java）。

转化操作  
map(func) | 返回一个新的分布数据集，由原数据集元素经func处理后的结果组成  
filter(func) | 返回一个新的数据集，由传给func返回True的原数据集元素组成  
flatMap(func) | 与map类似，但是每个传入元素可能有0或多个返回值，func可以返回一个序列而不是一个值  
mapParitions(func) | 类似map，但是RDD的每个分片都会分开独立运行，所以func的参数和返回值必须都是迭代器  
mapParitionsWithIndex(func) | 类似mapParitions，但是func有两个参数，第一个是分片的序号，第二个是迭代器。返回值还是迭代器  
sample(withReplacement, fraction, seed) | 使用提供的随机数种子取样，然后替换或不替换  
union(otherDataset) | 返回新的数据集，包括原数据集和参数数据集的所有元素  
intersection(otherDataset) | 返回新数据集，是两个集的交集  
distinct(\[numTasks\]) | 返回新的集，包括原集中的不重复元素  
groupByKey(\[numTasks\]) | 当用于键值对RDD时返回(键，值迭代器)对的数据集  
aggregateByKey(zeroValue)(seqOp, combOp, \[numTasks\]) | 用于键值对RDD时返回（K，U）对集，对每一个Key的value进行聚集计算  
sortByKey(\[ascending\], \[numTasks\])用于键值对RDD时会返回RDD按键的顺序排序，升降序由第一个参数决定  
join(otherDataset, \[numTasks\]) | 用于键值对(K, V)和(K, W)RDD时返回(K, (V, W))对RDD  
cogroup(otherDataset, \[numTasks\]) | 用于两个键值对RDD时返回  
(K, (V迭代器， W迭代器))RDD  
cartesian(otherDataset) | 用于T和U类型RDD时返回(T, U)对类型键值对RDD  
pipe(command, \[envVars\]) | 通过shell命令管道处理每个RDD分片  
coalesce(numPartitions) | 把RDD的分片数量降低到参数大小  
repartition(numPartitions) | 重新打乱RDD中元素顺序并重新分片，数量由参数决定  
repartitionAndSortWithinPartitions(partitioner) | 按照参数给定的分片器重新分片，同时每个分片内部按照键排序

### 启动操作 ###

下面的表格列出了Spark支持的部分常用启动操作。欲知细节，请查阅RDD API文档（Scala, Java, Python）和键值对RDD函数文档（Scala, Java）。  
（译者注：这部分翻译比较简略，仅供简单参考，具体细节请看文档）

### 行动操作 | 作用 ###

————| ——  
reduce(func) | 使用func进行聚集计算,func的参数是两个，返回值一个，两次func运行应当是完全解耦的，这样才能正确地并行运算  
collect() | 向驱动程序返回数据集的元素组成的数组  
count() | 返回数据集元素的数量  
first() | 返回数据集的第一个元素  
take(n) | 返回前n个元素组成的数组  
takeSample(withReplacement, num, \[seed\]) | 返回一个由原数据集中任意num个元素的suzuki，并且替换之  
takeOrder(n, \[ordering\]) | 返回排序后的前n个元素  
saveAsTextFile(path) | 将数据集的元素写成文本文件  
saveAsSequenceFile(path) | 将数据集的元素写成序列文件，这个API只能用于Java和Scala程序  
saveAsObjectFile(path) | 将数据集的元素使用Java的序列化特性写到文件中，这个API只能用于Java和Scala程序  
countByCount() | 只能用于键值对RDD，返回一个(K, int) hashmap，返回每个key的出现次数  
foreach(func) | 对数据集的每个元素执行func, 通常用于完成一些带有副作用的函数，比如更新累加器（见下文）或与外部存储交互等

## RDD持久化 ##

Spark的一个重要功能就是在将数据集持久化（或缓存）到内存中以便在多个操作中重复使用。当我们持久化一个RDD时，每一个节点将这个RDD的每一个分片计算并保存到内存中以便在下次对这个数据集（或者这个数据集衍生的数据集）的计算中可以复用。这使得接下来的计算过程速度能够加快（经常能加快超过十倍的速度）。缓存是加快迭代算法和快速交互过程速度的关键工具。你可以通过调用persist或cache方法来标记一个想要持久化的RDD。在第一次被计算产生之后，它就会始终停留在节点的内存中。Spark的缓存是具有容错性的——如果RDD的任意一个分片丢失了，Spark就会依照这个RDD产生的转化过程自动重算一遍。  
另外，每一个持久化的RDD都有一个可变的存储级别，这个级别使得用户可以改变RDD持久化的储存位置。比如，你可以将数据集持久化到硬盘上，也可以将它以序列化的Java对象形式（节省空间）持久化到内存中，还可以将这个数据集在节点之间复制，或者使用Tachyon将它储存到堆外。这些存储级别都是通过向persist()传递一个StorageLevel对象（Scala, Java, Python）来设置的。

注意：在Python中，储存的对象永远是通过Pickle库序列化过的，所以设不设置序列化级别不会产生影响。  
Spark还会在shuffle操作（比如reduceByKey）中自动储存中间数据，即使用户没有调用persist。这是为了防止在shuffle过程中某个节点出错而导致的全盘重算。不过如果用户打算复用某些结果RDD，  
我们仍然建议用户对结果RDD手动调用persist，而不是依赖自动持久化机制。

### 应该选择哪个存储级别？ ###

Spark的存储级别是为了提供内存使用与CPU效率之间的不同取舍平衡程度。我们建议用户通过考虑以下流程来选择合适的存储级别：

如果你的RDD很适合默认的级别（MEMORY\_ONLY）,那么久使用默认级别吧。这是CPU最高效运行的选择，能够让RDD上的操作以最快速度运行。

否则，试试MEMORY\_ONLY\_SER选项并且选择一个快的序列化库来使对象的空间利用率更高，同时尽量保证访问速度足够快。

不要往硬盘上持久化，除非重算数据集的过程代价确实很昂贵，或者这个过程过滤了巨量的数据。否则，重新计算分片有可能跟读硬盘速度一样快。

如果你希望快速的错误恢复（比如用Spark来处理web应用的请求），使用复制级别。所有的存储级别都提供了重算丢失数据的完整容错机制，但是复制一份副本能省去等待重算的时间。

在大内存或多应用的环境中，处于实验中的OFF\_HEAP模式有诸多优点：  
这个模式允许多个执行者共享Tachyon中的同一个内存池  
这个模式显著降低了垃圾回收的花销。  
在某一个执行者个体崩溃之后缓存的数据不会丢失。

## 删除数据 ##

Spark会自动监视每个节点的缓存使用同时使用LRU算法丢弃旧数据分片。如果你想手动删除某个RDD而不是等待它被自动删除，调用RDD.unpersist()方法。

## 共享变量 ##

通常情况下，当一个函数传递给一个在远程集群节点上运行的Spark操作（比如map和reduce）时，Spark会对涉及到的变量的所有副本执行这个函数。这些变量会被复制到每个机器上，而且这个过程不会被反馈给驱动程序。通常情况下，在任务之间读写共享变量是很低效的。但是，Spark仍然提供了有限的两种共享变量类型用于常见的使用场景：广播变量和累加器。

## 广播变量 ##

广播变量允许程序员在每台机器上保持一个只读变量的缓存而不是将一个变量的拷贝传递给各个任务。它们可以被使用，比如，给每一个节点传递一份大输入数据集的拷贝是很低效的。Spark试图使用高效的广播算法来分布广播变量，以此来降低通信花销。  
可以通过SparkContext.broadcast(v)来从变量v创建一个广播变量。这个广播变量是v的一个包装，同时它的值可以功过调用value方法来获得。以下的代码展示了这一点：

> > > broadcastVar = sc.broadcast(\[1, 2, 3\])  
> > >