【Python】PySpark 数据计算 ③ ( RDD#reduceByKey 函数概念 | RDD#reduceByKey 方法工作流程 | RDD#reduceByKey 语法 | 代码示例 )

文章目录

一、RDD#reduceByKey 方法
二、代码示例 - RDD#reduceByKey 方法
- 1、代码示例
- 2、执行结果
三、代码示例 - 使用 RDD#reduceByKey 统计文件内容
- 1、需求分析
- 2、代码示例

一、RDD#reduceByKey 方法

1、RDD#reduceByKey 方法概念

RDD#reduceByKey 方法是 PySpark 中提供的计算方法 ,

首先 , 对键值对 KV 类型 RDD 对象数据中相同键 key 对应的值 value 进行分组 ,
然后 , 按照开发者提供的算子 ( 逻辑 / 函数 ) 进行聚合操作 ;

上面提到的键值对 KV 型的数据 , 指的是二元元组 , 也就是 RDD 对象中存储的数据是二元元组 ;

元组可以看做为只读列表 ;

二元元组指的是元组中的数据 , 只有两个 , 如 :

("Tom", 18)
("Jerry", 12)

PySpark 中 , 将二元元组中

第一个元素称为键 Key ,
第二个元素称为值 Value ;

按照键 Key 分组 , 就是按照二元元组中的第一个元素的值进行分组 ;

[("Tom", 18), ("Jerry", 12), ("Tom", 17), ("Jerry", 13)]

将上述列表中的二元元组进行分组 , 按照二元元组第一个元素进行分组 ,

("Tom", 18) 和 ("Tom", 17) 元组分为一组 , 在这一组中 , 将 18 和 17 两个数据进行聚合 , 如 : 相加操作 , 最终聚合结果是 35 ;
("Jerry", 12) 和 ("Jerry", 13) 分为一组 ;

如果键 Key 有 A, B, C 三个值 Value 要进行聚合 , 首先将 A 和 B 进行聚合得到 X , 然后将 X 与 C 进行聚合得到新的值 Y ;

具体操作方法是 : 先将相同键 key 对应的值 value 列表中的元素进行 reduce 操作 , 返回一个减少后的值，并将该键值对存储在RDD中 ;

2、RDD#reduceByKey 方法工作流程

RDD#reduceByKey 方法工作流程 : reduceByKey(func) ;

首先 , 对 RDD 对象中的数据分区 , 每个分区中的相同键 key 对应的值 value 被组成一个列表 ;
然后 , 对于每个键 key 对应的值 value 列表 , 使用 reduceByKey 方法提供的函数参数 func 进行 reduce 操作 , 将列表中的元素减少为一个 ;
最后 , 将减少后的键值对存储在新的 RDD 对象中 ;

3、RDD#reduceByKey 函数语法

RDD#reduceByKey 语法 :

reduceByKey(func, numPartitions=None)

func 参数 : 用于聚合的函数 ;
numPartitions 是可选参数 , 指定 RDD 对象的分区数 ;

传入的 func 函数的类型为 :

(V, V) -> V

V 是泛型 , 指的是任意类型 , 上面的三个 V 可以是任意类型 , 但是必须是相同的类型 ;

该函数接收两个 V 类型的参数 , 参数类型要相同 , 返回一个 V 类型的返回值 , 传入的两个参数和返回值都是 V 类型的 ;

使用 reduceByKey 方法 , 需要保证函数的

可结合性 ( associativity ) : 将两个具有相同参数类型和返回类型的方法结合在一起 , 不会改变它们的行为的性质 ; 两个方法结合使用的结果与执行顺序无关 ;
可重入性 ( commutativity ) : 在多任务环境下 , 一个方法可以被多个任务调用 , 而不会出现数据竞争或状态错误的问题 ;

以便在并行计算时能够正确地聚合值列表 ;

二、代码示例 - RDD#reduceByKey 方法

1、代码示例

在下面的代码中 , 要处理的数据是列表 , 列表元素是二元元组 ;

[("Tom", 18), ("Tom", 3), ("Jerry", 12), ("Jerry", 21)]

对值 Value 进行的聚合操作就是相加 , 也就是把同一个键 Key 下的多个 Value 值进行相加操作 ,

# 应用 reduceByKey 操作，将同一个 Key 下的 Value 相加
rdd2 = rdd.reduceByKey(lambda a, b: a + b)

代码示例 :

"""
PySpark 数据处理
"""

# 导入 PySpark 相关包
from pyspark import SparkConf, SparkContext
# 为 PySpark 配置 Python 解释器
import os
os.environ['PYSPARK_PYTHON'] = "D:/001_Develop/022_Python/Python39/python.exe"

# 创建 SparkConf 实例对象 , 该对象用于配置 Spark 任务
# setMaster("local[*]") 表示在单机模式下 本机运行
# setAppName("hello_spark") 是给 Spark 程序起一个名字
sparkConf = SparkConf() 
    .setMaster("local[*]") 
    .setAppName("hello_spark")

# 创建 PySpark 执行环境 入口对象
sparkContext = SparkContext(conf=sparkConf)

# 打印 PySpark 版本号
print("PySpark 版本号 : ", sparkContext.version)

# 将 字符串列表 转为 RDD 对象
rdd = sparkContext.parallelize([("Tom", 18), ("Tom", 3), ("Jerry", 12), ("Jerry", 21)])

# 应用 reduceByKey 操作，将同一个 Key 下的 Value 相加
rdd2 = rdd.reduceByKey(lambda a, b: a + b)

# 打印新的 RDD 中的内容
print(rdd2.collect())

# 停止 PySpark 程序
sparkContext.stop()

2、执行结果

D:01_Develop22_PythonPython39python.exe D:/002_Project/011_Python/HelloPython/Client.py
23/08/01 10:16:04 WARN Shell: Did not find winutils.exe: java.io.FileNotFoundException: java.io.FileNotFoundException: HADOOP_HOME and hadoop.home.dir are unset. -see https://wiki.apache.org/hadoop/WindowsProblems
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
23/08/01 10:16:04 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
PySpark 版本号 :  3.4.1
D:01_Develop22_PythonPython39Libsite-packagespysparkpythonlibpyspark.zippysparkshuffle.py:65: UserWarning: Please install psutil to have better support with spilling
D:01_Develop22_PythonPython39Libsite-packagespysparkpythonlibpyspark.zippysparkshuffle.py:65: UserWarning: Please install psutil to have better support with spilling
D:01_Develop22_PythonPython39Libsite-packagespysparkpythonlibpyspark.zippysparkshuffle.py:65: UserWarning: Please install psutil to have better support with spilling
D:01_Develop22_PythonPython39Libsite-packagespysparkpythonlibpyspark.zippysparkshuffle.py:65: UserWarning: Please install psutil to have better support with spilling
D:01_Develop22_PythonPython39Libsite-packagespysparkpythonlibpyspark.zippysparkshuffle.py:65: UserWarning: Please install psutil to have better support with spilling
[('Jerry', 33), ('Tom', 21)]

Process finished with exit code 0

在这里插入图片描述

三、代码示例 - 使用 RDD#reduceByKey 统计文件内容

1、需求分析

给定一个文本文件 word.txt , 文件内容为 :

Tom Jerry
Tom Jerry Tom
Jack Jerry

在这里插入图片描述

读取文件中的内容 , 统计文件中单词的个数 ;

思路 :

先读取数据到 RDD 中 ,
然后按照空格分割开再展平 , 获取到每个单词 ,
根据上述单词列表 , 生成一个二元元组列表 , 列表中每个元素的键 Key 为单词 , 值 Value 为数字 1 ,
对上述二元元组列表进行聚合操作 , 相同的键 Key 对应的值 Value 进行相加 ;

2、代码示例

首先 , 读取文件 , 将文件转为 RDD 对象 , 该 RDD 对象中 , 列表中的元素是字符串类型 , 每个字符串的内容是整行的数据 ;

# 将 文件 转为 RDD 对象
rdd = sparkContext.textFile("word.txt")
# 内容为 ['Tom Jerry', 'Tom Jerry Tom', 'Jack Jerry']

然后 , 通过 flatMap 展平文件, 先按照空格切割每行数据为字符串列表 , 然后展平数据解除嵌套 ;

# 通过 flatMap 展平文件, 先按照 空格 切割每行数据为 字符串 列表
#   然后展平数据解除嵌套
rdd2 = rdd.flatMap(lambda element: element.split(" "))
# 内容为 : ['Tom', 'Jerry', 'Tom', 'Jerry', 'Tom', 'Jack', 'Jerry']

再后 , 将 rdd 数据的列表中的元素转为二元元组 , 第一个元素设置为单词字符串 , 第二个元素设置为 1

# 将 rdd 数据 的 列表中的元素 转为二元元组, 第二个元素设置为 1
rdd3 = rdd2.map(lambda element: (element, 1))
# 内容为 [('Tom', 1), ('Jerry', 1), ('Tom', 1), ('Jerry', 1), ('Tom', 1), ('Jack', 1), ('Jerry', 1)]

最后 , 应用 reduceByKey 操作 , 对相同键 Key 对应的值 Value 进行聚合操作 , 将同一个 Key 下的 Value 相加, 也就是统计键 Key 的个数 ;

# 应用 reduceByKey 操作，
#   将同一个 Key 下的 Value 相加, 也就是统计 键 Key 的个数
rdd4 = rdd3.reduceByKey(lambda a, b: a + b)
# [('Tom', 3), ('Jack', 1), ('Jerry', 3)]

代码示例 :

"""
PySpark 数据处理
"""

# 导入 PySpark 相关包
from pyspark import SparkConf, SparkContext
# 为 PySpark 配置 Python 解释器
import os
os.environ['PYSPARK_PYTHON'] = "D:/001_Develop/022_Python/Python39/python.exe"

# 创建 SparkConf 实例对象 , 该对象用于配置 Spark 任务
# setMaster("local[*]") 表示在单机模式下 本机运行
# setAppName("hello_spark") 是给 Spark 程序起一个名字
sparkConf = SparkConf() 
    .setMaster("local[*]") 
    .setAppName("hello_spark")

# 创建 PySpark 执行环境 入口对象
sparkContext = SparkContext(conf=sparkConf)

# 打印 PySpark 版本号
print("PySpark 版本号 : ", sparkContext.version)

# 将 文件 转为 RDD 对象
rdd = sparkContext.textFile("word.txt")
print("查看文件内容 : ", rdd.collect())

# 通过 flatMap 展平文件, 先按照 空格 切割每行数据为 字符串 列表
#   然后展平数据解除嵌套
rdd2 = rdd.flatMap(lambda element: element.split(" "))
print("查看文件内容展平效果 : ", rdd2.collect())

# 将 rdd 数据 的 列表中的元素 转为二元元组, 第二个元素设置为 1
rdd3 = rdd2.map(lambda element: (element, 1))
print("转为二元元组效果 : ", rdd3.collect())

# 应用 reduceByKey 操作，
#   将同一个 Key 下的 Value 相加, 也就是统计 键 Key 的个数
rdd4 = rdd3.reduceByKey(lambda a, b: a + b)
print("最终统计单词 : ", rdd4.collect())

# 停止 PySpark 程序
sparkContext.stop()

执行结果 :

D:01_Develop22_PythonPython39python.exe D:/002_Project/011_Python/HelloPython/Client.py
23/08/01 11:25:24 WARN Shell: Did not find winutils.exe: java.io.FileNotFoundException: java.io.FileNotFoundException: HADOOP_HOME and hadoop.home.dir are unset. -see https://wiki.apache.org/hadoop/WindowsProblems
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
23/08/01 11:25:24 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
PySpark 版本号 :  3.4.1
查看文件内容 :  ['Tom Jerry', 'Tom Jerry Tom', 'Jack Jerry']
查看文件内容展平效果 :  ['Tom', 'Jerry', 'Tom', 'Jerry', 'Tom', 'Jack', 'Jerry']
转为二元元组效果 :  [('Tom', 1), ('Jerry', 1), ('Tom', 1), ('Jerry', 1), ('Tom', 1), ('Jack', 1), ('Jerry', 1)]
D:01_Develop22_PythonPython39Libsite-packagespysparkpythonlibpyspark.zippysparkshuffle.py:65: UserWarning: Please install psutil to have better support with spilling
D:01_Develop22_PythonPython39Libsite-packagespysparkpythonlibpyspark.zippysparkshuffle.py:65: UserWarning: Please install psutil to have better support with spilling
D:01_Develop22_PythonPython39Libsite-packagespysparkpythonlibpyspark.zippysparkshuffle.py:65: UserWarning: Please install psutil to have better support with spilling
D:01_Develop22_PythonPython39Libsite-packagespysparkpythonlibpyspark.zippysparkshuffle.py:65: UserWarning: Please install psutil to have better support with spilling
最终统计单词 :  [('Tom', 3), ('Jack', 1), ('Jerry', 3)]

Process finished with exit code 0

在这里插入图片描述