• 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对象作为开始；
    1. # 导包
    2. from pyspark import SparkConf, SparkContext
    3.  
    4. # 创建SparkConf类对象
    5. conf = SparkConf().setMaster("local[*]").\
    6.     setAppName("test_spark_app")
    7.  
    8. # 基于SparkConf类对象创建SparkContext类对象
    9. sc = SparkContext(conf=conf)
    10.  
    11. # 打印PySpark的运行版本
    12. print(sc.version)
    13.  
    14. # 停止SparkContext对象的运行
    15. 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一样结果；
    1. from pyspark import SparkConf, SparkContext
    2.  
    3. conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app")
    4. sc = SparkContext(conf=conf)
    5.  
    6. # 通过parallelize方法将Python对象加载到Spark内，称为RDD对象
    7. rdd1 = sc.parallelize([1, 2, 3, 4, 5])
    8. rdd2 = sc.parallelize((1, 2, 3, 4, 5))
    9. rdd3 = sc.parallelize("abcdefg")
    10. rdd4 = sc.parallelize({1, 2, 3, 4, 5})
    11. rdd5 = sc.parallelize({"key1": "value1", "key2": "value2"})
    12.  
    13. # 使用collect方法查看RDD中的内容
    14. print(rdd1.collect())
    15. print(rdd2.collect())
    16. print(rdd3.collect())
    17. print(rdd4.collect())
    18. print(rdd5.collect())
    19.  
    20. sc.stop()

    4.3 RDD存在很多计算的方法

    4.4 读取文件转RDD对象：通过SparkContext入口对象来读取文件，构建RDD对象；
    1. from pyspark import SparkConf, SparkContext
    2.  
    3. conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app")
    4. sc = SparkContext(conf=conf)
    5.  
    6. # 通过textFile方法，读取文件数据加载到Spark内，成为RDD对象
    7. rdd = sc.textFile("dataText")
    8.  
    9. # 打印RDD内容
    10. print(rdd.collect())
    11. 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；

    1. from pyspark import SparkConf, SparkContext
    2. import os
    3.  
    4. # 配置Spark环境变量
    5. os.environ['PYSPARK_PYTHON'] = 'C:/Users/Administrator/AppData/Local/Programs/Python/Python310/python.exe'
    6.  
    7. # 检查PYSPARK_PYTHON路径
    8. print(os.path.exists('C:/Users/Administrator/AppData/Local/Programs/Python/Python310/python.exe'))
    9. # 检查PYSPARK_DRIVER_PYTHON路径
    10. 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操作；
    1. from pyspark import SparkConf, SparkContext
    2. import os
    3.  
    4. # 配置Spark环境变量
    5. os.environ['PYSPARK_PYTHON'] = 'C:/Users/Administrator/AppData/Local/Programs/Python/Python310/python.exe'
    6. conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app")
    7.  
    8. sc = SparkContext(conf=conf)
    9.  
    10. # 准备一个RDD
    11. rdd = sc.parallelize([1, 2, 3, 4, 5])
    12. rdd2 = rdd.map(lambda x: x * 10)  # 简写的函数
    13. print(rdd2.collect())
    14. sc.stop()

    4. map链式调用：

    1. from pyspark import SparkConf, SparkContext
    2. import os
    3.  
    4. # 配置Spark环境变量
    5. os.environ['PYSPARK_PYTHON'] = 'C:/Users/Administrator/AppData/Local/Programs/Python/Python310/python.exe'
    6. conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app")
    7.  
    8. sc = SparkContext(conf=conf)
    9.  
    10. rdd = sc.parallelize([1, 2, 3, 4, 5])
    11.  
    12. rdd2 = rdd.map(lambda x: x * 10).map(lambda x: x + 5)  # 链式调用：将map进行第一个*10数据计算，再进行map+5数据计算
    13.  
    14. print(rdd2.collect())

    5. 小结：

    1. map算子（成员方法）：
      • 接受一个处理函数，可用lambda表达式快速编写；
      • 对RDD内的元素逐个处理，并返回一个新的RDD；
    2. 链式调用：对于返回值是新RDD的算子，可以通过链式调用的方式多次调用算子；
    5.3 RDD的flatMap方法：基本和map一样，但是多一个功能：将嵌套list给转成单list；[[1, 2, 3], [4, 5, 6]]转成[1, 2, 3, 4, 5, 6]
    1. from pyspark import SparkConf, SparkContext
    2. import os
    3.  
    4. if __name__ == '__main__':
    5.     os.environ['PYSPARK_PYTHON'] = 'C:/Users/Administrator/AppData/Local/Programs/Python/Python310/python.exe'
    6.     conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app")
    7.     sc = SparkContext(conf=conf)
    8.  
    9.     rdd = sc.parallelize(["zhangSan lisi yiyi", "zhangSan yiyi wangWu", "wangWu yiyi zhangSan"])
    10.  
    11.     print(rdd.map(lambda x: x.split(" ")).collect())
    12.     print("-----------------------------------------")
    13.     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)： 
      1. from pyspark import SparkConf, SparkContext
      2. import os
      3.  
      4. if __name__ == '__main__':
      5.     os.environ['PYSPARK_PYTHON'] = 'C:/Users/Administrator/AppData/Local/Programs/Python/Python310/python.exe'
      6.     conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app")
      7.     sc = SparkContext(conf=conf)
      8.  
      9.     rdd = sc.parallelize([('a', 1), ('a', 1), ('b', 1), ('b', 1), ('b', 1)])
      10.  
      11.     result = rdd.reduceByKey(lambda a, b: a + b)  # 分组计算
      12.     print(result.collect())

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

    • 题目：读取文件，求出文件中单词出现的次数；
    • 文件：

    • 思路：

      先将字符串进行读取，然后按照空格分割['key', 'key']，在进行分割后的数组重组为（key, 1） 的形式，后面利用rdd的reduceByKey方法，将分组后的key，进行聚合操作，因为value都是1，所以可以得出对单词出现的次数，进行统计操作；

    • 根据 (key, 1) 重组后的数据应该是：

      [('key1', 1), ['key1', 1], ('key2', 1), ['key2', 1]]

    • 然后得出最终结果： 
      1. from pyspark import SparkConf, SparkContext
      2. import os
      3.  
      4. if __name__ == '__main__':
      5.     os.environ['PYSPARK_PYTHON'] = 'C:/Users/Administrator/AppData/Local/Programs/Python/Python310/python.exe'
      6.     conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app")
      7.     sc = SparkContext(conf=conf)
      8.  
      9.     # 1.读取数据文件
      10.     """
      11.     假设你有一个大文件，里面有 300MB 的数据，如果你指定分区数为 3，Spark 会尝试将这个文件分成 3 个分区，每个分区大约 100MB。
      12.     如果你的集群有 3 个节点，每个节点可以并行处理一个分区，这样就可以更快地完成任务。
      13.     """
      14.     file = sc.textFile("word", 3)  # ("xx" , 3)：3是指文件被分成的最小分区数（partitions）
      15.  
      16.     # 2.将所有单词读取出来
      17.     words = file.flatMap(lambda line: line.split(' '))  # 结果：['python', 'java', ...]
      18.  
      19.     # 3.将所有单词加1做value
      20.     word_one = words.map(lambda x: (x, 1))  # 结果：[('python', 1), ('java', 1), ('php', 1), ('c#', 1),...]
      21.  
      22.     # 4.分组并求和
      23.     result = word_one.reduceByKey(lambda a, b: a + b)
      24.  
      25.     # 5.打印结果
      26.     print(result.collect())

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

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

    • 功能：过滤想要的数据进行保留；
    • filter算子作用：
      • 接受一个处理函数，可用lambda快速编写；
      • 函数对RDD数据逐个处理，得到True的保留至返回值的RDD中；
    1. from pyspark import SparkConf, SparkContext
    2. import os
    3.  
    4. if __name__ == '__main__':
    5.     os.environ['PYSPARK_PYTHON'] = 'C:/Users/Administrator/AppData/Local/Programs/Python/Python310/python.exe'
    6.     conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app")
    7.     sc = SparkContext(conf=conf)
    8.  
    9.     rdd = sc.parallelize([1, 2, 3, 4, 5])
    10.  
    11.     # 保留基数
    12.     print(rdd.filter(lambda x: x % 2 == 1).collect())

    7.2 RDD的distinct方法：对RDD数据进行去重，返回新RDD；
    1. from pyspark import SparkConf, SparkContext
    2. import os
    3.  
    4. if __name__ == '__main__':
    5.     os.environ['PYSPARK_PYTHON'] = 'C:/Users/Administrator/AppData/Local/Programs/Python/Python310/python.exe'
    6.     conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app")
    7.     sc = SparkContext(conf=conf)
    8.  
    9.     rdd = sc.parallelize([1, 1, 2, 3, 4, 5, 4, 5])
    10.  
    11.     # 对rdd对象进行去重
    12.     print(rdd.distinct().collect())

    7.3 RDD的sortBy方法：对RDD的容器按照指定规则排序，返回新RDD；
    • func: (T) → U：告知按照rdd中的哪个数据进行排序，比如lambda x: x[1] 表示按照rdd中的第二列元素进行排序；
    • numPartitions：目前默认就为1；

    结果：

    按照元组tople中的第二位元素进行排序，按照降序；

    lambda x: x[1]：计算规则，将所有容器的每一个元素按照函数规则处理，x是遍历的元组，x[1]是传入的元组的第二位元素，所以规则就是按照元组的第二位元素进行降序排序；

    1. from pyspark import SparkConf, SparkContext
    2. import os
    3.  
    4. if __name__ == '__main__':
    5.     os.environ['PYSPARK_PYTHON'] = 'C:/Users/Administrator/AppData/Local/Programs/Python/Python310/python.exe'
    6.     conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app")
    7.     sc = SparkContext(conf=conf)
    8.  
    9.     rdd = sc.parallelize([("zhangSan", 99), ("lisi", 88), ("wangWu", 100)])
    10.  
    11.     # 对结果进行排序
    12.     final_rdd = rdd.sortBy(lambda x: x[1], ascending=False, numPartitions=1)
    13.     print(final_rdd.collect())

    • sortBy算子小结：
      • 接收一个处理函数，可用lambda快速编写；
      • 函数表示用来决定排序的依据；
      • 可以控制升序或降序；
      • 全局排序需要设置分区数为1；

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

    8.1 需求

    1. 需求一：各个城市销售额排名，从大到小；

      先按行读取文件，并对json进行split分割，按照|符号，得到最终的字典，使用Spark.reduceByKey进行分组，分组时传递func计算函数，将所有分组后的城市销售额进行a+b的形式，聚合起来，最终得到结果，并按照降序的排序方式排序输出；

    2. 需求二：全部城市，有哪些商品类别在售卖；

      文件读取后，将城市的categpry商品类别，distinct使用去重；

    3. 需求三：北京市有哪些商品类别在售卖；

      将除了北京市的所有数据进行filter过滤，过滤后只留下category并进行去重得到结果；

    8.2 文件数据

    1. {"id":1,"timestamp":"2024-06-01T01:03.00Z","category":"电脑","areaName":"杭州","money":"3000"}|{"id":2,"timestamp":"2024-06-01T01:03.00Z","category":"电脑","areaName":"杭州","money":"3500"}
    2. {"id":3,"timestamp":"2024-06-01T01:03.00Z","category":"食品","areaName":"杭州","money":"3000"}|{"id":4,"timestamp":"2024-06-01T01:03.00Z","category":"食品","areaName":"杭州","money":"3700"}
    3. {"id":5,"timestamp":"2024-06-01T01:03.00Z","category":"服饰","areaName":"北京","money":"3000"}|{"id":6,"timestamp":"2024-06-01T01:03.00Z","category":"服饰","areaName":"北京","money":"3900"}
    8.3 需求一实现：处理结果自动返回的是一个二元元组；
    1. from pyspark import SparkConf, SparkContext
    2. import os
    3.  
    4. if __name__ == '__main__':
    5.     os.environ['PYSPARK_PYTHON'] = 'C:/Users/Administrator/AppData/Local/Programs/Python/Python310/python.exe'
    6.     conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app")
    7.     sc = SparkContext(conf=conf)
    8.  
    9.     # 1.读取文件得到RDD
    10.     file_rdd = sc.textFile("orders")
    11.     # 2. 取出一个个JSON字符串
    12.     json_str_rdd = file_rdd.flatMap(lambda x: x.split("|"))
    13.     # 3. 将一个个JSON字符串转换为字典
    14.     dict_rdd = json_str_rdd.map(lambda x: json.loads(x))
    15.  
    16.     # print(dict_rdd.collect())
    17.  
    18.     # 4.取出城市和销售额数据
    19.     city_with_money_rdd = dict_rdd.map(lambda x: (x['areaName'], int(x['money'])))
    20.  
    21.     # 5.按城市分组按销售额聚合
    22.     city_result_rdd = city_with_money_rdd.reduceByKey(lambda a, b: a + b)
    23.  
    24.     # 6.按销售额聚合结果进行排序
    25.     result_rdd = city_result_rdd.sortBy(lambda x: x[1], ascending=False, numPartitions=1)
    26.     print("需求1的结果：", result_rdd.collect())

    前三步数据结果：

    完整数据结果：

    8.4 需求二实现：将字典中的数据处理，返回一个list；
    1. from pyspark import SparkConf, SparkContext
    2. import os
    3.  
    4. if __name__ == '__main__':
    5.     os.environ['PYSPARK_PYTHON'] = 'C:/Users/Administrator/AppData/Local/Programs/Python/Python310/python.exe'
    6.     conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app")
    7.     sc = SparkContext(conf=conf)
    8.  
    9.     # 1.读取文件得到RDD
    10.     file_rdd = sc.textFile("orders")
    11.     # 2. 取出一个个JSON字符串
    12.     json_str_rdd = file_rdd.flatMap(lambda x: x.split("|"))
    13.     # 3. 将一个个JSON字符串转换为字典
    14.     dict_rdd = json_str_rdd.map(lambda x: json.loads(x))
    15.     # 4.取出全部的商品类别
    16.     category_rdd = dict_rdd.map(lambda x: x['category']).distinct()
    17.     print("需求2的结果：", category_rdd.collect())

    8.5 需求三实现：过滤除北京的数据，并只返回一个参数category，是list列表，并进行去重，去重后的结果进行collect输出；
    1. from pyspark import SparkConf, SparkContext
    2. import os
    3.  
    4. if __name__ == '__main__':
    5.     os.environ['PYSPARK_PYTHON'] = 'C:/Users/Administrator/AppData/Local/Programs/Python/Python310/python.exe'
    6.     conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app")
    7.     sc = SparkContext(conf=conf)
    8.  
    9.     # 1.读取文件得到RDD
    10.     file_rdd = sc.textFile("orders")
    11.     # 2. 取出一个个JSON字符串
    12.     json_str_rdd = file_rdd.flatMap(lambda x: x.split("|"))
    13.     # 3. 将一个个JSON字符串转换为字典
    14.     dict_rdd = json_str_rdd.map(lambda x: json.loads(x))
    15.  
    16.     # 4. 过滤北京的数据
    17.     beijing_data_rdd = dict_rdd.filter(lambda x: x['areaName'] == '北京')
    18.     # 5.取出全部商品类别
    19.     result_rdd = beijing_data_rdd.map(lambda x: x['category']).distinct()
    20.     print("需求3的结果：", result_rdd.collect())

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

    • 数据输出：将RDD输出的值转成文件或Python对象；
    • collect算子：将各个分区内的数据，统一收集到Driver中，形成一个list对象；
    • reduce算子：对RDD数据集按照你传入的逻辑进行聚合；
    • task算子：取出RDD的前N个元素，组合成list返回；
    • count算子：计算RDD有多少条数据，返回值是一个数字；
    1. from pyspark import SparkConf, SparkContext
    2. import os
    3.  
    4. os.environ['PYSPARK_PYTHON'] = 'C:/Users/Administrator/AppData/Local/Programs/Python/Python310/python.exe'
    5. conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app")
    6. sc = SparkContext(conf=conf)
    7.  
    8. if __name__ == '__main__':
    9.     rdd = sc.parallelize([1, 2, 3, 4, 5])
    10.  
    11.     # collect算子，输出RDD为list对象
    12.     rdd_list: list = rdd.collect()
    13.     print("collect算子结果：", rdd_list)
    14.     print("collect算子类型是：", type(rdd_list))
    15.  
    16.     # reduce算子，对RDD进行两两聚合
    17.     num = rdd.reduce(lambda a, b: a + b)
    18.     print("reduce算子结果：", num)
    19.  
    20.     # take算子，取出RDD前N个元素，组成list返回
    21.     take_list = rdd.take(3)
    22.     print("take算子结果：", take_list)
    23.  
    24.     # count，统计rdd内有多少条数据，返回值为数字
    25.     num_count = rdd.count()
    26.     print("count算子结果：", num_count)

  • 相关阅读:
    2017英语一Text1
    河北吉力宝智能科技鞋：引领健康产业全面升级和互联网转型
    Unity3D 基础——通过四元数控制对象旋转
    oracle 设置表空间 表空间操作
    留言板完整版(代码）及完整版解析
    数据结构 | 顺序表
    springboot整合redis-sentinel哨兵模式集群(一)
    在移动硬盘上安装Win11系统（不使用工具）
    Dubbo3应用开发—XML形式的Dubbo应用开发和SpringBoot整合Dubbo开发
    我的个人网站不让接入微信登录，于是我做了这个
  • 原文地址：https://blog.csdn.net/weixin_63514301/article/details/139665465