spark的算子

spark的算子

1.spark的单Value算子

Spark中的单Value算子是指对一个RDD中的每个元素进行操作，并返回一个新的RDD。下面详细介绍一些常用的单Value算子及其功能：

1. map：逐条映射，将RDD中的每个元素通过指定的函数转换成另一个值，最终返回一个新的RDD。

rdd = sc.parallelize([1, 2, 3, 4, 5])
result = rdd.map(lambda x: x * 2)
# result: [2, 4, 6, 8, 10]

2. flatMap: 扁平化映射，将RDD中的每个元素通过指定的函数转换成多个值，并将这些值展开为一个新的RDD。

rdd = sc.parallelize([1, 2, 3, 4, 5])
result = rdd.flatMap(lambda x: range(x, x+3))
# result: [1, 2, 3, 2, 3, 4, 3, 4, 5, 4, 5, 6, 5, 6, 7]

3. glom：将一个分区中的多个单条数据转换为相同类型的单个数组进行处理。返回一个新的RDD，其中每个元素是一个数组。

rdd = sc.parallelize([1, 2, 3, 4, 5], 2)  # 两个分区
result = rdd.glom().collect()
# result: [[1, 2], [3, 4, 5]]

4. groupBy: 将RDD中的元素按照指定条件分组，返回一个键值对RDD，其中的每个元素是一个(key, iterator)对，key为分组的条件，iterator为对应分组的元素迭代器。

rdd = sc.parallelize(['apple', 'banana', 'cherry', 'date'])
result = rdd.groupBy(lambda x: x[0])
# result: [('a', ['apple']), ('b', ['banana']), ('c', ['cherry']), ('d', ['date'])]

5. filter：根据指定的规则过滤出符合条件的元素，返回一个新的RDD。

rdd = sc.parallelize([1, 2, 3, 4, 5])
result = rdd.filter(lambda x: x % 2 == 0)
# result: [2, 4]

6. sample：从RDD中进行采样，返回一个包含采样结果的新的RDD。

rdd = sc.parallelize(range(10))
result = rdd.sample(False, 0.5)
# result: [0, 2, 3, 4, 5, 7]

7. distinct（shuffle）：去重，将RDD中重复的元素去除，返回一个由不重复元素组成的新的RDD。

rdd = sc.parallelize([1, 2, 2, 3, 3, 3])
result = rdd.distinct()
# result: [1, 2, 3]

8. coalesce（shuffle）：缩减分区，将RDD的分区数缩减为指定的数量。

rdd = sc.parallelize([1, 2, 3, 4, 5], 4)  # 4个分区
result = rdd.coalesce(2)
# result: [1, 2, 3, 4, 5]（分区数变为2）

9. repartition（shuffle）：扩增分区数，底层是coalesce。将RDD的分区数扩增到指定的数量。

rdd = sc.parallelize([1, 2, 3, 4, 5], 2)  # 2个分区
result = rdd.repartition(4)
# result: [1, 2], [3, 4], [5]（分区数变为4）

10. sortBy（shuffle）：根据指定的规则对数据源中的数据进行排序，默认为升序。

rdd = sc.parallelize([3, 1, 4, 2, 5])
result = rdd.sortBy(lambda x: x)
# result: [1, 2, 3, 4, 5]

这些单Value算子能够对RDD中的每个元素进行处理，并返回一个新的RDD，可以用于各种数据转换、过滤、去重等操作。

2. Spark的双Value算子

双Value算子是指对两个RDD进行操作，并返回一个新的RDD。下面介绍一些常用的双Value算子及其功能：

1. union: 对两个RDD求并集，返回包含两个RDD中所有元素的新RDD。

rdd1 = sc.parallelize([1, 2, 3])
rdd2 = sc.parallelize([3, 4, 5])
result = rdd1.union(rdd2)
# result: [1, 2, 3, 3, 4, 5]

2. intersection: 对两个RDD求交集，返回包含两个RDD中共有元素的新RDD。

rdd1 = sc.parallelize([1, 2, 3])
rdd2 = sc.parallelize([3, 4, 5])
result = rdd1.intersection(rdd2)
# result: [3]

3. subtract: 对两个RDD求差集，返回只属于第一个RDD而不属于第二个RDD的元素的新RDD。

rdd1 = sc.parallelize([1, 2, 3])
rdd2 = sc.parallelize([3, 4, 5])
result = rdd1.subtract(rdd2)
# result: [1, 2]

4. cartesian: 对两个RDD进行笛卡尔积操作，返回所有可能的元素对组成的新RDD。

rdd1 = sc.parallelize([1, 2])
rdd2 = sc.parallelize(['a', 'b'])
result = rdd1.cartesian(rdd2)
# result: [(1, 'a'), (1, 'b'), (2, 'a'), (2, 'b')]

5. zip: 将两个RDD的元素按照索引位置进行配对，返回键值对组成的新RDD。

rdd1 = sc.parallelize([1, 2, 3])
rdd2 = sc.parallelize(['a', 'b', 'c'])
result = rdd1.zip(rdd2)
# result: [(1, 'a'), (2, 'b'), (3, 'c')]

6. join: 对两个键值对RDD进行内连接操作，返回具有相同键的元素对组成的新RDD。

rdd1 = sc.parallelize([(1, 'apple'), (2, 'banana')])
rdd2 = sc.parallelize([(1, 'red'), (2, 'yellow')])
result = rdd1.join(rdd2)
# result: [(1, ('apple', 'red')), (2, ('banana', 'yellow'))]

7. leftOuterJoin: 对两个键值对RDD进行左外连接操作，返回左侧RDD中所有元素以及与之匹配的右侧RDD中的元素对组成的新RDD。

rdd1 = sc.parallelize([(1, 'apple'), (2, 'banana')])
rdd2 = sc.parallelize([(1, 'red'), (3, 'yellow')])
result = rdd1.leftOuterJoin(rdd2)
# result: [(1, ('apple', 'red')), (2, ('banana', None))]

8. rightOuterJoin: 对两个键值对RDD进行右外连接操作，返回右侧RDD中所有元素以及与之匹配的左侧RDD中的元素对组成的新RDD。

rdd1 = sc.parallelize([(1, 'apple'), (2, 'banana')])
rdd2 = sc.parallelize([(1, 'red'), (3, 'yellow')])
result = rdd1.rightOuterJoin(rdd2)
# result: [(1, ('apple', 'red')), (3, (None, 'yellow'))]

这些双Value算子能够对两个RDD进行操作，并返回一个新的RDD，可以用于求并集、交集、差集等操作，也可以进行连接操作，根据键值对进行配对。

3. Spark的Key-Value算子

Key-Value算子是指对键值对RDD进行操作的算子，这些算子主要用于处理具有键值对结构的数据，其中键位于第一列，值位于第二列。下面介绍一些常用的Key-Value算子及其功能：

1. reduceByKey: 对具有相同键的元素进行聚合操作，返回一个新的键值对RDD。

rdd = sc.parallelize([(1, 2), (1, 3), (2, 4), (2, 5)])
result = rdd.reduceByKey(lambda x, y: x + y)
# result: [(1, 5), (2, 9)]

2. groupByKey: 对具有相同键的元素进行分组操作，返回一个新的键值对RDD。

rdd = sc.parallelize([(1, 2), (1, 3), (2, 4), (2, 5)])
result = rdd.groupByKey()
# result: [(1, <pyspark.resultiterable.ResultIterable object at 0x7f3128a3e370>), (2, <pyspark.resultiterable.ResultIterable object at 0x7f3128a3e3d0>)]

3. sortByKey: 按照键的顺序对RDD进行排序操作，默认升序排列。

rdd = sc.parallelize([(3, 'apple'), (1, 'banana'), (2, 'orange')])
result = rdd.sortByKey()
# result: [(1, 'banana'), (2, 'orange'), (3, 'apple')]

4. mapValues: 对键值对RDD中的值进行操作，返回一个新的键值对RDD。

rdd = sc.parallelize([(1, 'apple'), (2, 'banana')])
result = rdd.mapValues(lambda x: 'fruit ' + x)
# result: [(1, 'fruit apple'), (2, 'fruit banana')]

5. flatMapValues: 对键值对RDD中的值进行扁平化操作，返回一个新的键值对RDD。

rdd = sc.parallelize([(1, 'hello world'), (2, 'goodbye')])
result = rdd.flatMapValues(lambda x: x.split())
# result: [(1, 'hello'), (1, 'world'), (2, 'goodbye')]

6. keys: 返回所有键组成的一个新的RDD。

rdd = sc.parallelize([(1, 'apple'), (2, 'banana')])
result = rdd.keys()
# result: [1, 2]

7. values: 返回所有值组成的一个新的RDD。

rdd = sc.parallelize([(1, 'apple'), (2, 'banana')])
result = rdd.values()
# result: ['apple', 'banana']

除了上述提到的常用Key-Value算子，还有一些其他常见的Key-Value算子，它们在处理键值对RDD时也非常有用。以下是其中几个：

1. countByKey: 统计每个键出现的次数，返回一个字典。

rdd = sc.parallelize([(1, 'apple'), (1, 'banana'), (2, 'orange'), (2, 'banana')])
result = rdd.countByKey()
# result: {1: 2, 2: 2}

2. collectAsMap: 将键值对RDD转换为字典形式。

rdd = sc.parallelize([(1, 'apple'), (2, 'banana')])
result = rdd.collectAsMap()
# result: {1: 'apple', 2: 'banana'}

3. lookup: 查找具有给定键的所有值，并返回一个列表。

rdd = sc.parallelize([(1, 'apple'), (2, 'banana'), (1, 'orange')])
result = rdd.lookup(1)
# result: ['apple', 'orange']

4. foldByKey: 对具有相同键的元素进行折叠操作，类似于reduceByKey，但可以指定初始值。

rdd = sc.parallelize([(1, 2), (1, 3), (2, 4), (2, 5)])
result = rdd.foldByKey(0, lambda x, y: x + y)
# result: [(1, 5), (2, 9)]

5. aggregateByKey: 对具有相同键的元素进行聚合操作，可以指定初始值和两个函数：一个用于局部聚合，另一个用于全局聚合。

rdd = sc.parallelize([(1, 2), (1, 3), (2, 4), (2, 5)])
result = rdd.aggregateByKey(0, lambda x, y: x + y, lambda x, y: x + y)
# result: [(1, 5), (2, 9)]

这些Key-Value算子能够对键值对RDD进行操作，实现聚合、分组、排序、映射等功能。使用这些算子可以更方便地处理具有键值对结构的数据。