pyspark入门基础详细讲解

1.前言介绍

学习目标：了解什么是Speak、PySpark，了解为什么学习PySpark，了解课程是如何和大数据开发方向进行衔接

使用pyspark库所写出来的代码，既可以在电脑上简单运行，进行数据分析处理，又可以把代码无缝迁移到成百上千的服务器集群上去做分布式计算。

为什么要学习pyspark呢？

总结

2.基础准备

学习目标：掌握pyspark库的安装，掌握pyspark执行环境入口对象的构建，理解pyspark的编程模型。

建议使用国内代理镜像网站下载更快。

简化代码，本质上是同一个意思，链式结构，链式调用化简程序基本原则，就是我不管调用什么方法，我的返回值都是同一个对象啊

代码展示：
"""
演示获取pyspark的执行环境入库对象：SparkContext
并通过SparkContext对象获取当前PySpark的版本
"""

# 导包
from pyspark import SparkConf,SparkContext
# 创建SparkConf类对象  setMaster是描写运行模式   setAppName是设置当前Spark任务的名字
conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app")
# 同一个意思，链式结构，链式调用化简程序
# 基本原则，就是我不管调用什么方法，我的返回值都是同一个对象啊
# 基于SparkConf类对象创建SparkContext对象
sc = SparkContext(conf=conf)
# 打印PySpark的运行版本
print(sc.version)
# 停止SparkContext对象的运行（停止PySpark程序）
sc.stop()

spark需要启动时间，所以代码的运行一小会，3.5.3就是当前spark的运行版本

这个sc非常非常重要哦，后续给大家讲解。

通过sc拿到数据输入，数据处理计算是通过RDD类对象的一系列成员方法来对数据进行计算，然后把结果对外进行输出

我们只需要记住后期写spark代码的三大步，把数据加载进来，对数据进行计算，把结果输出去

总结

3.数据输入

学习目标：理解RDD对象，掌握PySpark数据输入的2种方法。

RDD就和列表等数据容器差不多

python数据容器转RDD对象

parallelize成员方法把数据容器存入RDD对象

如果要查看RDD里面有什么内容，需要用collect()方法

字符串会把每一个字符都拆出来，存入RDD对象，字典仅有key被存入RDD对象

读取文件转RDD对象

总结

4.数据计算

map方法

学习目标：掌握RDD的map方法

map会把传入的每一个参数都返回一个值

你会发现报错了，报错的原因是spark没有找到python解释器

给他指定一条路径，这样就没有问题了。如果指定路径之后还是没有解决的，可能是因为pycharm版本太新，降低版本就行了，建议是pycharm3.10

对于简单函数我们可以使用lambda匿名函数。

结果是一样的

链式调用

总结

flatMap方法

学习目标：掌握RDD的flatMap方法对数据进行计算。

通过map，可以看到尽管我们把数据分成一个一个的，但是还是存在嵌套，依旧被嵌套在list当中

当我们使用了flatMap方法后，发现解除了嵌套

from pyspark import SparkConf,SparkContext
import os
os.environ['PYSPARK_PYTHON'] ="D:/python/venv/Scripts/python.exe"
conf = SparkConf().setMaster("local[*]").setAppName("test_spark")
sc = SparkContext(conf=conf)

# 准备一个RDD
rdd = sc.parallelize(["ikun 22","ikun 3","ik unh hhhh"])
# 需求：将RDD数据里面的一个个单词提取出来
rdd2 = rdd.flatMap(lambda x: x.split(" ")) # 使用空格进行切分
print(rdd2.collect())

使用flatMap可以解除内部嵌套，语法与map一样

总结：

reduceByKey方法

学习目标：掌握RDD的reduceByKey方法

二元元组指的是元组里面存储的只有两个元素

KV型的RDD一般是两个元素，把第一个元素当成key，第二个当成value，自动按照key分组，然后根据你传入的逻辑计算value

（v,v)->(v) 意思是传入两个相同类型的参数，返回一个返回值，类型和传入要求一致

自动分组并且组内求和

总结

可以完成按key进行分组，并且组内进行逻辑计算

练习案例1

学习目标：完成使用PySpark进行单词计数的案例

数据文件

取出所有的单词，flatMap是把单词一个一个取出来，map是把单词一行一行取出来，一行是一个列表。

把单词转换成二元元组

完整代码

"""
完成练习案例：单词计数统计
"""
from pyspark import SparkConf,SparkContext
import os
# 1.构建执行环境入口对象
os.environ['PYSPARK_PYTHON'] ="D:/python/venv/Scripts/python.exe"
conf = SparkConf().setMaster("local[*]").setAppName("test_spark")
sc = SparkContext(conf=conf)
# 2.读取数据文件
rdd = sc.textFile("D:/word.txt")
# 3.取出全部单词
word_rdd = rdd.flatMap(lambda x: x.split(" "))
# 4.将所有单词都转换成二元元组，单词为key，value设置为1
# (hello,1) (spark,1) (itheima,1) (itcast,1)
word_with_one_rdd = word_rdd.map(lambda word: (word,1))
# 5.分组并求和
result_rdd = word_with_one_rdd.reduceByKey(lambda a,b:a+b)
# 6.打印输出结果
print(result_rdd.collect())

filter方法

学习目标：掌握RDD的filter方法

True被保留，False被丢弃

总结

distinct方法

学习目标：掌握RDD的distinct方法

不需要传入参数，功能简单就是去重操作

总结

sortBy方法

学习目标：掌握RDD的sortBy方法进行内容的排序

接收函数传入参数并且有一个返回值

目前我们没有解除到分布式，就先写上numPartitions=1

之前写过一个读取文件，统计单词的个数，现在让我们对他进行排序

可以自己控制升序或者降序，True升序，False降序

from pyspark import SparkConf,SparkContext
import os
# 1.构建执行环境入口对象
os.environ['PYSPARK_PYTHON'] ="D:/python/venv/Scripts/python.exe"
conf = SparkConf().setMaster("local[*]").setAppName("test_spark")
sc = SparkContext(conf=conf)
# 2.读取数据文件
rdd = sc.textFile("D:/word.txt")
# 3.取出全部单词
word_rdd = rdd.flatMap(lambda x: x.split(" "))
# 4.将所有单词都转换成二元元组，单词为key，value设置为1
# (hello,1) (spark,1) (itheima,1) (itcast,1)
word_with_one_rdd = word_rdd.map(lambda word: (word,1))
# 5.分组并求和
result_rdd = word_with_one_rdd.reduceByKey(lambda a,b:a+b)
# 6.对结果进行排序
final_rdd = result_rdd.sortBy(lambda x:x[1],ascending=False,numPartitions=1)
print(final_rdd.collect())

总结：

练习案例2

学习目标：完成练习案例2的开发

完整代码：

"""
完成练习案例：json商品统计
"""
# 1.各个城市销售额排名，从小到大
# 2.全部城市，有哪些商品类别在售卖
# 3.北京市有哪些商品类别在售卖
from pyspark import SparkConf,SparkContext
import os
import json
os.environ['PYSPARK_PYTHON'] ='D:/python/venv/Scripts/python.exe'
conf = SparkConf().setMaster("local[*]").setAppName("test_spark")
sc = SparkContext(conf=conf)
# 需求1：城市销售额排名
# 1.1 读取文件得到RDD
file_rdd = sc.textFile("D:/2222.txt")
# 1.2 取出一个个json字符串
json_str_rdd = file_rdd.flatMap(lambda x: x.split("|"))

# 1.3 将一个个json字符串转换为字典
dict_rdd = json_str_rdd.map(lambda x: json.loads(x))
# 1.4 取出城市和销售额数据
city_with_money_rdd = dict_rdd.map(lambda x: (x['areaName'],int(x['money'])))
# 1.5 按城市分组按销售额聚合
city_result_rdd = city_with_money_rdd.reduceByKey(lambda a,b:a+b)
# 1.6 按销售额聚合结果进行排序
result1_rdd = city_result_rdd.sortBy(lambda x:x[1],ascending=False,numPartitions=1)
print("需求1的结果：",result1_rdd.collect())
# 需求2：全部城市有哪些商品类别在售卖
# 2.1 取出全部的商品类别
# 2.2 对全部商品类别进行去重
category_rdd = dict_rdd.map(lambda x: x['category']).distinct()
print("需求2的结果：",category_rdd.collect())
# 需求3：北京市有哪些商品类别在售卖
# 3.1 过滤北京市的数据
beijing_data_rdd = dict_rdd.filter(lambda x: x['areaName'] == '北京')
# 3.2 取出全部商品类别
# 3.3 进行商品类别去重
result3_rdd = beijing_data_rdd.map(lambda x: x['category']).distinct()
print("需求3的结果：",result3_rdd.collect())

5.数据输出

输出为python对象

学习目标：掌握将RDD的结果输出为python对象的各类方法

collect算子

reduce算子

reduce和reducebykey的区别是reducebykey是获取key然后组内计算，reduce是单纯的直接计算

take算子

就是取前N个元素

count算子

总结

from pyspark import SparkConf,SparkContext
import os
import json
os.environ['PYSPARK_PYTHON'] ='D:/python/venv/Scripts/python.exe'
conf = SparkConf().setMaster("local[*]").setAppName("test_spark")
sc = SparkContext(conf=conf)
# 准备RDD
rdd = sc.parallelize([1,2,3,4,5])
# collect算子，输出RDD为list对象
rdd_list:list = rdd.collect()
print(rdd_list)
print(type(rdd_list))
# reduce算子，对RDD进行两两融合
num = rdd.reduce(lambda a,b: a+b)
print(num)
# take算子，取出RDD前N个元素，组成list返回
take_list = rdd.take(3)
print(take_list)
# count，统计rdd内有多少条数据，返回值为数字
num_count = rdd.count()
print(f"rdd内有{num_count}个元素")
sc.stop()