SparkRDD及算子-python版

RDD相关知识

RDD介绍

RDD 是Spark的核心抽象，即 弹性分布式数据集（residenta distributed dataset）。代表一个不可变，可分区，里面元素可并行计算的集合。其具有数据流模型的特点：自动容错，位置感知性调度和可伸缩性。 在Spark中，对数据的所有操作不外乎创建RDD、转化已有RDD以及调用 RDD操作进行求值。
 

RDD结构图

RDD具有五大特性

  1. 一组分片（Partition），即数据集的基本组成单位（RDD是由一系列的partition组成的）。将数据加载为RDD时，一般会遵循数据的本地性（一般一个HDFS里的block会加载为一个partition）。

  2. RDD之间的依赖关系。依赖还具体分为宽依赖和窄依赖，但并不是所有的RDD都有依赖。为了容错（重算，cache，checkpoint），也就是说在内存中的RDD操作时出错或丢失会进行重算。

  3. 由一个函数计算每一个分片。Spark中的RDD的计算是以分片为单位的，每个RDD都会实现compute函数以达到这个目的。compute函数会对迭代器进行复合，不需要保存每次计算的结果。

  4. （可选）如果RDD里面存的数据是key-value形式，则可以传递一个自定义的Partitioner进行重新分区。

  5. （可选）RDD提供一系列最佳的计算位置，即数据的本地性。

RDD之间的依赖关系

RDD之间有一系列的依赖关系，依赖关系又分为窄依赖和宽依赖。

窄依赖：父RDD和子RDD partition之间的关系是一对一的。或者父RDD一个partition只对应一个子RDD的partition情况下的父RDD和子RDD partition关系是多对一的，也可以理解为没有触发shuffle。

宽依赖：父RDD与子RDD partition之间的关系是一对多。 父RDD的一个分区的数据去到子RDD的不同分区里面。也可以理解为触发了shuffle。

特别说明：对于join操作有两种情况，如果join操作的使用每个partition仅仅和已知的Partition进行join，此时的join操作就是窄依赖；其他情况的join操作就是宽依赖。

RDD创建

  1. 从Hadoop文件系统（或与Hadoop兼容的其他持久化存储系统，如Hive、Cassandra、HBase）输入（例如HDFS）创建。

  2. 通过集合进行创建。

算子

算子可以分为Transformation 转换算子和Action 行动算子。 RDD是懒执行的，如果没有行动操作出现，所有的转换操作都不会执行。

RDD直观图，如下：

RDD 的 五大特性

  • 一组分片（Partition），即数据集的基本组成单位。对于RDD来说，每个分片都会被一个计算任务处理，并决定并行计算的粒度。用户可以在创建RDD时指定RDD的分片个数，如果没有指定，那么就会采用默认值。默认值就是程序所分配到的CPU Core的数目。

  • 一个计算每个分区的函数。Spark中RDD的计算是以分片为单位的，每个 RDD都会实现 compute 函数以达到这个目的。compute函数会对迭代器进行复合，不需要保存每次计算的结果。

  • RDD之间的依赖关系。RDD的每次转换都会生成一个新的RDD，所以RDD之间就会形成类似于流水线一样的前后依赖关系。在部分分区数据丢失时，Spark 可以通过这个依赖关系重新计算丢失的分区数据，而不是对RDD的所有分区进行重新计算。

  • 一个Partitioner，即RDD的分片函数。当前Spark中实现了两种类型的分片函数，一个是基于哈希的HashPartitioner，另外一个是基于范围的 RangePartitioner。只有对于于key-value的RDD，才会有Partitioner，非key-value的RDD的Parititioner的值是None。Partitioner函数不但决定了RDD本身的分片数量，也决定了parent RDD Shuffle输出时的分片数量。

  • 一个列表，存储存取每个Partition的优先位置（preferred location）。对于一个HDFS文件来说，这个列表保存的就是每个 Partition 所在的块的位置。按照“移动数据不如移动计算”的理念，Spark 在进行任务调度的时候，会尽可能地将计算任务分配到其所要处理数据块的存储位置。

相关API介绍
  • SparkContext创建；
  1. sc = SparkContext("local", "Simple App")

说明："local" 是指让Spark程序本地运行，"Simple App" 是指Spark程序的名称，这个名称可以任意（为了直观明了的查看，最好设置有意义的名称）。

  • 集合并行化创建RDD；
  1. data = [1,2,3,4]
  2. rdd = sc.parallelize(data)
  • collect算子：在驱动程序中将数据集的所有元素作为数组返回（注意数据集不能过大）；
  1. rdd.collect()
  • 停止SparkContext。
  1. sc.stop()
# -*- coding: UTF-8 -*-
from pyspark import SparkContextif __name__ == "__main__":#********** Begin **********## 1.初始化 SparkContext，该对象是 Spark 程序的入口sc = SparkContext("local", "Simple App")# 2.创建一个1到8的列表Listdata = [1, 2, 3, 4, 5, 6, 7, 8]# 3.通过 SparkContext 并行化创建 rddrdd = sc.parallelize(data)# 4.使用 rdd.collect() 收集 rdd 的内容。 rdd.collect() 是 Spark Action 算子，在后续内容中将会详细说明，主要作用是：收集 rdd 的数据内容result = rdd.collect()# 5.打印 rdd 的内容print(result)# 6.停止 SparkContextsc.stop()#********** End **********#

读取外部数据集创建RDD 

编写读取本地文件创建Spark RDD的程序。

相关知识

为了完成本关任务，你需要掌握：1.如何读取本地文件系统中的文件来创建Spark RDD。

textFile 介绍

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

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

  1. distFile = sc.textFile("data.txt") 
    # -*- coding: UTF-8 -*-
from pyspark import SparkContextif __name__ == '__main__':#********** Begin **********## 1.初始化 SparkContext，该对象是 Spark 程序的入口sc = SparkContext("local", "Simple App")# 文本文件 RDD 可以使用创建 SparkContext 的t extFile 方法。
#此方法需要一个 URI的 文件（本地路径的机器上，或一个hdfs://，s3a://等URI），
#并读取其作为行的集合# 2.读取本地文件，URI为：/root/wordcount.txtrdd = sc.textFile("/root/wordcount.txt")# 3.使用 rdd.collect() 收集 rdd 的内容。 
#rdd.collect() 是 Spark Action 算子，在后续内容中将会详细说明，主要作用是：收集 rdd 的数据内容result = rdd.collect()# 4.打印 rdd 的内容print(result)# 5.停止 SparkContextsc.stop()#********** End **********#

map 算子

本关任务：使用Spark的 map 算子按照相关需求完成转换操作。

相关知识

为了完成本关任务，你需要掌握：如何使用map算子。

map

将原来RDD的每个数据项通过map中的用户自定义函数 f 映射转变为一个新的元素。

图中每个方框表示一个RDD 分区，左侧的分区经过自定义函数 f:T->U 映射为右侧的新 RDD 分区。但是，实际只有等到 Action 算子触发后，这个 f 函数才会和其他函数在一个 Stage 中对数据进行运算。

map 案例
  1. sc = SparkContext("local", "Simple App")
data = [1,2,3,4,5,6]
rdd = sc.parallelize(data)
print(rdd.collect())
rdd_map = rdd.map(lambda x: x * 2)
print(rdd_map.collect())

输出：

[1, 2, 3, 4, 5, 6] [2, 4, 6, 8, 10, 12]

说明：rdd1 的元素（ 1 , 2 , 3 , 4 , 5 , 6 ）经过 map 算子( x -> x*2 )转换成了 rdd2 ( 2 , 4 , 6 , 8 , 10 )。

编程要求

请仔细阅读右侧代码，根据方法内的提示，在Begin - End区域内进行代码补充，具体任务如下：

需求：使用 map 算子，将rdd的数据 (1, 2, 3, 4, 5) 按照下面的规则进行转换操作，规则如下:

  • 偶数转换成该数的平方；
  • 奇数转换成该数的立方。
# -*- coding: UTF-8 -*-
from pyspark import SparkContextif __name__ == "__main__":#********** Begin **********## 1.初始化 SparkContext，该对象是 Spark 程序的入口sc = SparkContext("local", "Simple App")# 2.创建一个1到5的列表Listdata = [1, 2, 3, 4, 5]# 3.通过 SparkContext 并行化创建 rddrdd = sc.parallelize(data)# 4.使用rdd.collect() 收集 rdd 的元素。print(rdd.collect())"""使用 map 算子，将 rdd 的数据 (1, 2, 3, 4, 5) 按照下面的规则进行转换操作，规则如下:需求：偶数转换成该数的平方奇数转换成该数的立方"""# 5.使用 map 算子完成以上需求rdd_map = rdd.map(lambda x: x * x if x % 2 == 0 else x * x * x)# 6.使用rdd.collect() 收集完成 map 转换的元素print(rdd_map.collect())# 7.停止 SparkContextsc.stop()#********** End **********#

mapPartitions算子

mapPartitions

mapPartitions函数获取到每个分区的迭代器，在函数中通过这个分区整体的迭 代器对整个分区的元素进行操作。

图中每个方框表示一个RDD分区，左侧的分区经过自定义函数 f:T->U 映射为右侧的新RDD分区。

mapPartitions 与 map

map：遍历算子，可以遍历RDD中每一个元素，遍历的单位是每条记录。

mapPartitions：遍历算子，可以改变RDD格式，会提高RDD并行度，遍历单位是Partition，也就是在遍历之前它会将一个Partition的数据加载到内存中。

那么问题来了，用上面的两个算子遍历一个RDD谁的效率高？ mapPartitions算子效率高。

mapPartitions 案例
  1. def f(iterator):
list = []
for x in iterator:
list.append(x*2)
return listif __name__ == "__main__":
sc = SparkContext("local", "Simple App")
data = [1,2,3,4,5,6]
rdd = sc.parallelize(data)
print(rdd.collect())
partitions = rdd.mapPartitions(f)
print(partitions.collect())

输出：


[1, 2, 3, 4, 5, 6]
[2, 4, 6, 8, 10, 12]

mapPartitions()：传入的参数是rdd的 iterator（元素迭代器），返回也是一个iterator（迭代器）。

# -*- coding: UTF-8 -*-
from pyspark import SparkContext#********** Begin **********#
def f(iterator):list = []for x in iterator:list.append((x, len(x)))return list#********** End **********#
if __name__ == "__main__":#********** Begin **********## 1.初始化 SparkContext，该对象是 Spark 程序的入口sc = SparkContext("local", "Simple App")# 2. 一个内容为（"dog", "salmon", "salmon", "rat", "elephant"）的列表Listdata = ["dog", "salmon", "salmon", "rat", "elephant"]# 3.通过 SparkContext 并行化创建 rddrdd = sc.parallelize(data)# 4.使用rdd.collect() 收集 rdd 的元素。print(rdd.collect())"""使用 mapPartitions 算子，将 rdd 的数据 ("dog", "salmon", "salmon", "rat", "elephant") 按照下面的规则进行转换操作，规则如下:需求：将字符串与该字符串的长度组合成一个元组，例如：dog  -->  (dog,3)salmon   -->  (salmon,6)"""# 5.使用 mapPartitions 算子完成以上需求partitions = rdd.mapPartitions(f)# 6.使用rdd.collect() 收集完成 mapPartitions 转换的元素print(partitions.collect())# 7.停止 SparkContextsc.stop()#********** End **********#

filter算子。

filter

filter 函数功能是对元素进行过滤，对每个元素应用f函数，返 回值为 true的元素在RDD中保留，返回值为false的元素将被过滤掉。内部实现相当于生成。

  1. FilteredRDD(this，sc.clean(f))

下面代码为函数的本质实现：

  1. def filter(self, f):
"""
Return a new RDD containing only the elements that satisfy a predicate.>>> rdd = sc.parallelize([1, 2, 3, 4, 5])
>>> rdd.filter(lambda x: x % 2 == 0).collect()
[2, 4]
"""
def func(iterator):
return filter(fail_on_stopiteration(f), iterator)
return self.mapPartitions(func, True)

上图中每个方框代表一个 RDD 分区， T 可以是任意的类型。通过用户自定义的过滤函数 f，对每个数据项操作，将满足条件、返回结果为 true 的数据项保留。例如，过滤掉 V2 和 V3 保留了 V1，为区分命名为 V’1。

filter 案例
  1. sc = SparkContext("local", "Simple App")
data = [1,2,3,4,5,6]
rdd = sc.parallelize(data)
print(rdd.collect())
rdd_filter = rdd.filter(lambda x: x>2)
print(rdd_filter.collect())

输出：

  1. [1, 2, 3, 4, 5, 6]
  2. [3, 4, 5, 6]

说明：rdd1( [ 1 , 2 , 3 , 4 , 5 , 6 ] ) 经过 filter 算子转换成 rdd2( [ 3 ,4 , 5 , 6 ] )。

使用 filter 算子，将 rdd 中的数据 (1, 2, 3, 4, 5, 6, 7, 8) 按照以下规则进行过滤，规则如下:

  • 过滤掉rdd中的所有奇数。
# -*- coding: UTF-8 -*-
from pyspark import SparkContextif __name__ == "__main__":#********** Begin **********## 1.初始化 SparkContext，该对象是 Spark 程序的入口sc = SparkContext("local", "Simple App")# 2.创建一个1到8的列表Listdata = [1, 2, 3, 4, 5, 6, 7, 8]# 3.通过 SparkContext 并行化创建 rddrdd = sc.parallelize(data)# 4.使用rdd.collect() 收集 rdd 的元素。print(rdd.collect())"""使用 filter 算子，将 rdd 的数据 (1, 2, 3, 4, 5, 6, 7, 8) 按照下面的规则进行转换操作，规则如下:需求：过滤掉rdd中的奇数"""# 5.使用 filter 算子完成以上需求rdd_filter = rdd.filter(lambda x: x % 2 == 0)# 6.使用rdd.collect() 收集完成 filter 转换的元素print(rdd_filter.collect())# 7.停止 SparkContextsc.stop()#********** End **********#

flatMap算子

flatMap

将原来RDD中的每个元素通过函数f转换为新的元素，并将生成的RDD中每个集合的元素合并为一个集合，内部创建：

  1. FlatMappedRDD(this，sc.clean(f))

上图表示RDD的一个分区，进行flatMap函数操作，flatMap中传入的函数为f:T->U，T和U可以是任意的数据类型。将分区中的数据通过用户自定义函数f转换为新的数据。外部大方框可以认为是一个RDD分区，小方框代表一个集合。V1、V2、V3在一个集合作为RDD的一个数据项，可能存储为数组或其他容器，转换为V’1、V’2、V’3后，将原来的数组或容器结合拆散，拆散的数据形成RDD中的数据项。

flatMap 案例
sc = SparkContext("local", "Simple App")
data = [["m"], ["a", "n"]]
rdd = sc.parallelize(data)
print(rdd.collect())
flat_map = rdd.flatMap(lambda x: x)
print(flat_map.collect())

输出：

  1. [['m'], ['a', 'n']]
  2. ['m', 'a', 'n']

flatMap：将两个集合转换成一个集合
 

需求：使用 flatMap 算子，将rdd的数据 ([1, 2, 3], [4, 5, 6], [7, 8, 9]) 按照下面的规则进行转换操作，规则如下:

  • 合并RDD的元素，例如：
    1. ([1,2,3],[4,5,6]) --> (1,2,3,4,5,6)
    2. ([2,3],[4,5],[6]) --> (1,2,3,4,5,6) 
      from pyspark import SparkContextif __name__ == "__main__":#********** Begin **********## 1.初始化 SparkContext，该对象是 Spark 程序的入口sc = SparkContext("local", "Simple App")# 2.创建一个[[1, 2, 3], [4, 5, 6], [7, 8, 9]] 的列表Listlist = [[1, 2, 3], [4, 5, 6], [7, 8, 9]]# 3.通过 SparkContext 并行化创建 rddrdd = sc.parallelize(list)# 4.使用rdd.collect() 收集 rdd 的元素。print(rdd.collect())        """使用 flatMap 算子，将 rdd 的数据 ([1, 2, 3], [4, 5, 6], [7, 8, 9]) 按照下面的规则进行转换操作，规则如下:需求：合并RDD的元素，例如：([1,2,3],[4,5,6])  -->  (1,2,3,4,5,6)([2,3],[4,5],[6])  -->  (1,2,3,4,5,6)"""# 5.使用 filter 算子完成以上需求flat_map = rdd.flatMap(lambda x: x)# 6.使用rdd.collect() 收集完成 filter 转换的元素print(flat_map.collect())# 7.停止 SparkContextsc.stop()#********** End **********#

      distinct 算子

      distinct

      distinct 将 RDD 中的元素进行去重操作。

      上图中的每个方框代表一个 RDD 分区，通过 distinct 函数，将数据去重。 例如，重复数据 V1、 V1 去重后只保留一份 V1 。

      distinct 案例
      sc = SparkContext("local", "Simple App")
data = ["python", "python", "python", "java", "java"]
rdd = sc.parallelize(data)
print(rdd.collect())
distinct = rdd.distinct()

    3. 输出

      ['python', 'python', 'python', 'java', 'java']
['python', 'java']

    5. print(distinct.collect())

      sortByKey 算子

      sortByKey
    6. def sortByKey(self, ascending=True, numPartitions=None, keyfunc=lambda x: x):
if numPartitions is None:
numPartitions = self._defaultReducePartitions()memory = self._memory_limit()
serializer = self._jrdd_deserializerdef sortPartition(iterator):
sort = ExternalSorter(memory * 0.9, serializer).sorted
return iter(sort(iterator, key=lambda kv: keyfunc(kv[0]), reverse=(not ascending)))if numPartitions == 1:
if self.getNumPartitions() > 1:
self = self.coalesce(1)
return self.mapPartitions(sortPartition, True)# first compute the boundary of each part via sampling: we want to partition
# the key-space into bins such that the bins have roughly the same
# number of (key, value) pairs falling into them
rddSize = self.count()
if not rddSize:
return self # empty RDD
maxSampleSize = numPartitions * 20.0 # constant from Spark's RangePartitioner
f\fraction = min(maxSampleSize / max(rddSize, 1), 1.0)
samples = self.sample(False, f\fraction, 1).map(lambda kv: kv[0]).collect()
samples = sorted(samples, key=keyfunc)# we have numPartitions many parts but one of the them has
# an implicit boundary
bounds = [samples[int(len(samples) * (i + 1) / numPartitions)]
for i in range(0, numPartitions - 1)]def rangePartitioner(k):
p = bisect.bisect_left(bounds, keyfunc(k))
if ascending:
return p
else:
return numPartitions - 1 - preturn self.partitionBy(numPartitions, rangePartitioner).mapPartitions(sortPartition, True)

    7. 说明：ascending参数是指排序（升序还是降序），默认是升序。numPartitions参数是重新分区，默认与上一个RDD保持一致。keyfunc参数是排序规则。

      sortByKey 案例
    8. sc = SparkContext("local", "Simple App")
    9. data = [("a",1),("a",2),("c",1),("b",1)]
    10. rdd = sc.parallelize(data)
    11. key = rdd.sortByKey()
    12. print(key.collect())

    13. 输出：

    14. [('a', 1), ('a', 2), ('b', 1), ('c', 1)]

 需求：使用 sortBy 算子，将 rdd 中的数据进行排序（升序）。

from pyspark import SparkContextif __name__ == "__main__":# ********** Begin **********## 1.初始化 SparkContext，该对象是 Spark 程序的入口sc = SparkContext("local", "Simple App")# 2.创建一个内容为[('B',1),('A',2),('C',3)]的列表ListList = [('B',1),('A',2),('C',3)]# 3.通过 SparkContext 并行化创建 rddrdd = sc.parallelize(List)# 4.使用rdd.collect() 收集 rdd 的元素print(rdd.collect())"""使用 sortByKey 算子，将 rdd 的数据 ('B', 1), ('A', 2), ('C', 3) 按照下面的规则进行转换操作，规则如下:需求：元素排序，例如：[(3,3),(2,2),(1,1)]  -->  [(1,1),(2,2),(3,3)]"""# 5.使用 sortByKey 算子完成以上需求key = rdd.sortByKey()# 6.使用rdd.collect() 收集完成 sortByKey 转换的元素print(key.collect())# 7.停止 SparkContextsc.stop()# ********** End **********#

mapValues 算子

mapValues

mapValues ：针对（Key， Value）型数据中的 Value 进行 Map 操作，而不对 Key 进行处理。

上图中的方框代表 RDD 分区。 a=>a+2 代表对 (V1,1) 这样的 Key Value 数据对，数据只对 Value 中的 1 进行加 2 操作，返回结果为 3。

mapValues 案例
  1. sc = SparkContext("local", "Simple App")
data = [("a",1),("a",2),("b",1)]
rdd = sc.parallelize(data)
values = rdd.mapValues(lambda x: x + 2)
print(values.collect())

输出：

  1. [('a', 3), ('a', 4), ('b', 3)]

需求：使用mapValues算子，将rdd的数据 ("1", 1), ("2", 2), ("3", 3), ("4", 4), ("5", 5) 按照下面的规则进行转换操作，规则如下:

  • 偶数转换成该数的平方
  • 奇数转换成该数的立方 
    from pyspark import SparkContextif __name__ == "__main__":# ********** Begin **********## 1.初始化 SparkContext，该对象是 Spark 程序的入口sc = SparkContext("local", "Simple App")    # 2.创建一个内容为[("1", 1), ("2", 2), ("3", 3), ("4", 4), ("5", 5)]的列表ListList = [("1", 1), ("2", 2), ("3", 3), ("4", 4), ("5", 5)]# 3.通过 SparkContext 并行化创建 rddrdd = sc.parallelize(List)# 4.使用rdd.collect() 收集 rdd 的元素print(rdd.collect())"""使用 mapValues 算子，将 rdd 的数据 ("1", 1), ("2", 2), ("3", 3), ("4", 4), ("5", 5) 按照下面的规则进行转换操作，规则如下:需求：元素（key,value）的value进行以下操作：偶数转换成该数的平方奇数转换成该数的立方"""# 5.使用 mapValues 算子完成以上需求values = rdd.mapValues(lambda x: x + 2)# 6.使用rdd.collect() 收集完成 mapValues 转换的元素print(values.collect())# 7.停止 SparkContextsc.stop()# ********** End **********#

    reduceByKey 算子

    reduceByKey

    reduceByKey 算子，只是两个值合并成一个值，比如叠加。

    函数实现

    def reduceByKey(self, func, numPartitions=None, partitionFunc=portable_hash):
return self.combineByKey(lambda x: x, func, func, numPartitions, partitionFunc)

上图中的方框代表 RDD 分区。通过自定义函数 (A,B) => (A + B) ，将相同 key 的数据 (V1,2) 和 (V1,1) 的 value 做加法运算，结果为（ V1,3）。

reduceByKey 案例
sc = SparkContext("local", "Simple App")
data = [("a",1),("a",2),("b",1)]
rdd = sc.parallelize(data)
print(rdd.reduceByKey(lambda x,y:x+y).collect())

输出：

[('a', 3), ('b', 1)]

需求：使用 reduceByKey 算子，将 rdd(key-value类型) 中的数据进行值累加。

例如：

  1. ("soma",4), ("soma",1)， ("soma",2） -> ("soma",7)
from pyspark import SparkContextif __name__ == "__main__":# ********** Begin **********## 1.初始化 SparkContext，该对象是 Spark 程序的入口sc = SparkContext("local", "Simple App")    # 2.创建一个内容为[("python", 1), ("scala", 2), ("python", 3), ("python", 4), ("java", 5)]的列表ListList = [("python", 1), ("scala", 2), ("python", 3), ("python", 4), ("java", 5)]# 3.通过 SparkContext 并行化创建 rddrdd = sc.parallelize(List)  # 4.使用rdd.collect() 收集 rdd 的元素print(rdd.collect())"""使用 reduceByKey 算子，将 rdd 的数据[("python", 1), ("scala", 2), ("python", 3), ("python", 4), ("java", 5)] 按照下面的规则进行转换操作，规则如下:需求：元素（key-value）的value累加操作，例如：(1,1),(1,1),(1,2)  --> (1,4)(1,1),(1,1),(2,2),(2,2)  --> (1,2),(2,4)"""# 5.使用 reduceByKey 算子完成以上需求reduce = rdd.reduceByKey(lambda x,y:x+y)# 6.使用rdd.collect() 收集完成 reduceByKey 转换的元素print(reduce.collect())# 7.停止 SparkContextsc.stop()# ********** End **********#

Action 的常用算子

count

count()：返回 RDD 的元素个数。

示例：

  1. sc = SparkContext("local", "Simple App")
data = ["python", "python", "python", "java", "java"]
rdd = sc.parallelize(data)
print(rdd.count())

输出：

  1. 5
first

first()：返回 RDD 的第一个元素（类似于take(1)）。

示例：

  1. sc = SparkContext("local", "Simple App")
data = ["python", "python", "python", "java", "java"]
rdd = sc.parallelize(data)
print(rdd.first())

输出：

  1. python
take

take(n)：返回一个由数据集的前 n 个元素组成的数组。

示例：

  1. sc = SparkContext("local", "Simple App")
data = ["python", "python", "python", "java", "java"]
rdd = sc.parallelize(data)
print(rdd.take(2))

输出：

  1. ['python', 'python']
reduce

reduce()：通过func函数聚集 RDD 中的所有元素，该函数应该是可交换的和关联的，以便可以并行正确计算。

示例：

  1. sc = SparkContext("local", "Simple App")
data = [1,1,1,1]
rdd = sc.parallelize(data)
print(rdd.reduce(lambda x,y:x+y))

输出：

  1. 4
collect

collect()：在驱动程序中，以数组的形式返回数据集的所有元素。

示例：

  1. sc = SparkContext("local", "Simple App")
data = [1,1,1,1]
rdd = sc.parallelize(data)
print(rdd.collect())

输出：

  1. [1,1,1,1]
具体任务如下：

需求1：使用 count 算子，统计下 rdd 中元素的个数；

需求2：使用 first 算子，获取 rdd 首个元素；

需求3：使用 take 算子，获取 rdd 前三个元素；

需求4：使用 reduce 算子，进行累加操作；

需求5：使用 collect 算子，收集所有元素。
 

from pyspark import SparkContext
if __name__ == "__main__":# ********** Begin **********## 1.初始化 SparkContext，该对象是 Spark 程序的入口sc = SparkContext("local", "Simple App")# 2.创建一个内容为[1, 3, 5, 7, 9, 8, 6, 4, 2]的列表ListList = [1, 3, 5, 7, 9, 8, 6, 4, 2]  # 3.通过 SparkContext 并行化创建 rddrdd = sc.parallelize(List)# 4.收集rdd的所有元素并print输出print(rdd.collect())# 5.统计rdd的元素个数并print输出print(rdd.count())# 6.获取rdd的第一个元素并print输出print(rdd.first())# 7.获取rdd的前3个元素并print输出print(rdd.take(3))# 8.聚合rdd的所有元素并print输出print(rdd.reduce(lambda x,y:x+y))# 9.停止 SparkContextsc.stop()# ********** End **********#

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.rhkb.cn/news/205080.html

如若内容造成侵权/违法违规/事实不符，请联系长河编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

蓝桥杯day02——移动机器人

蓝桥杯day02——移动机器人

1.题目 有一些机器人分布在一条无限长的数轴上，他们初始坐标用一个下标从 0 开始的整数数组 nums 表示。当你给机器人下达命令时，它们以每秒钟一单位的速度开始移动。 给你一个字符串 s ，每个字符按顺序分别表示每个机器人移动的方向。L 表…
阅读更多...
《opencv实用探索·四》Mat图像数据类型转换和归一化显示

《opencv实用探索·四》Mat图像数据类型转换和归一化显示

一种数据类型转为另一种数据类型，不改变图像大小，但每个像素值可能会变 src.convertTo(dst, type, scale, shift);Scale和shitf默认为0（这两个参数也相当于对比度和亮度） 现在有个8位图像，把8位转成32位 可以看到像素…
阅读更多...
基于SSM的仓库管理系统的设计与实现

基于SSM的仓库管理系统的设计与实现

末尾获取源码 开发语言：Java Java开发工具：JDK1.8 后端框架：SSM 前端：Vue 数据库：MySQL5.7和Navicat管理工具结合 服务器：Tomcat8.5 开发软件：IDEA / Eclipse 是否Maven项目：是 目录…
阅读更多...
JS 倒计时方法（可改造）

JS 倒计时方法（可改造）

起因： 写好备用。 代码： // 直接把方法写在了原型上，通过原型调用 /*** 倒计时* time_str String 到期时间(2023-11-28 16:50:00)* dom_obj Object 需要显示的倒计时的dom对象*/ Date.prototype.countdown function (time_str, dom_obj…
阅读更多...
JAVA小游戏简易版王者荣耀

JAVA小游戏简易版王者荣耀

第一步是创建项目 项目名自拟 第二部创建个包名 来规范class 然后是创建类 GameFrame 运行类 package com.sxt; import java.awt.Graphics; import java.awt.Image; import java.awt.Toolkit; import java.awt.event.ActionEvent; import java.awt.event.ActionListener;…
阅读更多...
【企业微信连接问题】

【企业微信连接问题】

1、个人可以创建企业微信的企业账号么？ 答：可以的，只是没法认证。不过基础的功能还是有的。 注册步骤：企业微信注册步骤 2、集简云链接企业微信，在授权之后，找不到集简云怎么办？ 答&#xff1a…
阅读更多...
CSS特效021：蛇形左右扭动的效果

CSS特效021：蛇形左右扭动的效果

CSS常用示例100专栏目录 本专栏记录的是经常使用的CSS示例与技巧，主要包含CSS布局，CSS特效，CSS花边信息三部分内容。其中CSS布局主要是列出一些常用的CSS布局信息点，CSS特效主要是一些动画示例，CSS花边是描述了一些CSS…
阅读更多...
vue中下载文件后无法打开的坑

vue中下载文件后无法打开的坑

今天在项目开发的时候临时要添加个导出功能我就写了一份请求加导出得代码， 代码： //导出按钮放开exportDutySummarizing (dataRangeInfo) {const params {departmentName: dataRangeInfo.name,departmentQode: dataRangeInfo.qode}//拼接所需得urlcons…
阅读更多...
策略算法与Actor-Critic网络

策略算法与Actor-Critic网络

策略算法 教程链接 DataWhale强化学习课程JoyRL https://johnjim0816.com/joyrl-book/#/ch7/main 策略梯度 与前面的基于价值的算法不同，这类算法直接对策略本身进行近似优化。 在这种情况下，我们可以将策略描述成一个带有参数 θ θ θ的连续函数…
阅读更多...
与 PCIe 相比，CXL为何低延迟高带宽？

与 PCIe 相比，CXL为何低延迟高带宽？

文章目录 前言1. LatencyPCIE 生产者消费则模型结论Flit 包PCIE/CXL.ioCXL.cace & .mem总结 2. BandWidth常见开销CXL.IO Link efficiencyPCIe Link efficiencyCXL.IO bandwidthCXL.mem/.cache bandwidth 参考 前言 CXL 规范里没有具体描述与PCIe 相比低延时高带宽的原因&…
阅读更多...
Java基于springoot开发的企业招聘求职网站

Java基于springoot开发的企业招聘求职网站

演示视频： https://www.bilibili.com/video/BV1xw411n7Tu/?share_sourcecopy_web&vd_source11344bb73ef9b33550b8202d07ae139b 技术：springootmysqlvuejsbootstrappoi制作word模板 主要功能：求职者可以注册发布简历，选择简…
阅读更多...
Echarts+vue+java+mysql实现数据可视化

Echarts+vue+java+mysql实现数据可视化

一、折线图，柱状图 https://echarts.apache.org/zh/index.html echarts 官网 更多配置项可以去官网查看 在开始项目之前，确保您已经安装了以下工具和技术： MySQL 数据库：用于存储和管理数据。Java 后端：用于创建后端应…
阅读更多...
解决api-ms-win-crt-runtime-l1-1-0.dll丢失的问题，全是干货分享

解决api-ms-win-crt-runtime-l1-1-0.dll丢失的问题，全是干货分享

今天我的电脑中突然出现关于“api-ms-win-crt-runtime-l1-1-0.dll”的错误提示，关闭提示后再次打开程序依然不能正常打开，出现这样的问题突然不知道是因为什么，于是就去了解了关于出现api-ms-win-crt-runtime-l1-1-0.dll错误的问题&#xff0…
阅读更多...
elFinder ZIP 参数注入导致命令注入 （CVE-2021-32682）

elFinder ZIP 参数注入导致命令注入 （CVE-2021-32682）

漏洞描述 elFinder 是一个用于 Web 的开源文件管理器，使用 jQuery UI 用 JavaScript 编写。 在 elFinder 2.1.48 及更早版本中发现一个参数注入漏洞。此漏洞可能允许攻击者在托管 elFinder PHP 连接器的服务器上执行任意命令，即使配置最少也是如此。这…
阅读更多...
matlab画双坐标图的样式

matlab画双坐标图的样式

matlab画双坐标图的样式 %% clc,clear,close all; t0:0.1:9*pi; figure; [AX,Ha,Hb]plotyy(t,sin(t),t,exp(t)); % 绘图并创建句柄 % ----------------- 设置刻度 set(AX(1),yTick,[-1.250:0.25:1.25]) % 设置左边Y轴的刻度 set(AX(2),yTick,[0:50:350]) …
阅读更多...
Alivia 1.0 正式版来了，打造更懂企业的营销「工具箱」

Alivia 1.0 正式版来了，打造更懂企业的营销「工具箱」

上周，「Whale 帷幄」2023 秋季发布会圆满落下帷幕。发布会上，帷幄创始人 & CEO 叶生晅重磅发布了专为营销和销售设计的企业级 AGI 工具——Alivia 1.0 正式版，获得了广泛的反响和好评。 在这一年里，帷幄在 AGI 产品创新及落地…
阅读更多...
Redis 命令处理过程

Redis 命令处理过程

我们知道 Redis 是一个基于内存的高性能键值数据库, 它支持多种数据结构, 提供了丰富的命令, 可以用来实现缓存、消息队列、分布式锁等功能。 而在享受 Redis 带来的种种好处时, 是否曾好奇过 Redis 是如何处理我们发往它的命令的呢？ 本文将以伪代码的形式简单分析…
阅读更多...
基于SSM的高校学生实习管理系统设计与实现

基于SSM的高校学生实习管理系统设计与实现

末尾获取源码 开发语言：Java Java开发工具：JDK1.8 后端框架：SSM 前端：Vue 数据库：MySQL5.7和Navicat管理工具结合 服务器：Tomcat8.5 开发软件：IDEA / Eclipse 是否Maven项目：是 目录…
阅读更多...
Unity之ARFoundation如何实现BodyTracking人体跟踪

Unity之ARFoundation如何实现BodyTracking人体跟踪

前言 ARBodyTracking，就是指通过手机AR扫描并精确的捕获人物的肢体部位的技术。如下图所示 这项技术目前是有苹果的ARKit提供，苹果的body tracking 功能需要使用配备 TrueDepth 摄像头的设备，配备 A12 仿生芯片、运行 iOS 13 或更高版本的设备，比如 iPhone X 及更新机型。…
阅读更多...
可以在电脑桌面展示工作计划表的软件

可以在电脑桌面展示工作计划表的软件

很多上班族都表示自己在工作时，会面临大量且复杂的工作任务，这时候就会拖延工作，或者感觉时间不够用，所以需要有明确的工作计划来指导自己如何分类时间和精力，确保每项工作任务都能够按时完成。如果需要制定每天的工作…
阅读更多...
最新文章
推荐文章

Copyright @ 2022~2023