Python第二语言（十三、PySpark实战）

1.开篇

2. PySpark介绍

3. PySpark基础准备

3.1 PySpark安装

3.2 掌握PySpark执行环境入口对象的构建

3.3 理解PySpark的编程模型

4. PySpark：RDD对象数据输入

4.1 RDD对象概念：PySpark支持多种数据的输入，完成后会返回RDD类的对象；

4.2 Python数据容器转RDD对象.parallelize(数据容器对象)

4.3 RDD存在很多计算的方法

4.4 读取文件转RDD对象：通过SparkContext入口对象来读取文件，构建RDD对象；

5. PySpark：RDD对象数据计算（一）

5.1 给Spark设置环境变量（不设置的时候，控制台会报错，出现找不到python.exe解释器的情况）

5.2 RDD的map方法：将RDD的数据根据函数进行一条条处理

5.3 RDD的flatMap方法：基本和map一样，但是多一个功能：将嵌套list给转成单list；[[1, 2, 3], [4, 5, 6]]转成[1, 2, 3, 4, 5, 6]

5.4 RDD的reduceByKey方法：将key分组后进行value逻辑处理；

6. 数据计算案例（一）：完成使用PySpark进行单词技术的案例

7. PySpark：RDD对象数据计算（二）

7.1 RDD的filter方法：传入T泛型数据，返回bool，为false 的数据丢弃，为true的数据保留；（函数对RDD数据逐个处理，得到True的保留至返回值的RDD中）

7.2 RDD的distinct方法：对RDD数据进行去重，返回新RDD；

7.3 RDD的sortBy方法：对RDD的容器按照指定规则排序，返回新RDD；

8. 数据计算案例（二）：计算城市中的商品以及销售额

8.1 需求

8.2 文件数据

8.3 需求一实现：处理结果自动返回的是一个二元元组；

8.4 需求二实现：将字典中的数据处理，返回一个list；

8.5 需求三实现：过滤除北京的数据，并只返回一个参数category，是list列表，并进行去重，去重后的结果进行collect输出；

9. 将RDD的结果数据输出为Python对象的各类方法

导航：

Python第二语言（一、Python start）-CSDN博客

Python第二语言（二、Python语言基础）-CSDN博客

Python第二语言（三、Python函数def）-CSDN博客

Python第二语言（四、Python数据容器）-CSDN博客

Python第二语言（五、Python文件相关操作）-CSDN博客

Python第二语言（六、Python异常）-CSDN博客

Python第二语言（七、Python模块）-CSDN博客

Python第二语言（八、Python包）-CSDN博客

Python第二语言（九、Python第一阶段实操）-CSDN博客

Python第二语言（十、Python面向对象（上））-CSDN博客

Python第二语言（十一、Python面向对象（下））-CSDN博客

Python第二语言（十二、SQL入门和实战）-CSDN博客

Python第二语言（十三、PySpark实战）-CSDN博客

Python第二语言（十四、高阶基础）-CSDN博客

1.开篇

PySpark大数据计算第三方库，Spark是大数据开发的核心技术；
python的spark中使用map时 Python worker exited unexpectedly (crashed)：
- 将原本的python12解释器降低版本到python10版本解释器，降低python解释器版本，因为版本不兼容；
- 记得下载使用的包；

2. PySpark介绍

Apache Spark是用于大规模数据（large-scala data）处理的统一（unifield）分析引擎；
Spark是一款分布式的计算框架，用于调度成百上千的服务器集群，计算TB、PB乃至EB级别的海量数据；
Python On Spark：Python语言，是Spark重点支持的方向；

PySpark第三方库：

PySpark是由Spark官方开发的Python语言第三方库；
Python开发者可以使用pip程序快速安装PySpark并像其它第三方库一样使用；
主要作用：
- 进行数据处理；
- 提交至Spark集群，进行分布式集群计算；

3. PySpark基础准备

3.1 PySpark安装

安装命令： pip install pyspark

加速下载命令：pip install -i https://pypi.tuna.tsinghua.edu.cn/simple pyspark

3.2 掌握PySpark执行环境入口对象的构建

PySpark是分布式集群的操作，setMaster(xxx).\setAppName(xxx)是用来控制集群的代码，图中代码用的是单机的；
setAppName是Spark任务的名称；
PySpark的执行环境入口对象是：类SparkContext的类对象，所有PySpark的功能都是从SparkContext对象作为开始；

# 导包
from pyspark import SparkConf, SparkContext

# 创建SparkConf类对象
conf = SparkConf().setMaster("local[*]").\
    setAppName("test_spark_app")

# 基于SparkConf类对象创建SparkContext类对象
sc = SparkContext(conf=conf)

# 打印PySpark的运行版本
print(sc.version)

# 停止SparkContext对象的运行
sc.stop()

3.3 理解PySpark的编程模型

SparkContext类对象，是PySpark编程中一切功能的入口；

PySpark的编程三大步骤：
1. 数据输入：通过SparkContex类对象的成员方法完成数据的读取操作，读取后得到RDD类对象；
2. 数据处理计算：通过RDD类对象的成员方法，完成各种数据计算的需求；
3. 数据输出：将处理完成后的RDD对象，调用各种成员方法完成，写出文件，转换位list等操作；

4. PySpark：RDD对象数据输入

RDD就是PySpark计算后返回的对象容器；

4.1 RDD对象概念：PySpark支持多种数据的输入，完成后会返回RDD类的对象；

RDD全称为：弹性分布式数据集（Resilient Distributed Datasets）；

PySpark针对数据的处理，都是以RDD对象作为载体；
1. 数据存储在RDD内；
2. 各类数据的计算方法，也都是RDD的成员方法；
3. RDD的数据计算方法，返回值依旧是RDD对象；
比如说JSON文件、文本文件、数据库数据，都是可以通过SparkContext类对象，经过RDD对象的处理，并返回给文件文件或JSON文件，或者数据库；

4.2 Python数据容器转RDD对象.parallelize(数据容器对象)

提示：
1. 字符串会被拆分出1个个的字符，存入RDD对象；
2. 字典仅有key会被存入RDD对象；
3. RDD对象返回的是容器，与list一样结果；

from pyspark import SparkConf, SparkContext

conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app")
sc = SparkContext(conf=conf)

# 通过parallelize方法将Python对象加载到Spark内，称为RDD对象
rdd1 = sc.parallelize([1, 2, 3, 4, 5])
rdd2 = sc.parallelize((1, 2, 3, 4, 5))
rdd3 = sc.parallelize("abcdefg")
rdd4 = sc.parallelize({1, 2, 3, 4, 5})
rdd5 = sc.parallelize({"key1": "value1", "key2": "value2"})

# 使用collect方法查看RDD中的内容
print(rdd1.collect())
print(rdd2.collect())
print(rdd3.collect())
print(rdd4.collect())
print(rdd5.collect())

sc.stop()

4.3 RDD存在很多计算的方法

4.4 读取文件转RDD对象：通过SparkContext入口对象来读取文件，构建RDD对象；

from pyspark import SparkConf, SparkContext

conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app")
sc = SparkContext(conf=conf)

# 通过textFile方法，读取文件数据加载到Spark内，成为RDD对象
rdd = sc.textFile("dataText")

# 打印RDD内容
print(rdd.collect())
sc.stop()

小结：

RDD对象称之为分布式弹性数据集，是PySpark中数据计算的载体，可以：
1. 提供数据存储；
2. 提供数据计算的各类方法；
3. 数据计算的方法，返回值依旧是RDD（RDD迭代计算）；

5. PySpark：RDD对象数据计算（一）

可以对list容器计算，可以对dict字典容器计算，可以对str字符串进行计算，所有的容器都可以通过RDD计算；

5.1 给Spark设置环境变量（不设置的时候，控制台会报错，出现找不到python.exe解释器的情况）

os.path.exists 返回值为True或False；

from pyspark import SparkConf, SparkContext
import os

# 配置Spark环境变量
os.environ['PYSPARK_PYTHON'] = 'C:/Users/Administrator/AppData/Local/Programs/Python/Python310/python.exe'

# 检查PYSPARK_PYTHON路径
print(os.path.exists('C:/Users/Administrator/AppData/Local/Programs/Python/Python310/python.exe'))
# 检查PYSPARK_DRIVER_PYTHON路径
print(os.path.exists('C:/Users/Administrator/AppData/Local/Programs/Python/Python310/python.exe'))

5.2 RDD的map方法：将RDD的数据根据函数进行一条条处理

1. 介绍：

RDD对象内置丰富的：成员方法（算子）
map算子：是将RDD的数据一条条处理（处理的逻辑是将python中的函数作为参数进行传递，这个函数，参数会将RDD种的每条数据都进行处理）最终返回一个新的RDD对象；
- map()中的参数 (T) → U：T代表传入一个参数，U代表一个返回值；（意思代表传入的参数是一个，还有一个返回值，T是泛型，不用指定数据类型）
- map()中的参数 (T) → T：T代表传入一个参数，T代表一个返回值；（意思代表传入的参数是一个，还有一个返回值，T是泛型，传入的是什么值，那么返回的就是什么类型）

2. func函数传递：

func函数作为参数：代表的是RDD中的每个值，都会进行func函数的处理；是RDD中的每一个元素都会被RDD处理一遍；

可以简写成：rdd2 = rdd.map(lambda x: x * 10) # 简写的函数 ；

3. 案例：

这里存在一个大坑，如果是python312版本去使用map函数，会报错 Python worker exited unexpectedly (crashed) ，降低版本即可，我用的版本10；
结果：RDD中的每一个元素都会被传递给func进行处理，*10操作；

from pyspark import SparkConf, SparkContext
import os

# 配置Spark环境变量
os.environ['PYSPARK_PYTHON'] = 'C:/Users/Administrator/AppData/Local/Programs/Python/Python310/python.exe'
conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app")

sc = SparkContext(conf=conf)

# 准备一个RDD
rdd = sc.parallelize([1, 2, 3, 4, 5])
rdd2 = rdd.map(lambda x: x * 10)  # 简写的函数
print(rdd2.collect())
sc.stop()

4. map链式调用：

from pyspark import SparkConf, SparkContext
import os

# 配置Spark环境变量
os.environ['PYSPARK_PYTHON'] = 'C:/Users/Administrator/AppData/Local/Programs/Python/Python310/python.exe'
conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app")

sc = SparkContext(conf=conf)

rdd = sc.parallelize([1, 2, 3, 4, 5])

rdd2 = rdd.map(lambda x: x * 10).map(lambda x: x + 5)  # 链式调用：将map进行第一个*10数据计算，再进行map+5数据计算

print(rdd2.collect())

5. 小结：

map算子（成员方法）：
- 接受一个处理函数，可用lambda表达式快速编写；
- 对RDD内的元素逐个处理，并返回一个新的RDD；
链式调用：对于返回值是新RDD的算子，可以通过链式调用的方式多次调用算子；

5.3 RDD的flatMap方法：基本和map一样，但是多一个功能：将嵌套list给转成单list；[[1, 2, 3], [4, 5, 6]]转成[1, 2, 3, 4, 5, 6]

from pyspark import SparkConf, SparkContext
import os

if __name__ == '__main__':
    os.environ['PYSPARK_PYTHON'] = 'C:/Users/Administrator/AppData/Local/Programs/Python/Python310/python.exe'
    conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app")
    sc = SparkContext(conf=conf)

    rdd = sc.parallelize(["zhangSan lisi yiyi", "zhangSan yiyi wangWu", "wangWu yiyi zhangSan"])

    print(rdd.map(lambda x: x.split(" ")).collect())
    print("-----------------------------------------")
    print(rdd.flatMap(lambda x: x.split(" ")).collect())  # 将嵌套list转成单list，对数据接触嵌套

5.4 RDD的reduceByKey方法：将key分组后进行value逻辑处理；

二元元组：[('a', 1), ('a', 1), ('b', 1)]这就是二元元组，元组中只有两个元素；
自动按照key分组，完成组内数据（value）的聚合操作：就是会按照元组中的key，就是'a', 'a', 'b'进行key的value聚合，1, 1, 1是value；（value聚合的逻辑是，按照传入的func函数逻辑来进行聚合）

假设这是二元元组数据要进行reduceByKey算子处理：

reduceByKey计算方式：

1. 思路：

先分组，key值等于a和a一组，b和b一组：然后在进行函数lambda a, b: a+b进行处理，也即是分组后，a=a+a, b=b+b+b；结果[('b', 3), ('a', 2)]
再解释：b有三个值，那么lambda a, b: a+b中表示的是b：1, 1, 1 的三个值，去进行函数处理的时候，先是第一个1和第二1进行相加，这时候相加是a+b，分组后与key无关系，那么第一个1和第二个1相加后等于2，这时候发现还有第三个1，这时候再次把第一次相加的结果，与第三个1进行a+b处理，2+1；是前后者参数的相加处理；最终得到按照key分组聚合value的结果；
最终解释：将数据分组后，每个组的数据进行lambda a, b: a + b 操作，每个组中的数据，进行a + b操作，意思就是将当前组的所有value进行相加操作；

2. 实现：

功能：针对KV型RDD，自动按照key分组，然后根据提供的聚合逻辑，完成组内数据（value）的聚合操作；

rdd.reduceByKey(func)：

from pyspark import SparkConf, SparkContext
import os

if __name__ == '__main__':
    os.environ['PYSPARK_PYTHON'] = 'C:/Users/Administrator/AppData/Local/Programs/Python/Python310/python.exe'
    conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app")
    sc = SparkContext(conf=conf)

    rdd = sc.parallelize([('a', 1), ('a', 1), ('b', 1), ('b', 1), ('b', 1)])

    result = rdd.reduceByKey(lambda a, b: a + b)  # 分组计算
    print(result.collect())