浅学实战：探索PySpark实践，解锁大数据魔法！

╰半橙微兮° 2024-03-23 22:20 5阅读 0赞

#### 文章目录 ####

*  Spark和PySpark概述
 *   *  1.1 Spark简介
     *  1.2 PySpark简介
 *  二 基础准备
 *   *  2.1 PySpark库的安装
     *  2.2 构建SparkContext对象
     *  2.3 SparkContext和SparkSession
     *  2.4 构建SparkSession对象
     *  2.5 PySpark的编程模型
 *  三 数据输入
 *   *  3.1 RDD对象
     *  3.2 Python数据容器转RDD对象
     *  3.3 读取文件转RDD对象
 *  四 数据计算
 *   *  4.1 map算子
     *  4.2 flatMap算子
     *  4.3 reduceByKey算子
     *  4.4 使用SparkSession可能出现的问题
     *  4.5 filter算子
     *  4.6 distinct算子
     *  4.7 sortBy算子
     *  4.8 综合案例
 *  五 数据输出
 *   *  5.1 输出为python对象
     *   *  5.1.1 collect算子
         *  5.1.2 reduce算子
         *  5.1.3 take算子
         *  5.1.4 count算子
     *  5.2 输出到文件
     *   *  5.2.1 环境准备
         *  5.2.2 saveAsTextFile算子
         *  5.2.3 coalesce/repartition调整分区
 *  六 综合案例：搜索引擎日志分析
 *   *  6.1 需求
     *  6.2 参考代码
     *  6.3 输出结果

## Spark和PySpark概述 ##

### 1.1 Spark简介 ###

![在这里插入图片描述][96f30d15eb224b799427ce2cfda2ca40.png]

*  PySpark是Apache Spark的Python库，Apache Spark是一个开源的大数据处理框架。Apache Spark旨在进行大规模数据处理和分析，并提供统一的API，用于批处理、实时流处理、机器学习和图处理。
 *  PySpark允许开发人员使用Python编程语言与Spark交互，使更多习惯于Python而不是传统的Scala或Java的数据工程师、数据科学家和分析师能够使用Spark。
 *  简单来说，Spark是一款分布式的计算框架，用于调度成百上千的服务器集群，计算TB、PB乃至EB级别的海量数据

### 1.2 PySpark简介 ###

*  PySpark是由Spark官方开发的Python语言第三方库。  
    Python开发者可以使用pip程序快速的安装PySpark并像其它三方库那样直接使用。  
    ![在这里插入图片描述][602cf00df4fd4b3684e2609144f6e4db.png]

## 二 基础准备 ##

### 2.1 PySpark库的安装 ###

*  PySpark同样可以使用pip程序进行安装。
 *  在”CMD”命令提示符程序内，输入：

pip install pyspark

或使用国内代理镜像网站（清华大学源）

pip install -i https://pypi.tuna.tsinghua.edu.cn/simple pyspark

![在这里插入图片描述][361f66aa2d154cb2aa2f636daf03f14d.png]

### 2.2 构建SparkContext对象 ###

*  想要使用PySpark库完成数据处理，首先需要构建一个执行环境入口对象。
 *  PySpark的执行环境入口对象是：**类`SparkContext` 的类对象**
 *  使用PySpark库来创建和管理一个Spark应用程序的基本示例

# 导入必要的模块
    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对象的运行（停止PySpark程序）
    sc.stop()

1.  导入必要的模块

from pyspark import SparkConf, SparkContext

*  `pyspark`是PySpark库的主要模块，`SparkConf`和`SparkContext`是两个重要的类，用于配置和管理Spark应用程序

1.  创建SparkConf对象

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

*  `SparkConf`对象用于配置Spark应用程序的各种属性
 *  `.setMaster("local[*]")`设置了Spark的运行模式为本地模式，其中`"local[*]"`表示使用所有可用的本地处理核心
 *  `.setAppName("test_spark_app")`设置了应用程序的名称为"test\_spark\_app"

1.  创建SparkContext对象

sc = SparkContext(conf=conf)

*  基于前面创建的`SparkConf`对象，创建`SparkContext`对象，它是与Spark集群通信的主要入口点。
 *  通过此对象，您可以在集群上创建RDD（弹性分布式数据集）并执行各种操作。

1.  当前SparkContext对象所使用的Spark版本

print(sc.version)

1.  停止SparkContext对象

sc.stop()

### 2.3 SparkContext和SparkSession ###

*  `SparkContext` 和 `SparkSession` 都是 Apache Spark 中用于创建和管理 Spark 应用程序的关键组件，但它们在 Spark 版本 2.0 之后的引入中扮演了不同的角色。

**SparkContext:**

*  `SparkContext`（简称为 `sc`）是 Spark 的早期上下文对象，它是连接 Spark 集群的主要入口点。在 Spark 2.0 之前的版本中，开发人员使用 `SparkContext` 来创建 RDD、广播变量和累加器等，以及执行各种操作。它充当了应用程序与 Spark 集群之间的中间人，负责管理与集群的通信，任务调度，数据分布等。
 *  在 Spark 2.0 之后，随着 `SparkSession` 的引入，`SparkContext` 仍然可用，但更多的功能被集成到了 `SparkSession` 中，因此在新的应用程序中更常用 `SparkSession`。

**SparkSession:**

*  `SparkSession` 是在 Spark 2.0 中引入的新概念，它是创建 Spark 应用程序的主要入口点。`SparkSession` 继承自 `SparkContext`，并将许多原来与 `SparkContext` 相关的功能整合到一个对象中，包括创建 RDD、DataFrame 和 Dataset，以及执行 SQL 查询和 Spark 应用程序的配置。它还提供了一个更统一的 API，使得在 Spark 中处理不同类型的数据更加便捷。
 *  通过 `SparkSession`，可以使用 DataFrame API 来处理结构化数据，进行 SQL 查询，还可以使用 RDD API 进行更底层的操作。此外，`SparkSession` 也提供了与 Hive 集成的功能，可以在 Spark 中执行 Hive 查询。

总结：

*  `SparkContext` 是早期的 Spark 上下文，负责连接 Spark 集群和管理任务。它在 Spark 2.0 后仍然可用，但在新应用程序中更常用 `SparkSession`。
 *  `SparkSession` 是 Spark 2.0 引入的主要上下文，继承了 `SparkContext` 的功能，并在其基础上整合了更多功能，用于创建和管理 Spark 应用程序，处理不同类型的数据和执行查询。

### 2.4 构建SparkSession对象 ###

*  以下是创建SparkSession的简单示例：

from pyspark.sql import SparkSession
    # 设置本地python解释器 请注意修改为自己的python解释器目录
    import os
    os.environ['PYSPARK_PYTHON'] = "C:\environment\Python3.11.4\python.exe"
    
    # 创建一个SparkSession
    spark = SparkSession.builder \
        .appName("PySparkExample") \
        .getOrCreate()
    
    # 将数据加载到DataFrame中
    data = [("Alice", 25), ("Bob", 30), ("Charlie", 22)]
    columns = ["Name", "Age"]
    df = spark.createDataFrame(data, columns)
    
    # 执行DataFrame操作
    df.show()

### 2.5 PySpark的编程模型 ###

*  SparkContext类对象，是PySpark编程中一切功能的入口。  
    PySpark的编程，主要分为如下三大步骤：  
    ![在这里插入图片描述][d175af397cee422885c014d0aa8e71b9.png]

1.  数据输入：通过SparkContext完成数据读取
2.  数据计算：读取到的数据转换为RDD对象，调用RDD的成员方法完成计算
3.  数据输出：调用RDD的数据输出相关成员方法，将结果输出到list、元组、字典、文本文件、数据库等

![在这里插入图片描述][e9520de7369c4120ac7eae2f629033c4.gif_pic_center]

## 三 数据输入 ##

### 3.1 RDD对象 ###

*  PySpark支持多种数据的输入，在输入完成后，都会得到一个：RDD类的对象
 *  **RDD全称为：弹性分布式数据集（Resilient Distributed Datasets）**
 *  PySpark针对数据的处理，都是以RDD对象作为载体，即：
    
     *  数据存储在RDD内
     *  各类数据的计算方法，也都是RDD的成员方法
     *  RDD的数据计算方法，返回值依旧是RDD对象  
        ![在这里插入图片描述][bc180ff7b8544244a36a8528374485fd.png]

### 3.2 Python数据容器转RDD对象 ###

*  **PySpark支持通过SparkContext对象的parallelize成员方法，将：list、tuple、set、dict、str转换为PySpark的RDD对象**

from pyspark import SparkConf, SparkContext
    
    conf = SparkConf().setMaster("local[*]").setAppName("test_spark")
    sc = SparkContext(conf=conf)
    
     # 通过parallelize方法将Python对象加载到Spark内，成为RDD对象
    rdd1 = sc.parallelize([1, 2, 3, 4, 5]) #list
    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"}) # 字典
    
    # 如果要查看RDD里面有什么内容，需要用collect()方法
    print(rdd1.collect())
    print(rdd2.collect())
    print(rdd3.collect())
    print(rdd4.collect())
    print(rdd5.collect())
    sc.stop()

### 3.3 读取文件转RDD对象 ###

*  PySpark也支持通过SparkContext入口对象，来读取文件，来构建出RDD对象。

"""
    演示通过PySpark代码加载数据，即数据输入
    """
    from pyspark import SparkConf, SparkContext
    
    conf = SparkConf().setMaster("local[*]").setAppName("test_spark")
    sc = SparkContext(conf=conf)
    
    #用过textFile方法，读取文件数据加载到Spark内，成为RDD对象
    rdd = sc.textFile("C:/hello.txt")
    print(rdd.collect())
    rdd.map()
    sc.stop()

## 四 数据计算 ##

*  PySpark的数据计算，都是基于RDD对象来进行，依赖RDD对象内置丰富的：成员方法（算子）进行实现

### 4.1 map算子 ###

*  功能：map算子，是将RDD的数据一条条处理（处理的逻辑基于map算子中接收的处理函数）返回新的RDD
 *  map 是 PySpark 中的一个转换操作，用于对 RDD 中的每个元素应用一个函数，并返回一个新的 RDD，其中包含应用函数后的结果。
 *  链式调用：对于返回值是新RDD的算子，可以通过链式调用的方式多次调用算子
 *  其基本语法如下：

new_rdd = old_rdd.map(lambda x: function(x))

*  `old_rdd`: 原始的 RDD，即要进行转换的数据集。
 *  `new_rdd`: 经过 map 转换后生成的新 RDD。
 *  `lambda x: function(x)`: 这是一个匿名函数（lambda 函数），它定义了应用于每个元素 x 的转换操作。你可以根据需要定义任何适当的操作和函数。
 *  `x`: 表示 RDD 中的每个元素，function(x) 就是将该函数应用于元素 x 的结果
 *  使用 map 算子将 RDD 中的每个元素进行平方处理

from pyspark.sql import SparkSession
    # 设置本地python解释器 请注意修改为自己的python解释器目录
    import os
    os.environ['PYSPARK_PYTHON'] = "C:\environment\Python3.11.4\python.exe"
    # 创建 SparkSession
    spark = SparkSession.builder.appName("RDDMapExample").getOrCreate()
    
    # 创建一个 RDD
    data = [1, 2, 3, 4, 5]
    rdd = spark.sparkContext.parallelize(data)
    
    # 使用 map 算子对 RDD 中的每个元素进行平方
    squared_rdd = rdd.map(lambda x: x**2)
    # 支持链式调用
    # squared_rdd = rdd.map(lambda x: x**2).map(lambda x: x + 5)
    # 收集结果并打印
    result = squared_rdd.collect()
    print(result)
    
    # 停止 SparkSession
    spark.stop()

*  使用 map 转换操作对一个包含字符串的 RDD 进行转换，将每个字符串转换为大写形式

from pyspark.sql import SparkSession
    # 设置本地python解释器 请注意修改为自己的python解释器目录
    import os
    os.environ['PYSPARK_PYTHON'] = "C:\environment\Python3.11.4\python.exe"
    
    # 创建 SparkSession
    spark = SparkSession.builder.appName("MapExample").getOrCreate()
    
    # 创建一个包含字符串的 RDD
    data = ["hello", "world", "pyspark", "map"]
    rdd = spark.sparkContext.parallelize(data)
    
    # 使用 map 转换操作将每个字符串转换为大写形式
    upper_case_rdd = rdd.map(lambda x: x.upper())
    
    # 收集结果并打印
    result = upper_case_rdd.collect()
    print(result)
    
    # 停止 SparkSession
    spark.stop()

### 4.2 flatMap算子 ###

*  `flatMap` 是 PySpark 中的另一个转换操作，类似于 `map`，但是它会将每个输入元素映射为零个或多个输出元素，然后将所有输出元素组合成一个扁平的 RDD。
 *  `flatMap` 可以用于在一个 RDD 中对每个元素应用一个函数，并生成多个结果元素，最终将这些结果组合成一个新的 RDD。**简单说，对rdd执行map操作，然后进行解除嵌套操作**  
    ![在这里插入图片描述][25e806b11f424d409d7e9c56976d4ad3.png]
 *  `flatMap` 的语法如下：
    
        new_rdd = old_rdd.flatMap(lambda x: function(x))
    
     *  `old_rdd` 是要进行转换的原始 RDD，
     *  `function` 是一个应用于每个元素的函数，`x` 是 RDD 中的每个元素。`flatMap` 算子会将 `function(x)` 应用于每个元素，并返回一个或多个结果元素，然后将所有结果元素扁平化为一个新的 RDD。
 *  使用 `flatMap` 转换操作对一个包含单词列表的 RDD 进行转换，将每个单词拆分为字符，并生成一个扁平的字符列表

from pyspark.sql import SparkSession
    # 设置本地python解释器 请注意修改为自己的python解释器目录
    import os
    os.environ['PYSPARK_PYTHON'] = "C:\environment\Python3.11.4\python.exe"
    
    # 创建 SparkSession
    spark = SparkSession.builder.appName("FlatMapExample").getOrCreate()
    
    # 创建一个包含单词列表的 RDD
    data = ["hello world", "pyspark flatMap", "example"]
    rdd = spark.sparkContext.parallelize(data)
    
    # 使用 flatMap 转换操作将每个单词拆分为字符
    character_list_rdd = rdd.flatMap(lambda line: line.split())
    
    # 收集结果并打印
    result = character_list_rdd.collect()
    print(result)
    # ['hello', 'world', 'pyspark', 'flatMap', 'example']
    
    # 停止 SparkSession
    spark.stop()

### 4.3 reduceByKey算子 ###

*  `reduceByKey` 是 PySpark 中的一个转换操作，用于**对键值对形式的 RDD 进行聚合计算**。它按照键对 RDD 中的值（键值对中的键）进行分组，并应用一个指定的**聚合函数**来合并每个键对应的值。

>  *  这个操作在执行分布式计算时特别有用，因为它能够将具有相同键的数据分布在同一个分区上，从而减少数据传输和处理开销。

*  `reduceByKey` 的基本语法如下：
    
        new_rdd = old_rdd.reduceByKey(lambda x, y: function(x, y))
    
     *  `old_rdd` 是原始的键值对形式的 RDD
     *  `function` 是一个应用于每对值的聚合函数
     *  `x` 和 `y` 是每对值的两个元素。
     *  `reduceByKey` 算子将具有相同键的值进行聚合，将聚合函数应用于每对值，并返回一个新的键值对形式的 RDD。
 *  使用 `reduceByKey` 转换操作对一个包含单词计数的 RDD 进行聚合计算
    
        from pyspark.sql import SparkSession
        # 设置本地python解释器 请注意修改为自己的python解释器目录
        import os
        os.environ['PYSPARK_PYTHON'] = "C:\environment\Python3.11.4\python.exe"
        
        # 创建 SparkSession
        spark = SparkSession.builder.appName("ReduceByKeyExample").getOrCreate()
        
        # 创建一个包含单词的 RDD
        data = ["hello", "world", "hello", "pyspark", "world"]
        rdd = spark.sparkContext.parallelize(data)
        
        # 将单词映射为键值对，并使用 reduceByKey 进行单词计数
        word_counts_rdd = rdd.map(lambda word: (word, 1)).reduceByKey(lambda x, y: x + y)
        
        # 收集结果并打印
        result = word_counts_rdd.collect()
        print(result)
        
        # 停止 SparkSession
        spark.stop()
    
     *  输出结果
    
        [('hello', 2), ('world', 2), ('pyspark', 1)]
 *  注意：`SparkContext` 是 `Spark` 的基础上下文，而 `SparkSession` 是在 `Spark 2.0` 之后引入的高级概念，用于在 `Spark` 中创建和管理 `DataFrame` 和 `Dataset`

### 4.4 使用SparkSession可能出现的问题 ###

Please install psutil to have better support with spilling

*  需要安装 `psutil` 库，以便在 PySpark 中更好地监控和管理系统资源使用，包括内存、磁盘和网络资源。
 *  要安装 `psutil`，在Python 环境中使用以下命令：
    
        pip install psutil

### 4.5 filter算子 ###

*  `filter` ：用于从 RDD 中筛选出满足指定条件的元素，并返回一个包含筛选结果的新 RDD。
 *  `filter` 方法的基本语法如下：
    
        new_rdd = old_rdd.filter(lambda x: condition(x))
    
     *  `old_rdd` 是要进行筛选的原始 RDD，`condition` 是一个用于筛选元素的函数，`x` 是 RDD 中的每个元素。
     *  `filter` 方法将 `condition(x)` 应用于每个元素，并将满足条件的元素组成一个新的 RDD。
 *  使用 `filter` 方法从包含数字的 RDD 中筛选出偶数

from pyspark.sql import SparkSession
    # 设置本地python解释器 请注意修改为自己的python解释器目录
    import os
    os.environ['PYSPARK_PYTHON'] = "C:\environment\Python3.11.4\python.exe"
    # 创建 SparkSession
    spark = SparkSession.builder.appName("FilterExample").getOrCreate()
    
    # 创建一个包含数字的 RDD
    data = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
    rdd = spark.sparkContext.parallelize(data)
    
    # 使用 filter 方法筛选出偶数
    even_rdd = rdd.filter(lambda x: x % 2 == 0)
    
    # 收集结果并打印
    result = even_rdd.collect()
    print(result) # [2, 4, 6, 8, 10]
    
    # 停止 SparkSession
    spark.stop()

### 4.6 distinct算子 ###

*  `distinct` ：对RDD数据进行去重操作，并返回一个包含这些不重复元素的新 RDD。
 *  `distinct` 方法的基本语法如下：
    
        new_rdd = old_rdd.distinct()
    
     *  `old_rdd` 是要进行操作的原始 RDD，`distinct` 方法将会返回一个新的 RDD，其中包含了所有不重复的元素。
 *  使用 `distinct` 方法从包含重复元素的 RDD 中获取不重复的元素

from pyspark.sql import SparkSession
    # 设置本地python解释器 请注意修改为自己的python解释器目录
    import os
    os.environ['PYSPARK_PYTHON'] = "C:\environment\Python3.11.4\python.exe"
    
    # 创建 SparkSession
    spark = SparkSession.builder.appName("DistinctExample").getOrCreate()
    
    # 创建一个包含重复元素的 RDD
    data = [1, 2, 3, 2, 4, 3, 5, 6, 1]
    rdd = spark.sparkContext.parallelize(data)
    
    # 使用 distinct 方法获取不重复的元素
    distinct_rdd = rdd.distinct()
    
    # 收集结果并打印
    result = distinct_rdd.collect()
    print(result) # [1, 2, 3, 4, 5, 6]
    
    # 停止 SparkSession
    spark.stop()

### 4.7 sortBy算子 ###

*  `sortBy` ：用于对 RDD 中的元素根据指定的排序依据进行排序，并返回一个新的包含排序结果的 RDD。
 *  `sortBy` 方法的基本语法如下：
    
        new_rdd = old_rdd.sortBy(lambda x: key_function(x), ascending=True,numPartitions=1)
    
     *  `old_rdd` 是要进行排序的原始 RDD
     *  `key_function` 是一个用于从每个元素中提取排序关键字的函数
     *  `x` 是 RDD 中的每个元素。
     *  `ascending` 参数指定排序顺序，为 `True` 表示升序，为 `False` 表示降序。
     *  `numPartitions` （可选参数），用于指定输出 `RDD` 的分区数量，全局排序需要设置分区数为1

--------------------

*  使用 `sortBy` 方法对包含数字的 RDD 进行升序排序

from pyspark.sql import SparkSession
    
    # 创建 SparkSession
    spark = SparkSession.builder.appName("SortByExample").getOrCreate()
    
    # 创建一个包含数字的 RDD
    data = [5, 2, 8, 1, 3, 7, 4, 6]
    rdd = spark.sparkContext.parallelize(data)
    
    # 使用 sortBy 方法对 RDD 进行升序排序
    sorted_rdd = rdd.sortBy(lambda x: x, ascending=True)
    
    # 收集结果并打印
    result = sorted_rdd.collect()
    print(result) # [1, 2, 3, 4, 5, 6, 7, 8]
    
    # 停止 SparkSession
    spark.stop()

### 4.8 综合案例 ###

{
        "id":1,"timestamp":"2019-05-08T01:03.00Z","category":"平板电脑","areaName":"北京","money":"1450"}|{
        "id":2,"timestamp":"2019-05-08T01:01.00Z","category":"手机","areaName":"北京","money":"1450"}|{
        "id":3,"timestamp":"2019-05-08T01:03.00Z","category":"手机","areaName":"北京","money":"8412"}
    {
        "id":4,"timestamp":"2019-05-08T05:01.00Z","category":"电脑","areaName":"上海","money":"1513"}|{
        "id":5,"timestamp":"2019-05-08T01:03.00Z","category":"家电","areaName":"北京","money":"1550"}|{
        "id":6,"timestamp":"2019-05-08T01:01.00Z","category":"电脑","areaName":"杭州","money":"1550"}
    {
        "id":7,"timestamp":"2019-05-08T01:03.00Z","category":"电脑","areaName":"北京","money":"5611"}|{
        "id":8,"timestamp":"2019-05-08T03:01.00Z","category":"家电","areaName":"北京","money":"4410"}|{
        "id":9,"timestamp":"2019-05-08T01:03.00Z","category":"家具","areaName":"郑州","money":"1120"}
    {
        "id":10,"timestamp":"2019-05-08T01:01.00Z","category":"家具","areaName":"北京","money":"6661"}|{
        "id":11,"timestamp":"2019-05-08T05:03.00Z","category":"家具","areaName":"杭州","money":"1230"}|{
        "id":12,"timestamp":"2019-05-08T01:01.00Z","category":"书籍","areaName":"北京","money":"5550"}
    {
        "id":13,"timestamp":"2019-05-08T01:03.00Z","category":"书籍","areaName":"北京","money":"5550"}|{
        "id":14,"timestamp":"2019-05-08T01:01.00Z","category":"电脑","areaName":"北京","money":"1261"}|{
        "id":15,"timestamp":"2019-05-08T03:03.00Z","category":"电脑","areaName":"杭州","money":"6660"}
    {
        "id":16,"timestamp":"2019-05-08T01:01.00Z","category":"电脑","areaName":"天津","money":"6660"}|{
        "id":17,"timestamp":"2019-05-08T01:03.00Z","category":"书籍","areaName":"北京","money":"9000"}|{
        "id":18,"timestamp":"2019-05-08T05:01.00Z","category":"书籍","areaName":"北京","money":"1230"}
    {
        "id":19,"timestamp":"2019-05-08T01:03.00Z","category":"电脑","areaName":"杭州","money":"5551"}|{
        "id":20,"timestamp":"2019-05-08T01:01.00Z","category":"电脑","areaName":"北京","money":"2450"}
    {
        "id":21,"timestamp":"2019-05-08T01:03.00Z","category":"食品","areaName":"北京","money":"5520"}|{
        "id":22,"timestamp":"2019-05-08T01:01.00Z","category":"食品","areaName":"北京","money":"6650"}
    {
        "id":23,"timestamp":"2019-05-08T01:03.00Z","category":"服饰","areaName":"杭州","money":"1240"}|{
        "id":24,"timestamp":"2019-05-08T01:01.00Z","category":"食品","areaName":"天津","money":"5600"}
    {
        "id":25,"timestamp":"2019-05-08T01:03.00Z","category":"食品","areaName":"北京","money":"7801"}|{
        "id":26,"timestamp":"2019-05-08T01:01.00Z","category":"服饰","areaName":"北京","money":"9000"}
    {
        "id":27,"timestamp":"2019-05-08T01:03.00Z","category":"服饰","areaName":"杭州","money":"5600"}|{
        "id":28,"timestamp":"2019-05-08T01:01.00Z","category":"食品","areaName":"北京","money":"8000"}|{
        "id":29,"timestamp":"2019-05-08T02:03.00Z","category":"服饰","areaName":"杭州","money":"7000"}

*  需求，复制以上内容到文件中，使用Spark读取文件进行计算：
    
    1.  各个城市销售额排名，从大到小
    2.  全部城市，有哪些商品类别在售卖
    3.  北京市有哪些商品类别在售卖

--------------------

*  参考代码：

"""
    完成练习案例：JSON商品统计
    需求：
    1. 各个城市销售额排名，从大到小
    2. 全部城市，有哪些商品类别在售卖
    3. 北京市有哪些商品类别在售卖
    """
    from pyspark import SparkConf, SparkContext
    import os
    import json
    os.environ['PYSPARK_PYTHON'] = 'D:/dev/python/python310/python.exe'
    conf = SparkConf().setMaster("local[*]").setAppName("test_spark")
    sc = SparkContext(conf=conf)
    
    # TODO 需求1： 城市销售额排名
    # 1.1 读取文件得到RDD
    file_rdd = sc.textFile("D:/orders.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())
    # TODO 需求2： 全部城市有哪些商品类别在售卖
    # 2.1 取出全部的商品类别
    category_rdd = dict_rdd.map(lambda x: x['category']).distinct()
    print("需求2的结果：", category_rdd.collect())
    # 2.2 对全部商品类别进行去重
    # TODO 需求3： 北京市有哪些商品类别在售卖
    # 3.1 过滤北京市的数据
    beijing_data_rdd = dict_rdd.filter(lambda x: x['areaName'] == '北京')
    # 3.2 取出全部商品类别并进行商品类别去重
    result3_rdd = beijing_data_rdd.map(lambda x: x['category']).distinct()
    print("需求3的结果：", result3_rdd.collect())

## 五 数据输出 ##

### 5.1 输出为python对象 ###

#### 5.1.1 collect算子 ####

![在这里插入图片描述][df5dba67d1d14142bf83bd6c93ac5509.png]

*  功能:将RDD各个分区内的数据,统一收集到Driver中,形成一个List对象
 *  `collect` 操作会触发实际的计算操作，因此在处理大量数据时需要谨慎使用，以避免资源耗尽或内存溢出的问题。
 *  `collect` 操作的基本语法
    
        result = rdd.collect()
    
     *  `rdd` 是要执行 `collect` 操作的 RDD
     *  `result` 是一个包含所有元素的本地 Python 列表。
 *  使用 `collect` 操作将 RDD 中的元素收集到本地列表并打印
    
        from pyspark.sql import SparkSession
        
        # 创建 SparkSession
        spark = SparkSession.builder.appName("CollectExample").getOrCreate()
        
        # 创建一个包含数字的 RDD
        data = [1, 2, 3, 4, 5]
        rdd = spark.sparkContext.parallelize(data)
        
        # 使用 collect 操作将 RDD 中的元素收集到本地列表
        result = rdd.collect()
        
        # 打印结果
        print(result)
        
        # 停止 SparkSession
        spark.stop()
 *  请注意，`collect` 操作将所有数据从分布式计算节点收集到驱动程序节点，因此适用于数据量较小的情况。对于大规模数据集，使用 `collect` 可能会导致驱动程序节点的内存溢出。如果需要对数据进行分析、汇总或展示，建议使用其他适当的操作，如 `take`、`first`、`reduce` 等，以避免在内存方面的问题。

#### 5.1.2 reduce算子 ####

*  `reduce` 是一种动作（action）操作，用于对 RDD 中的元素进行累积计算。它可以将一个二元操作应用于 RDD 中的所有元素，从而将元素逐个合并以得到一个单一的结果。
 *  `reduce` 操作是一个逐步的、迭代的过程，它从左到右依次将每个元素与累积值进行二元操作，直到得到最终结果。
 *  `reduce` 操作的基本语法如下：
    
        result = rdd.reduce(lambda x, y: binary_function(x, y))
    
     *  `rdd` 是要执行 `reduce` 操作的 RDD
     *  `binary_function` 是一个二元操作的函数，`x` 和 `y` 分别是 RDD 中的两个元素
     *  `reduce` 操作将会从左到右依次将每对元素应用二元操作，得到一个最终的结果。
 *  使用 `reduce` 操作计算 RDD 中所有元素的累加和

from pyspark.sql import SparkSession
    
    # 创建 SparkSession
    spark = SparkSession.builder.appName("ReduceExample").getOrCreate()
    
    # 创建一个包含数字的 RDD
    data = [1, 2, 3, 4, 5]
    rdd = spark.sparkContext.parallelize(data)
    
    # 使用 reduce 操作计算 RDD 中所有元素的累加和
    total_sum = rdd.reduce(lambda x, y: x + y)
    
    # 打印结果
    print("Total sum:", total_sum)
    
    # 停止 SparkSession
    spark.stop()

#### 5.1.3 take算子 ####

*  `take` 是一种动作（action）操作，用于从 RDD 中获取指定数量的元素，并返回一个包含这些元素的本地 Python 列表。与 `collect` 不同，`take` 操作仅获取指定数量的元素，而不是整个 RDD 的所有元素。
 *  `take` 操作的基本语法如下：
    
        result = rdd.take(num_elements)
    
     *  `rdd` 是要执行 `take` 操作的 RDD
     *  `num_elements` 是要获取的元素数量
     *  `result` 是一个包含获取的元素的本地 Python 列表。

使用 `take` 操作从 RDD 中获取指定数量的元素

from pyspark.sql import SparkSession
    
    # 创建 SparkSession
    spark = SparkSession.builder.appName("TakeExample").getOrCreate()
    
    # 创建一个包含数字的 RDD
    data = [1, 2, 3, 4, 5]
    rdd = spark.sparkContext.parallelize(data)
    
    # 使用 take 操作从 RDD 中获取前三个元素
    selected_elements = rdd.take(3)
    
    # 打印结果
    print(selected_elements) # [1, 2, 3]
    
    # 停止 SparkSession
    spark.stop()

*  注意：`take` 操作仅在集群中获取指定数量的元素，而不会将整个 RDD 的所有元素都收集到驱动程序节点上。这使得 `take` 在需要获取一小部分数据样本时非常有用，而不会耗尽内存。

#### 5.1.4 count算子 ####

*  `count` 是一种动作（action）操作，用于计算 RDD 中元素的数量。它返回一个整数，表示 RDD 中的元素个数。
 *  `count` 操作的基本语法如下：
    
        num_elements = rdd.count()
    
     *  `rdd` 是要执行 `count` 操作的 RDD
     *  `num_elements` 是包含 RDD 中元素数量的整数。
 *  使用 `count` 操作计算 RDD 中元素的数量

from pyspark.sql import SparkSession
    
    # 创建 SparkSession
    spark = SparkSession.builder.appName("CountExample").getOrCreate()
    
    # 创建一个包含数字的 RDD
    data = [1, 2, 3, 4, 5]
    rdd = spark.sparkContext.parallelize(data)
    
    # 使用 count 操作计算 RDD 中元素的数量
    element_count = rdd.count()
    
    # 打印结果
    print("Number of elements:", element_count) # Number of elements: 5
    
    # 停止 SparkSession
    spark.stop()

*  注意，`count` 操作会触发对 RDD 的完整扫描，以确定元素的数量。对于大型数据集，可能需要一些时间来计算数量。

### 5.2 输出到文件 ###

#### 5.2.1 环境准备 ####

*  **这里使用的是Hadoop3.3.5，并使用兼容版本版本的winutils**
 *  调用保存文件的算子，需要配置Hadoop依赖

1.  [下载Hadoop安装包][Hadoop]，解压到电脑任意位置
    
     *  可进入Hadoop按照下面的步骤进行操作，也可直接点击连接进行跳转  
        ![在这里插入图片描述][68110a45efcd49f7be86907d5cb6e3eb.png]  
        ![在这里插入图片描述][3a166d029484447aade95f84c78c4f96.png]  
        ![在这里插入图片描述][467a7e0267c940009d4ec7c0e4fbcab4.png]  
        ![在这里插入图片描述][414781fa247041ccaed2ee5116d1cf8c.png]

![在这里插入图片描述][d4db9e663c40499b892e7a31f486c222.png]

1.  在Python代码中使用os模块配置：os.environ\[‘HADOOP\_HOME’\] = ‘HADOOP解压文件夹路径’  
    感谢cdarlint的贡献
2.  下载[winutils.exe][]，并放入`Hadoop`解压文件夹的`bin`目录内  
    ![在这里插入图片描述][2d1226aa399d433ebd5db4414332ea9d.png]  
    ![在这里插入图片描述][c2c9c0a2a3bc4fae95b69098d9d56404.png]
3.  下载hadoop.dll，并放入:`C:/Windows/System32` 文件夹内
    
     *  下载方法：点击文件，跳转页面，点击右侧下载按钮即可  
        ![在这里插入图片描述][1fd78c6a9fdd4a9180f1d5acf8aa9767.png]  
        ![在这里插入图片描述][b405b3769af440fd9c91a3e2e393e231.png]

#### 5.2.2 saveAsTextFile算子 ####

*  `saveAsTextFile` 是一种动作（action）操作，用于将 RDD 中的内容保存到文本文件中。
 *  `saveAsTextFile` 操作的基本语法如下：
    
        rdd.saveAsTextFile(output_path)
    
     *  `rdd` 是要保存的 RDD
     *  `output_path` 是指定的输出目录路径，用于存储 RDD 的内容。当执行 `saveAsTextFile` 操作时，PySpark 将会将 RDD 中的每个元素转换为字符串，并将这些字符串逐行写入到文本文件中。

--------------------

*  修改rdd分区为1个
    
     *  方式1，SparkConf对象设置属性全局并行度为1：
    
        conf = SparkConf().setMaster("local[*]").setAppName("test_spark")
        conf.set("spark.default.parallelism","1")
        sc = SparkContext(conf=conf)
    
     *  方式2，创建RDD的时候设置（parallelize方法传入numSlices参数为1）

rdd1 = sc.parallelize([1, 2, 3, 4, 5], numSlices=1)
    rdd2 = sc.parallelize([("Hello", 3), ("Spark", 5), ("Hi", 7)], 1)
    rdd3 = sc.parallelize([[1, 3, 5], [6, 7, 9], [11, 13, 11]], 1)

*  使用 `saveAsTextFile` 操作将 RDD 中的元素保存到文本文件
    
     *  注意，`saveAsTextFile` 操作会自动分区数据并将数据保存到多个文件中，每个分区一个文件。**一般默认分区数和cpu核心数相同**
    
        from pyspark import SparkConf, SparkContext
        import os
        import json
        os.environ['PYSPARK_PYTHON'] = 'C:/environment/Python3.11.4/python.exe'
        os.environ['HADOOP_HOME'] = "C:/environment/hadoop-3.3.5"
        conf = SparkConf().setMaster("local[*]").setAppName("test_spark")
        # conf.set("spark.default.parallelism","1")
        sc = SparkContext(conf=conf)
        
        # 准备RDD1 numSlices设置分区为1
        rdd1 = sc.parallelize([1, 2, 3, 4, 5], numSlices=1)
        
        # 准备RDD2 numSlices设置分区为1
        rdd2 = sc.parallelize([("Hello", 3), ("Spark", 5), ("Hi", 7)], 1)
        
        # 准备RDD3 numSlices设置分区为1
        rdd3 = sc.parallelize([[1, 3, 5], [6, 7, 9], [11, 13, 11]], 1)
        
        # 输出到文件中
        rdd1.saveAsTextFile("C:/output1")
        rdd2.saveAsTextFile("C:/output2")
        rdd3.saveAsTextFile("C:/output3")

![在这里插入图片描述][3d1c323e1eb84c25be604ac8638b8e6f.png]

#### 5.2.3 coalesce/repartition调整分区 ####

*  `saveAsTextFile` 操作默认会将数据分区并保存到多个文件中，每个分区一个文件，以便更好地利用分布式存储。**如果希望将数据保存为单个文本文件，可以使用 `coalesce` 或 `repartition` 操作来调整 RDD 的分区数量，将所有数据合并到一个分区中，然后再使用 `saveAsTextFile` 操作。**
 *  演示：使用 `coalesce` 或 `repartition` 的示例，以将数据保存为单个文本文件
    
     *  使用 `coalesce`：
    
        from pyspark.sql import SparkSession
        
        # 创建 SparkSession
        spark = SparkSession.builder.appName("SaveSingleTextFileExample").getOrCreate()
        
        # 创建一个包含数字的 RDD
        data = [1, 2, 3, 4, 5]
        rdd = spark.sparkContext.parallelize(data, numSlices=4)  # 指定初始分区数为 4
        
        # 将数据合并到一个分区
        single_partition_rdd = rdd.coalesce(1)
        
        # 将合并后的数据保存为单个文本文件
        output_path = "single_text_file"
        single_partition_rdd.saveAsTextFile(output_path)
        
        # 停止 SparkSession
        spark.stop()
    
     *  使用 `repartition`：
    
        from pyspark.sql import SparkSession
        
        # 创建 SparkSession
        spark = SparkSession.builder.appName("SaveSingleTextFileExample").getOrCreate()
        
        # 创建一个包含数字的 RDD
        data = [1, 2, 3, 4, 5]
        rdd = spark.sparkContext.parallelize(data, numSlices=4)  # 指定初始分区数为 4
        
        # 将数据重新分区为一个分区
        single_partition_rdd = rdd.repartition(1)
        
        # 将重新分区后的数据保存为单个文本文件
        output_path = "single_text_file"
        single_partition_rdd.saveAsTextFile(output_path)
        
        # 停止 SparkSession
        spark.stop()

## 六 综合案例：搜索引擎日志分析 ##

### 6.1 需求 ###

![在这里插入图片描述][9a5ef9925f84428a9aa8627317a9b69d.png]  
读取文件转换成RDD，并完成：

*  打印输出：热门搜索时间段（小时精度）Top3
 *  打印输出：热门搜索词Top3
 *  打印输出：统计黑马程序员关键字在哪个时段被搜索最多
 *  将数据转换为JSON格式，写出为文件

### 6.2 参考代码 ###

from pyspark import SparkConf, SparkContext
    import os
    import json
    os.environ['PYSPARK_PYTHON'] = 'C:/environment/Python3.11.4/python.exe'
    os.environ['HADOOP_HOME'] = "C:/environment/hadoop-3.3.5"
    conf = SparkConf().setMaster("local[*]").setAppName("test_spark")
    conf.set("spark.default.parallelism", "1")
    sc = SparkContext(conf=conf)
    
    # 读取文件转换成RDD
    file_rdd = sc.textFile("C:/search_log.txt")
    # TODO 需求1： 热门搜索时间段Top3（小时精度）
    # 1.1 取出全部的时间并转换为小时
    # 1.2 转换为(小时, 1) 的二元元组
    # 1.3 Key分组聚合Value
    # 1.4 排序（降序）
    # 1.5 取前3
    result1 = file_rdd.map(lambda x: x.split("\t")[0][:2])\
                        .map(lambda x:(x, 1))\
                        .reduceByKey(lambda a, b: a + b)\
                        .sortBy(lambda x: x[1], ascending=False, numPartitions=1)\
                        .take(3)
    print("需求1的结果：", result1)
    
    # TODO 需求2： 热门搜索词Top3
    # 2.1 取出全部的搜索词
    # 2.2 (词, 1) 二元元组
    # 2.3 分组聚合
    # 2.4 排序
    # 2.5 Top3
    result2 = file_rdd.map(lambda x: (x.split("\t")[2], 1)).\
        reduceByKey(lambda a, b: a + b).\
        sortBy(lambda x: x[1], ascending=False, numPartitions=1).\
        take(3)
    print("需求2的结果：", result2)
    
    # TODO 需求3： 统计黑马程序员关键字在什么时段被搜索的最多
    # 3.1 过滤内容，只保留黑马程序员关键词
    # 3.2 转换为(小时, 1) 的二元元组
    # 3.3 Key分组聚合Value
    # 3.4 排序（降序）
    # 3.5 取前1
    result3 = file_rdd.map(lambda x: x.split("\t")).\
        filter(lambda x: x[2] == '黑马程序员').\
        map(lambda x: (x[0][:2], 1)).\
        reduceByKey(lambda a, b: a + b).\
        sortBy(lambda x: x[1], ascending=False, numPartitions=1).\
        take(1)
    print("需求3的结果：", result3)
    
    # TODO 需求4： 将数据转换为JSON格式，写出到文件中
    # 4.1 转换为JSON格式的RDD
    # 4.2 写出为文件
    file_rdd.map(lambda x: x.split("\t")).\
        map(lambda x: {
        "time": x[0], "user_id": x[1], "key_word": x[2], "rank1": x[3], "rank2": x[4], "url": x[5]}).\
        saveAsTextFile("C:/output_json")

### 6.3 输出结果 ###

Setting default log level to "WARN".
    To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
    23/08/14 13:18:50 WARN Utils: Service 'SparkUI' could not bind on port 4040. Attempting port 4041.
    需求1的结果： [('20', 3479), ('23', 3087), ('21', 2989)]
    需求2的结果： [('scala', 2310), ('hadoop', 2268), ('博学谷', 2002)]
    需求3的结果： [('22', 245)]

![在这里插入图片描述][ce9c39c886094485b9f660f2fb40fc0e.png]

[96f30d15eb224b799427ce2cfda2ca40.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/6786d9982e514d31b7906116f2af8ce3.png
[602cf00df4fd4b3684e2609144f6e4db.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/4c937be4cb374a44acd87fc5a277b337.png
[361f66aa2d154cb2aa2f636daf03f14d.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/156a3b0eaa3a4c19850d66de876d8e2f.png
[d175af397cee422885c014d0aa8e71b9.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/9620f12e72c046998ab630abb2e29484.png
[e9520de7369c4120ac7eae2f629033c4.gif_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/09c43fa5061d42e384fd9ff675ee1f4b.png
[bc180ff7b8544244a36a8528374485fd.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/3098158e863e481fb62593ddd5cef9b2.png
[25e806b11f424d409d7e9c56976d4ad3.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/a0128097ef7c4eab87d08b764357c92f.png
[df5dba67d1d14142bf83bd6c93ac5509.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/ac2759b644d3436fa89be174bf1c5b3d.png
[Hadoop]: https://dlcdn.apache.org/hadoop/common/hadoop-3.3.5/
[68110a45efcd49f7be86907d5cb6e3eb.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/f7b7afd346a346f38019b277c240d7d7.png
[3a166d029484447aade95f84c78c4f96.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/92486da3691142d2b0f694df9b079e7e.png
[467a7e0267c940009d4ec7c0e4fbcab4.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/43c36348cba146b6adbd596acc14b003.png
[414781fa247041ccaed2ee5116d1cf8c.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/dbf05dbed6cc4b88b7da1022aa3479d0.png
[d4db9e663c40499b892e7a31f486c222.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/edf395e85a774d3991345dfafd4c2ce1.png
[winutils.exe]: https://github.com/cdarlint/winutils/tree/master/hadoop-3.3.5/bin
[2d1226aa399d433ebd5db4414332ea9d.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/dcd140007ad14a2780dd347d60cbc431.png
[c2c9c0a2a3bc4fae95b69098d9d56404.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/3953652a1a4c45c183c0a26bf2bdb0fe.png
[1fd78c6a9fdd4a9180f1d5acf8aa9767.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/e4e0cfbe99fe4756babe892c3fdbf9e3.png
[b405b3769af440fd9c91a3e2e393e231.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/d22e9351427e4609a9087955b572582a.png
[3d1c323e1eb84c25be604ac8638b8e6f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/9b05359e4a1f430ca93d3abca0122aa7.png
[9a5ef9925f84428a9aa8627317a9b69d.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/e1f691730c2543078423d8ccccc2d29f.png
[ce9c39c886094485b9f660f2fb40fc0e.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/d0bfd929a8fa4bb79706cc5ade05a181.png