兰州建设工程信息网站,做收益的网站多少钱,视频模板套用免费,做译员的网站RDD基本介绍
什么是RDD
RDD:英文全称Resilient Distributed Dataset#xff0c;叫做弹性分布式数据集#xff0c;是Spark中最基本的数据抽象#xff0c;代表一个不可变、可分区、里面的元素可并行计算的集合。
Resilient弹性#xff1a;RDD的数据可以存储在内存或者磁盘…RDD基本介绍
什么是RDD
RDD:英文全称Resilient Distributed Dataset叫做弹性分布式数据集是Spark中最基本的数据抽象代表一个不可变、可分区、里面的元素可并行计算的集合。
Resilient弹性RDD的数据可以存储在内存或者磁盘当中RDD的数据可以分区Distributed分布式RDD的数据可以分布式存储可以进行并行计算Dataset数据集一个用于存放数据的集合
RDD的五大特征
1、必须的RDD是由一系列分区组成的 2、必须的对RDD做计算相当于对RDD的每个分区做计算 3、必须的RDD之间存在着依赖关系宽依赖和窄依赖 4、可选的对于KV类型的RDD我们可以进行自定义分区方案 5、可选的移动数据不如移动计算让计算程序离数据越近越好
RDD的五大特点
1、分区RDD逻辑上是分区的仅仅是定义分区的规则并不是直接对数据进行分区操作因为RDD本身不存储数据。 2、只读RDD是只读的要想改变RDD中的数据只能在现有的RDD基础上创建新的RDD。 3、依赖RDD之间存在着依赖关系宽依赖和窄依赖 4、缓存如果在应用程序中多次使用同一个RDD可以将该RDD缓存起来该RDD只有在第一次计算的时候会根据血缘关系得到分区的数据 5、checkpoint与缓存类似的都是可以将中间某一个RDD的结果保存起来只不过checkpoint支持持久化保存
如何构建RDD
构建RDD对象的方式主要有两种
1、通过 textFiledata: 通过读取外部文件的方式来初始化RDD对象实际工作中经常使用。
2、通过 parallelizedata: 通过自定义列表的方式初始化RDD对象。一般用于测试并行化本地集合方式
from pyspark import SparkConf, SparkContext
import os# 绑定指定的Python解释器
os.environ[SPARK_HOME] /export/server/spark
os.environ[PYSPARK_PYTHON] /root/anaconda3/bin/python3
os.environ[PYSPARK_DRIVER_PYTHON] /root/anaconda3/bin/python3if __name__ __main__:print(并行化本地集合创建RDD)# 1- 创建SparkContext对象conf SparkConf().setAppName(parallelize_rdd).setMaster(local[1])sc SparkContext(confconf)# 2- 数据输入# 并行化本地集合得到RDDinit_rdd sc.parallelize([1,2,3,4,5], numSlices6)# 3- 数据处理# 4- 数据输出# 获取分区数print(init_rdd.getNumPartitions())# 获取具体分区内容print(init_rdd.glom().collect())# 5- 释放资源sc.stop()读取外部数据源方式
from pyspark import SparkConf, SparkContext
import os# 绑定指定的Python解释器
os.environ[SPARK_HOME] /export/server/spark
os.environ[PYSPARK_PYTHON] /root/anaconda3/bin/python3
os.environ[PYSPARK_DRIVER_PYTHON] /root/anaconda3/bin/python3if __name__ __main__:print(读取文件创建RDD)# 1- 创建SparkContext对象conf SparkConf().setAppName(textfile_rdd).setMaster(local[1])sc SparkContext(confconf)# 2- 数据输入# 读取文件得到RDDinit_rdd sc.textFile(file:///export/data/gz16_pyspark/01_spark_core/data/content.txt,minPartitions4)# 3- 数据处理# 4- 数据输出# 获取分区数print(init_rdd.getNumPartitions())# 获取具体分区内容print(init_rdd.glom().collect())# 5- 释放资源sc.stop()处理小文件的操作
常规处理小文件的办法:
1- 大数据框架提供的现有的工具或者命令1.1- hadoop fs -getmerge /input/small_files/*.txt /output/merged_file.txt1.2- hadoop archive -archiveName myhar.har -p /small_files /big_files
2- 可以通过编写自定义的代码将小文件读取进来在代码中输出的时候输出形成大的文件wholeTextFiles: 读取小文件。1-支持本地文件系统和HDFS文件系统。参数minPartitions指定最小的分区数。2-通过该方式读取文件会尽可能使用少的分区数可能会将多个小文件的数据放到同一个分区中进行处理。3-一个文件要完整的存放在一个元组中也就是不能将一个文件分成多个进行读取。文件是不可分割的。4-RDD分区数量既受到minPartitions参数的影响同时受到小文件个数的影响RDD分区数量如何确定 1- RDD的分区数量一般设置为机器CPU核数的2-3倍。为了充分利用服务器的硬件资源2- RDD的分区数据量受到多个因素的影响例如:机器CPU的核数、调用的算子、算子中参数的设置、集群的类型等。RDD具体有多少个分区直接通过getNumPartitions查看3- 当初始化SparkContext对象的时候其实就确定了一个参数spark.default.parallelism默认为CPU的核数。如果是本地集群就取决于local[num]中设置的数字大小如果是集群默认至少有2个分区4- 通过parallelize来构建RDD如果没有指定分区数默认就取spark.default.parallelism参数值如果指定了分区数也就是numSlices参数那么numSlices的优先级会更高一些最终RDD的分区数取该参数的值。5- 通过textFile来构建RDD5.1- 首先确认defaultMinPartition参数的值。该参数的值如果没有指定textFile的minPartition参数那么就根据公式min(spark.default.parallelism,2)如果有指定textFile的minPartition参数那么就取设置的值5.2- 再根据读取文件所在的文件系统的不同来决定最终RDD的分区数5.2.1- 本地文件系统: RDD分区数 max(本地文件分片数,defaultMinPartition)5.2.2- HDFS文件系统: RDD分区数 max(文件block块的数量,defaultMinPartition)RDD相关算子
RDD算子 指的是RDD对象中提供了非常多的具有特殊功能的函数 我们将这些函数称为算子函数/方法/API 相关的算子的官方文档 https://spark.apache.org/docs/3.1.2/api/python/reference/pyspark.html#rdd-apis
RDD算子的分类
整个RDD算子 共分为两大类
Transformation(转换算子): 返回值: 是一个新的RDD特点: 转换算子只是定义数据的处理规则并不会立即执行是lazy惰性的。需要由Action算子触发Action(动作算子):返回值: 要么没有返回值None或者返回非RDD类型的数据特点: 动作算子都是立即执行。执行的时候会将它上游的其他算子一同触发执行相关转换算子: 相关的动作算子:
RDD的转换算子
(单)值类型算子
map算子 格式rdd.map(fn) 说明: 主要根据传入的函数对数据进行一对一的转换操作传入一行返回一行
输入: init_rdd sc.parallelize([0,1,2,3,4,5,6,7,8,9])
需求: 数字加一后返回
代码: init_rdd.map(lambda num:num1).collect()
结果: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]groupBy算子: 格式: groupBy(fn)说明: 根据用户传入的自定义函数对数据进行分组操作
输入: init_rdd sc.parallelize([0,1,2,3,4,5,6,7,8,9])
需求: 将数据分成奇数和偶数
代码: init_rdd.groupBy(lambda num:偶数 if num%20 else 奇数).mapValues(list).collect()
结果: [(偶数, [0, 2, 4, 6, 8]), (奇数, [1, 3, 5, 7, 9])]
总结: mapValues(list)将数据类型转成List列表filter算子: 格式filter(fn)说明根据用户传入的自定义函数对数据进行过滤操作。自定义函数的返回值类型是bool类型。True表示满足过滤条件会将数据保留下来False会将数据丢弃掉
输入: init_rdd sc.parallelize([0,1,2,3,4,5,6,7,8,9])
需求: 过滤掉数值3的数据
代码: init_rdd.filter(lambda num:num3).collect()
结果: [4, 5, 6, 7, 8, 9]flatMap算子: 格式rdd.flatMap(fn)说明在map算子的基础上加入一个压扁的操作 主要适用于一行中包含多个内容的操作实现一转多的操作
输入: init_rdd sc.parallelize([张三 李四 王五,赵六 周日])
需求: 将姓名一个一个的输出
代码: init_rdd.flatMap(lambda line:line.split()).collect()
结果: [张三, 李四, 王五, 赵六, 周日]
说明: split()默认会按照空白字符对内容进行切分处理。例如空格、制表符、回车。还是推荐明确指定你所需要分割的符号。双值类型算子
union(并集) 和intersection(交集) 格式: rdd1.union(rdd2) rdd1.intersection(rdd2)
输入: rdd1 sc.parallelize([3,3,2,6,8,0])rdd2 sc.parallelize([3,2,1,5,7])并集: rdd1.union(rdd2).collect()
结果: [3, 3, 2, 6, 8, 0, 3, 2, 1, 5, 7]
说明: union取并集不会对重复出现的数据去重对并集的结果进行去重: rdd1.union(rdd2).distinct().collect()
结果: [8, 0, 1, 5, 2, 6, 3, 7]
说明: distinct()是转换算子用来对RDD中的元素进行去重处理交集: rdd1.intersection(rdd2).collect()
结果: [2, 3]
说明: 交集会对结果数据进行去重处理key-value数据类型算子 groupByKey() 格式: rdd.groupByKey()说明: 对键值对类型的RDD中的元素按照键key进行分组操作。只会进行分组
输入: rdd sc.parallelize([(c01,张三),(c02,李四),(c02,王五),(c01,赵六),(c03,田七),(c03,周八),(c02,李九)])
需求: 对学生按照班级分组统计
代码: rdd.groupByKey().mapValues(list).collect()
结果: [(c01, [张三, 赵六]), (c02, [李四, 王五, 李九]), (c03, [田七, 周八])]reduceByKey() 格式: rdd.reduceByKey(fn)说明: 根据key进行分组将一个组内的value数据放置到一个列表中对这个列表基于fn进行聚合计算操作
输入: rdd sc.parallelize([(c01,张三),(c02,李四),(c02,王五),(c01,赵六),(c03,田七),(c03,周八),(c02,李九)])
需求: 统计每个班级学生人数
代码: rdd.map(lambda tup:(tup[0],1)).reduceByKey(lambda agg,curr:aggcurr).collect()
结果: [(c01, 2), (c02, 3), (c03, 2)]sortByKey()算子: 格式rdd.sortByKey(ascendingTrue|False)说明: 根据key进行排序操作默认按照key进行升序排序如果需要降序设置 ascending 参数的值为False
输入: 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)]需求: 根据key进行排序操作演示降序
代码: rdd.sortByKey(ascendingFalse).collect()
结果: [(15, 3), (12, 4), (10, 2), (8, 4), (7, 4), (2, 4)]输入: rdd sc.parallelize([(a01,2),(A01,3),(a011,2),(a03,2),(a021,2),(a04,2)])
需求: 根据key进行排序操作演示升序
代码: rdd.sortByKey().collect()
结果: [(A01, 3), (a01, 2), (a011, 2), (a021, 2), (a03, 2), (a04, 2)]
总结: 对字符串类型的key进行排序的时候按照ASCII码表进行排序。大写字母排在小写字母的前面如果前缀一样短的排在前面长的排在后面。RDD的动作算子 collect() 算子: 格式: collect() 作用: 收集各个分区的数据将数据汇总到一个大的列表返回 reduce() 算子: 格式: reduce(fn)作用: 根据用户传入的自定义函数对数据进行聚合操作。该算子是Action动作算子而reduceByKey是Transformation转换算子。
输入: rdd sc.parallelize([1,2,3,4,5,6,7,8,9,10])
需求: 统计所有元素之和是多少
代码:
def mysum(agg,curr):print(f中间临时聚合结果{agg}当前遍历到的元素{curr})return aggcurrrdd.reduce(mysum)rdd.reduce(lambda agg,curr:aggcurr)
结果:
中间临时聚合结果6当前遍历到的元素7
中间临时聚合结果13当前遍历到的元素8
中间临时聚合结果21当前遍历到的元素9
中间临时聚合结果30当前遍历到的元素10
中间临时聚合结果1当前遍历到的元素2
中间临时聚合结果3当前遍历到的元素3
中间临时聚合结果6当前遍历到的元素4
中间临时聚合结果10当前遍历到的元素5
中间临时聚合结果15当前遍历到的元素40
55说明: 初始化的时候agg表示中间临时聚合结果默认取列表中的第一个元素值curr表示当前遍历到的元素默认取列表中的第二个元素的值。first() 算子: 格式: rdd.first()说明: 取RDD中的第一个元素。不会对RDD中的数据排序
输入: rdd sc.parallelize([3,1,2,4,5,6,7,8,9,10])
需求: 获取第一个元素
代码: rdd.first()
结果: 3take() 算子
格式: rdd.take(N)说明: 取RDD中的前N元素。不会对RDD中的数据排序
输入: rdd sc.parallelize([3,1,2,4,5,6,7,8,9,10])
需求: 获取前3个元素
代码: rdd.take(3)
结果: [3, 1, 2]
说明: 返回结果是List列表。必须要传递参数N而且不能是负数。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),(c04,1),(c03,4)])
需求: 按照班级人数降序排序取前2个
代码: rdd.top(2,keylambda tup:tup[1])
结果: [(c02, 8), (c01, 5)]需求: 按照班级人数升序排序取前2个
代码: rdd.top(2,keylambda tup:-tup[1])
结果: [(c04, 1), (c03, 4)]count() 算子 说明统计RDD中一共有多少个元素
输入: rdd sc.parallelize([1,2,3,4,5,6,7,8,9,10])
需求: 获取一共有多少个元素
代码: rdd.count()
结果: 10foreach() 算子 格式: foreach(fn)作用: 遍历RDD中的元素对元素根据传入的函数进行自定义的处理
输入: rdd sc.parallelize([1,2,3,4,5,6,7,8,9,10])
需求: 对数据进行遍历打印
代码: rdd.foreach(lambda num:print(num))
结果:
6
7
8
9
10
1
2
3
4
5
说明: 1- foreach()算子对自定义函数不要求有返回值另外该算子也没有返回值2- 因为底层是多线程运行的因此输出结果分区间可能是乱序3- 该算子一般用来对结果数据保存到数据库或者文件中RDD的重要算子
基本算子 分区算子
分区算子针对整个分区数据进行处理的算子。 mapPartitions和foreachPartition 说明map和foreach算子都有对应的分区算子。分区算子适用于有反复消耗资源的操作例如文件的打开和关闭、数据库的连接和关闭等能够减少操作的次数。
输入: 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]]演示: map和mapPartitions
需求: 对数字加一
map
自定义函数:
def my_add(num):print(f传递进来的数据{num})return num1rdd.map(my_add).collect()
结果:
传递进来的数据4
传递进来的数据5
传递进来的数据6
传递进来的数据1
传递进来的数据2
传递进来的数据3
传递进来的数据7
传递进来的数据8
传递进来的数据9
传递进来的数据10
[2, 3, 4, 5, 6, 7, 8, 9, 10, 11]请问: my_add被调用了几次
回答: 10
弊端: 会导致消耗资源的操作反复多次的执行非常消耗资源
def my_add(num):# 打开数据库连接# 将数据保存到数据库# 关闭数据库连接print(f传递进来的数据{num})return num1mapPartitions
自定义函数:
def my_add(list):print(输入的参数,list)new_list []for i in list:new_list.append(i 1)return new_listrdd.mapPartitions(my_add).collect()
结果:
输入的参数 itertools.chain object at 0x7ff21ae9d940
输入的参数 itertools.chain object at 0x7ff21ae9d940
输入的参数 itertools.chain object at 0x7ff21ae94e50
[2, 3, 4, 5, 6, 7, 8, 9, 10, 11]演示: foreach和foreachPartition
需求: 遍历打印
foreach
自定义函数:
def my_print(num):print(f传递进来的数据{num})print(num)rdd.foreach(my_print)
结果:
传递进来的数据1
1
传递进来的数据2
2
传递进来的数据3
3
传递进来的数据4
4
传递进来的数据5
5
传递进来的数据6
6
传递进来的数据7
7
传递进来的数据8
8
传递进来的数据9
9
传递进来的数据10
10foreachPartition
自定义函数:
def my_print(list):print(f传递进来的数据{list})for i in list:print(i)rdd.foreachPartition(my_print)结果:
传递进来的数据itertools.chain object at 0x7ff21ae9d2b0
1
2
3
传递进来的数据itertools.chain object at 0x7ff21ae9d2b0
4
5
6
传递进来的数据itertools.chain object at 0x7ff21ae94a60
7
8
9
10总结: 1- map和foreach算子都有对应的分区算子分别是mapPartitions和foreachPartition2- 分区算子适用于有反复消耗资源的操作例如文件的打开和关闭、数据库的连接和关闭等能够减少操作的次数。3- 如果没有反复消耗资源的操作调用两类算子效果一样重分区算子
重分区算子对RDD的分区重新进行分区操作的算子也就是改变RDD分区数的算子。
repartition算子 格式repartition(num)作用改变RDD分区数。既能够增大RDD分区数也能够减小RDD分区数。但是都会导致发生Shuffle过程。
输入: 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.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算子 格式coalesce(num,shuffleTrue|False)作用改变RDD分区数。但是默认只能减小RDD分区数不能增大减小过程中不会发生Shuffle过程。如果想增大分区需要将参数shuffle设置为True但是会导致Shuffle过程。
输入: 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再增大分区: rdd.coalesce(5,shuffleTrue).glom().collect()
结果: [[], [1, 2, 3], [7, 8, 9, 10], [4, 5, 6], []]将参数2设置为True再减小分区: rdd.coalesce(2,shuffleTrue).glom().collect()
结果: [[1, 2, 3, 7, 8, 9, 10], [4, 5, 6]]repartition 和 coalesce总结1- 这两个算子都是用来改变RDD的分区数2- repartition 既能够增大RDD分区数也能够减小RDD分区数。但是都会导致发生Shuffle过程。3- 默认只能减小RDD分区数不能增大减小过程中不会发生Shuffle过程。如果想增大分区需要将参数shuffle设置为True但是会导致Shuffle过程。4- repartition 底层实际上是调用了coalesce算子并且将shuffle参数设置为了TruepartitionBy算子 格式partitionBy(num,[fn])作用该算子主要是用来改变key-value键值对数据类型RDD的分区数的。num表示要设置的分区数fn参数是可选用来让用户自定义分区规则。
注意
默认情况下根据key进行Hash取模分区。
如果对默认分区规则不满意可以传递参数fn来自定义分区规则。
但是自定义分区规则函数需要满足两个条件
条件一分区编号的数据类型需要是int类型
条件二传递给自定义分区函数的参数是key输入: 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)]]需求: 增大分区尝试分为20个分区
代码: rdd.partitionBy(20).glom().collect()
结果: [[], [(1, 1)], [(2, 2)], [(3, 3)], [(4, 4)], [(5, 5)], [(6, 6)], [(7, 7)], [(8, 8)], [(9, 9)], [(10, 10)], [], [], [], [], [], [], [], [], []]需求: 减少分区尝试分为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)]]需求: 将 key5 放置在一个分区剩余放置到另一个分区
代码: rdd.partitionBy(2,partitionFunclambda key:0 if key5 else 1).glom().collect()
结果: [[(6, 6), (7, 7), (8, 8), (9, 9), (10, 10)], [(1, 1), (2, 2), (3, 3), (4, 4), (5, 5)]]
注意: 分区编号的数据类型需要是int类型聚合算子
单值类型的聚合算子 reduce(fn1)根据传入函数对数据进行聚合处理fold(defaultAgg,fn1)根据传入函数对数据进行聚合处理同时支持给agg设置初始值aggregate(defaultAgg, fn1, fn2)根据传入函数对数据进行聚合处理。defaultAgg设置agg的初始值fn1对各个分区内的数据进行聚合计算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]]
需求: 求和计算, 求所有数据之和reduce
代码:
def my_sum(agg,curr):return aggcurrrdd.reduce(my_sum)
结果: 55fold
代码:
def my_sum(agg,curr):return aggcurrrdd.fold(5,my_sum)
结果: 75aggregate
代码:
def my_sum_1(agg,curr):return aggcurrdef my_sum_2(agg,curr):return aggcurrrdd.aggregate(5,my_sum_1,my_sum_2)
结果: 75总结
reduce、fold、aggregate算子都能实现聚合操作。reduce的底层是foldfold底层是aggregate。在工作中如果能够通过reduce实现的就优先选择reduce否则选择fold实在不行就选择aggregateKV类型的聚合函数 相关的算子 reduceByKey(fn1)foldByKey(defaultAgg, fn1)aggregateByKey(defaultAgg, fn1, fn2);
以上三个与单值是一样的只是在单值的基础上加了分组的操作而已针对每个分组内的数据进行聚合而已。另外有一个groupByKey() 仅分组不聚合统计
问题groupByKey() 聚合操作 和 reduceByKey() 都可以完成分组聚合统计谁的效率更高一些? reduceByKey()因为底层会进行局部的聚合操作会减小后续处理的数据量关联算子
关联函数主要是针对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,1),(c02,2),(c03,3),(c05,5)])join
代码: rdd1.join(rdd2).collect()
结果:
[(c01, (张三, 1)), (c01, (赵六, 1)), (c02, (李四, 2)), (c02, (王五, 2)), (c02, (李九, 2)), (c03, (田七, 3)), (c03, (周八, 3))
]leftOuterJoin
代码: rdd1.leftOuterJoin(rdd2).collect()
结果:
[(c04, (老张, None)), (c01, (张三, 1)), (c01, (赵六, 1)), (c02, (李四, 2)), (c02, (王五, 2)),(c02, (李九, 2)),(c03, (田七, 3)), (c03, (周八, 3))
]rightOuterJoin
代码: rdd1.rightOuterJoin(rdd2).collect()
结果:
[(c05, (None, 5)), (c01, (张三, 1)), (c01, (赵六, 1)), (c02, (李四, 2)), (c02, (王五, 2)),(c02, (李九, 2)), (c03, (田七, 3)), (c03, (周八, 3))
]fullOuterJoin
代码: rdd1.fullOuterJoin(rdd2).collect()
结果:
[(c04, (老张, None)), (c05, (None, 5)),(c01, (张三, 1)), (c01, (赵六, 1)), (c02, (李四, 2)),(c02, (王五, 2)), (c02, (李九, 2)),(c03, (田七, 3)),(c03, (周八, 3))
]
