网站建设亿码酷专注,app开发公司上海,唐山网站制作服务公司,高校网站建设制度人总是会执着于失去的#xff0c;而又不珍惜现在所拥有的 —— 24.11.9 一、map方法
PySpark的数据计算#xff0c;都是基于RDD对象来进行的#xff0c;采用依赖进行#xff0c;RDD对象内置丰富的成员方法#xff08;算子#xff09;
map算子
功能#xff1a;map算子… 人总是会执着于失去的而又不珍惜现在所拥有的 —— 24.11.9 一、map方法
PySpark的数据计算都是基于RDD对象来进行的采用依赖进行RDD对象内置丰富的成员方法算子
map算子
功能map算子是将RDD的数据一条条处理处理的逻辑基于map算子中接收的处理函数)返回新的RDD
语法:
from pyspark import SparkConf,SparkContext# 设置spark中的python解释器对象
import os
os.environ[PYSPARK_PYTHON] E:/python.learning/pyt/scripts/python.execonf SparkConf().setMaster(local[*]).setAppName(test_spark)
sc SparkContext(confconf)# 准备一个RDD对象
rdd sc.parallelize([1,2,3,4,5,6,7,8,9])
# 通过map方法将全部的数据乘以10
# 能够接受一个函数并且将函数作为参数传递进去
# 方法1接受一个匿名函数lambda
rdd1 rdd.map(lambda x:x*10)
print(rdd1:,rdd1.collect())# 方法2接受一个函数
def multi(x):return x * 10rdd2 rdd.map(multi)
print(rdd2,rdd2.collect())# 匿名函数链式调用
# 将每一个数乘以100再加上7再减去114
rdd3 rdd.map(lambda x:x*100).map(lambda x:x7).map(lambda x:x-114)
print(rdd3:,rdd3.collect()) 注
map算子可以通过lambda匿名函数进行链式调用处理复杂的功能 二、flatMap方法
flatMap算子
计算逻辑和map一样
比map多出解除一层嵌套的功能 功能
对rdd执行map操作然后进行 解除嵌套 操作 用法
from pyspark import SparkConf,SparkContext# 设置spark中的python解释器对象
import os
os.environ[PYSPARK_PYTHON] E:/python.learning/pyt/scripts/python.execonf SparkConf().setMaster(local[*]).setAppName(test_spark)
sc SparkContext(confconf)rdd sc.parallelize([一切都会解决 回头看,轻舟已过万重山 一切都会好的,我一直相信])# 需求将RDD数据里面的一个个单词提取出来
rdd1 rdd.map(lambda x:x.split( ))
print(rdd1, rdd2.collect())rdd2 rdd.flatMap(lambda x:x.split( ))
print(rdd2, rdd3.collect()) 注
计算逻辑和map一样比map多出解除一层嵌套的功能 三、reduceByKey方法
reduceByKey算子
功能:
① 自动分组针对KV型二元元组RDD,自动按照 key 分组
② 分组聚合接受一个处理函数根据你提供的聚合逻辑完成组内数据 (valve) 的聚合操作. 用法
rdd.reduceByKey(func)
# func:(VV)→V
# 接受2个传入参数(类型要一致)返回一个返回值类型和传入要求一致
reduceByKey的聚合逻辑是
比如有[12345]然后聚合函数是lambda aba b
将容器中的所有元素进行聚合 语法
from pyspark import SparkConf,SparkContext# 设置spark中的python解释器对象
import os
os.environ[PYSPARK_PYTHON] E:/python.learning/pyt/scripts/python.execonf SparkConf().setMaster(local[*]).setAppName(test_spark)
sc SparkContext(confconf)# 准备一个二元元组rdd对象
rdd sc.parallelize([(男,99),(男,88),(女,99),(男,77),(女,88)])# 求男生和女生两个组的成绩之和
rdd2 rdd.reduceByKey(lambda x , y : x y)
print(rdd2.collect()) 注
1.reduceByKey算子接受一个处理函数对数据进行两两计算 四、WordCount案例
使用PySpark进行单词计数的案例
读取文件统计文件内单词的出现数量
WordCount文件 So long as men can breathe or eyes can see, So long lives thisand this gives life to thee. 代码
将所有单词都转换成二元元组单词为keyvalue设置为1value表示每个单词出现的次数作为value初始化为1若单词相等则表示key相同value值进行累加
from pyspark import SparkConf,SparkContext# 设置spark中的python解释器对象
import os
os.environ[PYSPARK_PYTHON] E:/python.learning/pyt/scripts/python.execonf SparkConf().setMaster(local[*]).setAppName(test_spark)
sc SparkContext(confconf)# 读取数据文件
rdd sc.textFile(D:/2LFE\Desktop\WordCount.txt)
# 取出全部单词
word_rdd rdd.flatMap(lambda x:x.split( ))
print(word_rdd.collect())
# 将所有单词都转换成二元元组单词为keyvalue设置为1value表示每个单词出现的次数作为value
# 若单词相等则表示value相同key值进行累加
word_with_one_rdd word_rdd.map(lambda word:(word,1))
# 分组并求和
result_rdd word_with_one_rdd.reduceByKey(lambda a,b:ab)
# 打印并输出结果
print(result_rdd.collect())
