Python学习从0到1 day26 第三阶段 Spark ② 数据计算Ⅰ
人总是会执着于失去的，而又不珍惜现在所拥有的
—— 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.exe"
conf = SparkConf().setMaster("local[*]").setAppName("test_spark")
sc = SparkContext(conf=conf)
# 准备一个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 * 10
rdd2 = rdd.map(multi)
print("rdd2：",rdd2.collect())
# 匿名函数链式调用
# 将每一个数乘以100再加上7再减去114
rdd3 = rdd.map(lambda x:x*100).map(lambda x:x+7).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.exe"
conf = SparkConf().setMaster("local[*]").setAppName("test_spark")
sc = SparkContext(conf=conf)
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:(V，V)→V
# 接受2个传入参数(类型要一致)，返回一个返回值，类型和传入要求一致reduceByKey的聚合逻辑是：
比如，有[1，2，3，4，5]，然后聚合函数是：lambda a，b：a + b
将容器中的所有元素进行聚合
语法：
from pyspark import SparkConf,SparkContext
# 设置spark中的python解释器对象
import os
os.environ['PYSPARK_PYTHON'] = "E:/python.learning/pyt/scripts/python.exe"
conf = SparkConf().setMaster("local[*]").setAppName("test_spark")
sc = SparkContext(conf=conf)
# 准备一个二元元组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 this，and this gives life to thee.
代码
将所有单词都转换成二元元组，单词为key，value设置为1，value表示每个单词出现的次数，作为value，初始化为1，若单词相等，则表示key相同，value值进行累加
from pyspark import SparkConf,SparkContext
# 设置spark中的python解释器对象
import os
os.environ['PYSPARK_PYTHON'] = "E:/python.learning/pyt/scripts/python.exe"
conf = SparkConf().setMaster("local[*]").setAppName("test_spark")
sc = SparkContext(conf=conf)
# 读取数据文件
rdd = sc.textFile("D:/2LFE\Desktop\WordCount.txt")
# 取出全部单词
word_rdd = rdd.flatMap(lambda x:x.split(" "))
print(word_rdd.collect())
# 将所有单词都转换成二元元组，单词为key，value设置为1，value表示每个单词出现的次数，作为value，
# 若单词相等，则表示value相同，key值进行累加
word_with_one_rdd = word_rdd.map(lambda word:(word,1))
# 分组并求和
result_rdd = word_with_one_rdd.reduceByKey(lambda a,b:a+b)
# 打印并输出结果
print(result_rdd.collect())