大数据Spark框架 7：RDD算子相关操作

迷南。 2024-02-20 08:56 27阅读 0赞

### 3. RDD算子相关的操作 ###

RDD算子： 指的是RDD对象中提供了非常多的具有特殊功能的函数， 我们将这些函数称为算子 （大白话：指的RDD的API）

相关的算子的官方文档： [https://spark.apache.org/docs/3.1.2/api/python/reference/pyspark.html\#rdd-apis][https_spark.apache.org_docs_3.1.2_api_python_reference_pyspark.html_rdd-apis]

#### 3.1 RDD算子的分类 ####

整个RDD算子， 共分为二大类： transformation(转换)算子 和 action(动作)算子

转换算子：
        1- 所有的转换算子在执行完成后， 都会返回一个新的RDD
        2- 所有的转换算子都是lazy(惰性)，并不会立即执行， 认为仅仅是在定义计算的规则
        3- 转换算子必须遇到action算子才会触发执行
    
    动作算子：
        1- 动作算子在执行后， 不会返回一个RDD， 要不然没有返回值， 要不就返回其他的
        2- 动作算子都是立即执行的， 一个动作算子就会产生一个Job的任务， 运行这个动作算子所依赖的所有的RDD

相关的动作算子：

![e619e288bd85930a765ef76368d134eb.png][]

#### 3.2 RDD的转换算子 ####

值类型的算子：

*  map算子：
 *  格式： rdd.map(fn)
 *  说明: 主要根据传入的函数, 对数据进行一对一的转换操作, 传入一行, 返回一行

rdd = sc.parallelize([1,2,3,4,5,6,7,8,9,10])
    
    需求: 请对每一个元素进行 + 1返回
    rdd.map(lambda num: num + 1).collect()
    结果:
        [2, 3, 4, 5, 6, 7, 8, 9, 10, 11]

*  groupBy 算子:
 *  格式: groupBy(fn)
 *  说明: 根据传入的函数对数据进行分组操作

rdd = sc.parallelize([1,2,3,4,5,6,7,8,9,10])
    
    需求: 请将数据分为 奇数和偶数   三目操作
    rdd.groupBy(lambda num : 'o' if num % 2 == 0 else 'j' ).collect()
    结果为:
        [
            ('j', <pyspark.resultiterable.ResultIterable object at 0x7f895bd93dc0>), 
            ('o', <pyspark.resultiterable.ResultIterable object at 0x7f895bd7e190>)
        ]
    
    mapValues(list): 将kv中的value转换为list
    
    rdd.groupBy(lambda num : 'o' if num % 2 == 0 else 'j' ).mapValues(list).collect()
    结果:
        [('j', [1, 3, 5, 7, 9]), ('o', [2, 4, 6, 8, 10])]
    
    
    def  if_j_o(num):
        if num % 2 == 0:
            return 'o'
        else:
            return 'j'
    rdd.groupBy(if_j_o).mapValues(list).collect()
    结果:
        [
            ('j', [1, 3, 5, 7, 9]), 
            ('o', [2, 4, 6, 8, 10])
        ]

*  filter算子:
 *  格式: filter(fn)
 *  说明: 过滤算子, 可以根据函数中指定的过滤条件, 对数据进行过滤操作, 条件返回True表示保留, 返回False 表示过滤

rdd = sc.parallelize([1,2,3,4,5,6,7,8,9,10])
    
    需求: 请将 <=3的数据过滤掉
    rdd.filter(lambda num: num > 3).collect()
    结果:
        [4, 5, 6, 7, 8, 9, 10]

*  flatMap算子:
 *  格式: flatMap(fn)
 *  说明: 在map算子的基础上, 加入一个压扁的操作, 主要适用于一行中包含多个内容的操作, 实现一转多的操作

rdd = sc.parallelize(['张三 李四 王五 赵六','田七 周八 李九'])
    
    需求: 将其转换为一个个的姓名:
    rdd.flatMap(lambda line: line.split()).collect()
    结果:
        ['张三', '李四', '王五', '赵六', '田七', '周八', '李九']

双值类型算子:

*  union(并集) 和 intersection(交集)
 *  格式: rdd1.intersection|union(rdd2)

rdd1 = sc.parallelize([1,3,5,2,7,8])
    rdd2 = sc.parallelize([2,3,1,5,9,10])
    
    并集:
    rdd1.union(rdd2).collect()
    结果:
        [1, 3, 5, 2, 7, 8, 2, 3, 1, 5, 9, 10]
    
    想对并集的结果进行去重:
    rdd1.union(rdd2).distinct().collect()
    结果:
        [8, 1, 5, 9, 2, 10, 3, 7]
    
    交集:
    rdd1.intersection(rdd2).collect()
    结果:
        [1, 5, 2, 3]

**kv类型的相关算子:**

*  groupByKey()
 *  格式: groupByKey()
 *  说明: 根据key进行分组操作

rdd = sc.parallelize([('c01','张三'), ('c02','李四'), ('c02','王五'), ('c01','赵六'), ('c03','田七'), ('c03','周八'), ('c02','李九')])
    
    需求: 根据班级来分组统计
    rdd.groupByKey().mapValues(list).collect() 
    结果:
        [('c01', ['张三', '赵六']), ('c02', ['李四', '王五', '李九']), ('c03', ['田七', '周八'])]
    
    rdd.groupByKey().mapValues(list).map(lambda kv:(kv[0],len(kv[1]))).collect()
    结果:
        [('c01', 2), ('c02', 3), ('c03', 2)]

*  reduceByKey()
 *  格式: reduceByKey(fn)
 *  说明: 根据key进行分组, 将一个组内的value数据放置到一个列表中, 对这个列表基于fn进行聚合计算操作

rdd = sc.parallelize([('c01','张三'), ('c02','李四'), ('c02','王五'), ('c01','赵六'), ('c03','田七'), ('c03','周八'), ('c02','李九')])
    
    需求: 根据班级来分组统计
    rdd.reduceByKey(lambda agg,curr: agg + curr).collect()
    结果:
        [('c01', '张三赵六'), ('c02', '李四王五李九'), ('c03', '田七周八')]
    
    rdd.map(lambda kv:(kv[0],1)).reduceByKey(lambda agg,curr: agg + curr).collect()   
    结果:
        [('c01', 2), ('c02', 3), ('c03', 2)]

*  sortByKey()算子:
 *  格式: sortByKey(ascending=True|False)
 *  说明: 根据key进行排序操作, 默认按照key进行升序排序, 如果需要倒序,设置 ascending 参数的值为Flase

rdd = sc.parallelize([(10, 2), (15, 3), (8, 4), (7, 4), (2, 4), (12, 4)])
    
    根据key进行排序操作: 演示升序和倒序
    rdd.sortByKey().collect()
    结果:
        [(2, 4), (7, 4), (8, 4), (10, 2), (12, 4), (15, 3)]
    rdd.sortByKey(ascending=False).collect()
    结果:
        [(15, 3), (12, 4), (10, 2), (8, 4), (7, 4), (2, 4)]
    
    rdd = sc.parallelize([('c01', 2), ('c02', 3), ('c03', 2), ('c04', 2), ('c08', 2), ('c09', 2)])
    rdd.sortByKey().collect()
    结果:
    [('c01', 2), ('c02', 3), ('c03', 2), ('c04', 2), ('c08', 2), ('c09', 2)]
    
    rdd = sc.parallelize([('c01', 2), ('c02', 3), ('c03', 2), ('c011', 2), ('c08', 2), ('c09', 2)])
    rdd.sortByKey().collect()
    结果:
        [('c01', 2), ('c011', 2), ('c02', 3), ('c03', 2), ('c08', 2), ('c09', 2)]
    
    
    字典序: 主要是发生在字符串的排序中
        '1','3','5','2','111','22','36','4','52','8'
    
        先看第一位, 如果相等, 看第二位, 没有第二位的要比有第二位的要小, 依次类推, 每个字符排序按照ASCII排序顺序
    
        从小到大:  数字 < 大写字母 < 小写字母
    
        以上数字的字符串, 按照从小到大排序: 
            1 111 2 22 3 36 4 5 52 8
    
    数字序:
        1,3,8,4,9 --> 1,3,4,8,9

*  countByKey() 和 countByValue() ----- 了解
 *  说明:
 *  countByKey() 根据key 分组 统计每个分组下有多少个元素
 *  countByValue() 根据value进行分组, 统计相同的value有多少个

rdd = sc.parallelize([('c01','张三'), ('c02','李四'), ('c02','王五'), ('c01','赵六'), ('c03','田七'), ('c03','周八'), ('c02','李九')])
    
    
    rdd.countByKey()
    defaultdict(<class 'int'>, {'c01': 2, 'c02': 3, 'c03': 2})
    
    rdd.countByValue()  将列表的每一个元素 整体作为一个value来计算
    defaultdict(<class 'int'>, {('c01', '张三'): 1, ('c02', '李四'): 1, ('c02', '王五'): 1, ('c01', '赵六'): 1, ('c03', '田七'): 1, ('c03', '周八'): 1, ('c02', '李九'): 1})
    
    
    rdd = sc.parallelize([1,2,3,1,3,1,34,5,3,3,4])
    
    rdd.countByValue()
    defaultdict(<class 'int'>, {1: 3, 2: 1, 3: 4, 34: 1, 5: 1, 4: 1})

#### 3.3 RDD的动作算子 ####

*  collect() 算子:
 *  格式: collect()
 *  作用: 收集各个分区的数据, 将数据汇总到一个大的列表返回
 *  reduce() 算子:
 *  格式: reduce(fn)
 *  作用: 根据传入的函数对数据进行聚合操作

rdd = sc.parallelize([1,2,3,4,5,6,7,8,9,10])
    
    需求: 统计所有元素之和是多少
    rdd.reduce(lambda agg,curr: agg + curr)
    结果:
    55

*  first() 算子:
 *  格式: first()
 *  说明: 获取第一个元素

rdd = sc.parallelize([1,2,3,4,5,6,7,8,9,10])
    
    获取第一个元素:
    rdd.first()
    1

*  take() 算子
 *  格式: take(N)
 *  说明: 获取前N个元素 , 类似于 limit操作

rdd = sc.parallelize([1,2,3,4,5,6,7,8,9,10])
    
    获取 前3个元素
    rdd.take(3)
    结果:
        [1, 2, 3]

*  top()算子:
 *  格式: top(N,\[fn\])
 *  说明: 对数据集进行倒序排序操作, 如果kv类型, 针对key进行排序, 获取前N个元素
 *  fn: 可以自定义排序, 按照谁来排序

rdd = sc.parallelize([1,2,3,4,5,6,7,8,9,10])
    
    获取 前3个元素
    rdd.top(3)
    [10, 9, 8]
    
    rdd = sc.parallelize([('c01',5),('c02',8),('c03',4)])
    
    rdd.top(3,lambda res:res[1])
    结果:
        [('c02', 8), ('c01', 5), ('c03', 4)]

*  count() 算子

rdd = sc.parallelize([1,2,3,4,5,6,7,8,9,10])
    
    获取一共有多少个元素:
    rdd.count()
    结果:10

*  foreach() 算子
 *  格式: foreach(fn)
 *  作用: 对数据集进行遍历操作, 遍历后做什么, 取决于传入的fn

rdd = sc.parallelize([1,2,3,4,5,6,7,8,9,10],1)
    
    对数据进行遍历操作:
    rdd.foreach(lambda num: print(num))
    结果:
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    
    思考: 为啥有时候打出来结果不一样呢?  是由于RDD是有分区的, 每一个分区会对应一个线程, 多个线程都在执行foreach的打印操作, 各个线程互相争抢打印, 所有看到不一样的结果(而且每一次都有可能不一样)

*  takeSample() 算子
 *  格式: takeSample(True|False, N,seed(种子值))
 *  参数1: 是否允许重复采样
 *  参数2: 采样多少个, 如果允许重复采样, 采样个数不限制, 否则最多等于本身数量个数
 *  参数3: 设置种子值, 值可以随意填写, 一旦固定了, 表示每次采样的内容也是固定的 (可选的),如果没有特殊要求, 一般不设置
 *  作用: 数据采样工作

rdd = sc.parallelize([1,2,3,4,5,6,7,8,9,10])
    
    演示数据采样:
    >>> rdd.takeSample(True,5)
    [10, 3, 10, 8, 3]
    >>> rdd.takeSample(True,5)
    [1, 9, 2, 8, 9]
    >>> rdd.takeSample(False,5)
    [9, 10, 7, 5, 3]
    >>> rdd.takeSample(False,5)
    [9, 10, 4, 3, 7]
    >>> rdd.takeSample(False,20)
    [9, 5, 2, 7, 4, 6, 1, 3, 10, 8]
    >>> rdd.takeSample(True,20) 
    [8, 6, 1, 8, 3, 2, 5, 2, 5, 8, 9, 5, 8, 5, 6, 4, 10, 2, 1, 1]
    
    >>> rdd.takeSample(False,8,5)
    [3, 4, 2, 1, 9, 8, 7, 6]
    >>> rdd.takeSample(False,8,5)
    [3, 4, 2, 1, 9, 8, 7, 6]
    >>> rdd.takeSample(False,3,5)
    [3, 4, 2]
    >>> rdd.takeSample(False,4,5)
    [3, 4, 2, 1]
    >>> rdd.takeSample(False,5,5)
    [3, 4, 2, 1, 9]

#### 3.4 RDD的重要算子 ####

*  1- 基本函数:

![7855206a31692ad6240630d6502e7dda.png][]

*  2- 分区函数:

在spark中, 对于map算子 和 foreach算子都提供了分区函数, 分别为 mapPartitions() 和 foreachPartition()
    
        map算子和foreach算子是针对RDD中每一个分区下的每一个数据, 而分区函数, 则是针对每个分区的数据
    
    
        假设有一份这样的数据:
            rdd = sc.parallelize([1,2,3,4,5,6,7,8,9,10],3)
    
        查看各分区的结果:
            rdd.glom().collect()
            结果;
            [
                [1, 2, 3], 
                [4, 5, 6], 
                [7, 8, 9, 10]
            ] 
    
        执行相关的操作:
            def fn1(num):
                print(num)
                return num + 1
            rdd.map(fn1)
    
    
        思考: fn1会被触发多少次呢?  10次   在第一个分区触发了3次  第二个分区 触发了3次  第三个分区触发了4次
    
        假设:
            如果在fn1函数中, 执行连接数据库的操作, 然后读取数据库的数据, 处理后, 将连接数据库的连接关闭, 请问这个打开连接和关闭连接会触发多少次呢?  至少会被调用10次 每一次连接和关闭都是要花费时间的. 而且对资源也是一种浪费 造成效率比较底下
    
        思考: 打开连接和关闭连接, 由于业务要求, 必须放置到函数中, 但是又想提升效率, 如何处理呢?
    
        处理方案: 可以采用分区函数来解决问题, 将map替换为mapPartitions  分区函数只针对整个分区来执行处理的
    
        同样还是刚刚这个代码:
            rdd = sc.parallelize([1,2,3,4,5,6,7,8,9,10],3)
    
            查看各分区的结果:
                rdd.glom().collect()
                结果;
                [
                    [1, 2, 3], 
                    [4, 5, 6], 
                    [7, 8, 9, 10]
                ] 
    
            执行相关的操作:
                def fn1(arr):
                    arr2 = []
                    print(arr)
                    for i in arr:
                        arr2.append(i+1)
                    return arr2
    
                rdd.mapPartitions(fn1)
    
           思考: 此时的fn1被调用多少次了呢?  3次(有多少个分区, 就会触发多少次)

演示分区函数:

演示: map 和 mapPartitions:
    rdd = sc.parallelize([1,2,3,4,5,6,7,8,9,10],3)
    
    查看各分区的结果:
        rdd.glom().collect()
        结果;
        [
            [1, 2, 3], 
            [4, 5, 6], 
            [7, 8, 9, 10]
        ] 
    
    执行相关的操作:
    def fn1(num):
        print(num)
       return num + 1
    
    rdd.map(fn1)
    
    
    def fn1(arr):
        arr2 = []
        print(arr)
        for i in arr:
            arr2.append(i+1)
        return arr2
    
    简单写法:
    def fn1(arr):
        for i in arr:
            yield i + 1
    
    rdd.mapPartitions(fn1)
    
    
    
    演示 foreach  和 foreachPartition
    rdd = sc.parallelize([1,2,3,4,5,6,7,8,9,10],3)
    
    查看各分区的结果:
        rdd.glom().collect()
        结果;
        [
            [1, 2, 3], 
            [4, 5, 6], 
            [7, 8, 9, 10]
        ] 
    
    执行遍历操作:
    
    def fn1(num):
        print(num)
    
    
    rdd.foreach(fn1)
    结果:
    4
    5
    6
    1
    2
    3
    7
    8
    9
    10
    
    def fn1(arr):
        for i in arr:
            print(i)
    
    rdd.foreachPartition(fn1)
    结果:
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    
    
    
    
    总结说明:
        建议, 在使用map和foreach的时候, 建议更换为 mapPartitions和 foreachPartition, 尤其是在函数中存在一些与资源相关的操作. 比如 文件操作, 数据库的操作等等....
    
        如果没有资源相关的代码, map 和 foreach 以及 mapPartitions和foreachPartition 基本上效率差不多
    
    
    分区函数:  作用在每一个分区上的
    
    非分区函数:  作用在每一个分区的每一个数据

*  3- 重分区函数
 *  重新对RDD分区数量进行调整

repartition(N):
        作用: 改变RDD的分区数量, 得到一个新的RDD 可以增大分区 也可以减少分区, 都会产生shuffle操作
    
    演示:
    rdd = sc.parallelize([1,2,3,4,5,6,7,8,9,10],3)
    [
        [1, 2, 3], 
        [4, 5, 6], 
        [7, 8, 9, 10]
    ]
    
    尝试增大分区:
    rdd.repartition(5).glom().collect()
    结果:
    [
        [], 
        [1, 2, 3], 
        [7, 8, 9, 10], 
        [4, 5, 6], 
        []
    ]
    
    尝试减少分区:
    rdd.repartition(2).glom().collect()
    结果:
    [
        [1, 2, 3, 7, 8, 9, 10], 
        [4, 5, 6]
    ]
    
    
    coalesce(N) 函数:
        作用: 改变RDD的分区数量, 得到一个新的RDD, 默认只能进行减少分区数量
    
    演示:
        rdd = sc.parallelize([1,2,3,4,5,6,7,8,9,10],3)
        rdd.glom().collect()
        [
            [1, 2, 3], 
            [4, 5, 6], 
            [7, 8, 9, 10]
        ]
    
        尝试减少分区:
        rdd.coalesce(2).glom().collect()
        结果:
        [
            [1, 2, 3], 
            [4, 5, 6, 7, 8, 9, 10]
        ]
    
        尝试增大分区:
        rdd.coalesce(5).glom().collect()
        结果: 默认无法增大分区
        [[1, 2, 3], [4, 5, 6], [7, 8, 9, 10]]
    
        尝试将参数2设置为True:
        参数2: 表示的是否开启shuffle, 如果开启了, 即可实现增大分区, 如果不开启, 仅能减少分区, 默认是关闭shuffle
        rdd.coalesce(5,True).glom().collect()
        结果:
        [[], [1, 2, 3], [7, 8, 9, 10], [4, 5, 6], []]
    
    
    
    说明:
        repartition 本质上是coalesce的一种当参数2为True的简写方案, 因为repartition底层调度coalesce函数, 将参数2设置为True
    
    
    repartition 和 coalesce区别:
        联系:
            repartition底层调用的是coalesce, 只是将coalesce的参数2设置为True
            两个函数都是可以进行重分区的
    
        区别:
            repartition: 既可以增大分区, 也可以减少分区, 触发shuffle
            coalesce: 默认只能减少分区, 无法增大分区, 不触发shuffle, 如果要增大分区, 需要将参数2调整为True
    
    
    专门针对kv类型重分区函数:  partitionBy(N,[FN])
    
        作用: 改变RDD的分区数量, 得到一个新的RDD  可以增大分区 也可以减少分区, 但是会产生shuffle
    
        默认: 根据key进行hash取模划分操作,如果不满意这个分区方案, 可以通过参数2自定义分区规则
            注意:  自定义分区规则返回的必须是回一个int类型的数据, 返回为分区的编号, 编号从0开始
    
    
        演示:
        rdd = sc.parallelize([(1,1),(2,2),(3,3),(4,4),(5,5),(6,6),(7,7),(8,8),(9,9),(10,10)],5)
    
        rdd.glom().collect()
        结果:
        [
            [(1, 1), (2, 2)], 
            [(3, 3), (4, 4)], 
            [(5, 5), (6, 6)], 
            [(7, 7), (8, 8)], 
            [(9, 9), (10, 10)]
        ]
    
        尝试重分区操作:  重新划分为5个分区
        rdd.partitionBy(5).glom().collect()
        结果:
        [
            [(5, 5), (10, 10)], 
            [(1, 1), (6, 6)], 
            [(2, 2), (7, 7)], 
            [(3, 3), (8, 8)], 
            [(4, 4), (9, 9)]
        ]
    
        尝试分为2个分区:
        rdd.partitionBy(2).glom().collect()
        结果:
        [[(2, 2), (4, 4), (6, 6), (8, 8), (10, 10)], [(1, 1), (3, 3), (5, 5), (7, 7), (9, 9)]]
    
        希望实现: >=5 放置在一个分区, 剩余放置到另一个分区
    
        rdd.partitionBy(2,lambda k: 0 if k <=5  else 1).glom().collect()
        结果:
        [
            [(1, 1), (2, 2), (3, 3), (4, 4), (5, 5)], 
            [(6, 6), (7, 7), (8, 8), (9, 9), (10, 10)]
        ]

*  4- 聚合函数:

单值类型的聚合函数:
        格式:
            reduce(fn1):  根据传入函数对数据进行聚合处理
            fold(defaultAgg,fn1): 根据传入函数对数据进行聚合处理, 同时支持给agg设置初始值
            aggregate(defaultAgg,fn1,fn2): 根据传入函数对数据进行聚合处理,参数1 设置agg的初始值, fn2对各个分区内的数据进行聚合计算, fn2 负责将各个分区的聚合结果进行汇总聚合
    
    案例:
        rdd = sc.parallelize([1,2,3,4,5,6,7,8,9,10],3)
        rdd.glom().collect()
        结果:
            [
                [1, 2, 3], 
                [4, 5, 6], 
                [7, 8, 9, 10]
            ]
    
        需求: 求和计算, 求所有数据之和
        def fn1(agg,curr):
            print(agg)
            return agg+curr
    
        rdd.reduce(fn1) 
        结果: 55
    
        agg值 默认为 每个分区的第一个值
    
    
        rdd.fold(5,lambda agg,curr:agg+curr)
        结果: 75 
    
        def fn1(agg,curr):
            return agg + curr
    
            def fn2(a,b):
            return agg + curr
    
        rdd.aggregate(5,fn1,fn2)
    
    说明:
        reduce底层为fold, fold底层为aggregate
    
        在实际使用中 能用reduce解决的, 优先使用reduce, 如果不行, 尝试fold  如果fold也不行, 尝试aggregate
    
    双值类型的聚合函数:

![66ceede350a9071e8a5d9151d6ed6336.png][]

KV类型的聚合函数:
        格式; 
            reduceByKey(fn1)
            foldByKey(defaultAgg,fn1);
            aggregateByKey(defaultAgg,fn1);
    
        以上三个与单值是一样的, 知识在单值的基础上加了分组的操作而已,针对每个分组内的数据进行聚合而已
    
        额外有一个: groupByKey() 仅分组, 不聚合统计
    
    
    思考点: groupByKey() + 聚合操作 和  reduceByKey()  都可以完成分组聚合统计, 谁的效率更高一些?  reduceByKey()

reduceByKey:

![da431ef8d0a3215bc7559014df1a8d9b.png][]

groupByKey:

![4106dd7a7d978ef41b740aba5c3866ed.png][]

*  5- 关联函数

关联函数, 主要是针对kv类型的数据, 根据key进行关联操作
    
    相关的算子:
        join: 实现两个RDD的join关联操作
        leftOuterJoin: 实现两个RDD的左关联操作
        rightOuterJoin: 实现两个RDD的右关联操作
        fullOuterJoin: 实现两个RDD的满外(全外)关联操作
    
    演示:
        rdd1 = sc.parallelize([('c01','张三'),('c02','李四'),('c02','王五'),('c01','赵六'),('c03','田七'),('c03','周八'),('c02','李九'),('c04','老张')])
    
        rdd2 = sc.parallelize([('c01','大数据一班'),('c02','大数据二班'),('c03','大数据三班'),('c05','大数据五班')])
    
         rdd1.join(rdd2).collect()
         结果:
            [
                ('c01', ('张三', '大数据一班')), 
                ('c01', ('赵六', '大数据一班')), 
                ('c02', ('李四', '大数据二班')), 
                ('c02', ('王五', '大数据二班')), 
                ('c02', ('李九', '大数据二班')),
                ('c03', ('田七', '大数据三班')),
                ('c03', ('周八', '大数据三班'))
            ]
    
         rdd1.leftOuterJoin(rdd2).collect()
         结果:
            [
                ('c04', ('老张', None)),
                ('c01', ('张三', '大数据一班')), 
                ('c01', ('赵六', '大数据一班')), 
                ('c02', ('李四', '大数据二班')), 
                ('c02', ('王五', '大数据二班')), 
                ('c02', ('李九', '大数据二班')), 
                ('c03', ('田七', '大数据三班')), 
                ('c03', ('周八', '大数据三班'))
            ]
    
        rdd1.rightOuterJoin(rdd2).collect()
        结果:
            [
                ('c05', (None, '大数据五班')), 
                ('c01', ('张三', '大数据一班')), 
                ('c01', ('赵六', '大数据一班')), 
                ('c02', ('李四', '大数据二班')), 
                ('c02', ('王五', '大数据二班')), 
                ('c02', ('李九', '大数据二班')), 
                ('c03', ('田七', '大数据三班')), 
                ('c03', ('周八', '大数据三班'))
            ]
    
        rdd1.fullOuterJoin(rdd2).collect() 
        结果:
            [
                ('c04', ('老张', None)), 
                ('c05', (None, '大数据五班')), 
                ('c01', ('张三', '大数据一班')), 
                ('c01', ('赵六', '大数据一班')), 
                ('c02', ('李四', '大数据二班')), 
                ('c02', ('王五', '大数据二班')), 
                ('c02', ('李九', '大数据二班')), 
                ('c03', ('田七', '大数据三班')),
                ('c03', ('周八', '大数据三班'))
            ]

[https_spark.apache.org_docs_3.1.2_api_python_reference_pyspark.html_rdd-apis]: https://link.zhihu.com/?target=https%3A//spark.apache.org/docs/3.1.2/api/python/reference/pyspark.html%23rdd-apis
[87154c295626de894eb87cca51d503e9.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/20/f46a4861570b4f4faa7a22dbb6423eaa.png
[e619e288bd85930a765ef76368d134eb.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/20/e5c017d7cb88428ead4a6bc366b4ac5e.png
[7855206a31692ad6240630d6502e7dda.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/20/2b492bbe23af4125b9c366ed09ad8555.png
[66ceede350a9071e8a5d9151d6ed6336.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/20/a186a48a45494b238bad7072554c6f89.png
[da431ef8d0a3215bc7559014df1a8d9b.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/20/fa3afdcab44b4ca480567b5c338a2adf.png
[4106dd7a7d978ef41b740aba5c3866ed.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/20/66e4c5caf5e84daf9bb25995a930952e.png