【提示】点击每一关参考答案可以快速复制。

相关技术：Spark StreamingSpark Streaming

目录

第1关：Spark Streaming 处理实时数据流

任务描述

相关知识

编程要求

测试说明

参考答案

第2关：支付成功率分析

任务描述

相关知识

编程要求

测试说明

参考答案 

第3关：充值请求分析

任务描述

相关知识

编程要求

测试说明

参考答案

第4关：充值请求分析

任务描述

相关知识

编程要求

测试说明

参考答案

第5关：不同省份移动充值分析

任务描述

相关知识

编程要求

测试说明

参考答案

本关任务：通过 Spark 结构化流处理 JSON 数据流，对数据进行简单处理，然后将处理后的数据实时打印到控制台。

为了完成本关任务，你需要掌握：

Apache Spark Streaming 是 Apache Spark 生态系统中的一个关键模块，专注于实时数据流处理。它采用微批处理模型，将连续的数据流划分成小的微批次，以便使用 Spark 的强大批处理引擎进行数据分析。这种设计既保留了 Spark 的强大功能，又能够处理实时数据，使得 Spark Streaming 成为一种强大的实时数据处理工具。

一个核心概念是 DStream（离散流），它是 Spark Streaming 的基本数据抽象，代表了来自各种数据源的数据流。 DStream 可以通过一系列转换操作（例如 map 、reduce 、join ）来处理数据，从而实现各种实时分析和计算需求。这使得 Spark Streaming 非常灵活，适用于各种不同类型的数据处理任务。

SparkSession 是 Spark 应用程序的入口点初始化，它是与 Spark 进行交互的关键组件。

SparkSession 是 Spark 2.0+ 版本中引入的，用于替代旧版本的 SQLContext 和 HiveContext 。它是与 Spark 集群通信的入口点，负责初始化Spark应用程序所需的各种配置和资源。

SparkSession.builder 创建了一个 SparkSession 构建器，通过该构建器可以设置应用程序的配置参数。

appName("JSONStreamingExample") 设置应用程序的名称，这个名称在 Spark 集群监控工具中可用于标识应用程序。

.getOrCreate() 方法创建或获取现有的 SparkSession 实例。如果当前没有可用的 SparkSession ，它将创建一个新的实例，否则将返回已经存在的实例。

SparkSession 是 Spark 应用程序的核心入口点，它负责管理各种资源、配置信息和上下文，以便您能够进行 Spark 数据处理任务。

定义 JSON 数据的结构部分是关于如何为即将处理的 JSON 数据流定义其结构的步骤。在这里，需明确定义 JSON 数据中的各个字段，以便 Spark 能够正确解释和处理这些字段。

jsonSchema 是一个变量，用于存储数据结构的定义。这个结构被定义为一个 StructType ，表示这是一个结构化的数据类型。

StructType 是 Spark SQL 中的一种数据类型，它表示一个由多个字段组成的结构，类似于数据库表中的行。在这个示例中，定义了一个包含多个字段的数据结构。

StructField("bussinessRst", StringType(), True) 表示了结构中的一个字段定义。这里有三个参数：

这部分的代码负责定义预期的 JSON 数据结构，以确保 Spark 能够正确解析和处理数据。这对于后续的数据处理非常重要，因为它定义了字段的名称、数据类型和是否允许为空值等信息。定义正确的数据结构有助于确保数据被正确解释和分析。

使用 Spark 结构化流的 readStream 方法来读取 JSON 数据流的步骤。

streamingDF 是一个数据帧，它用于表示正在读取的实时 JSON 数据流。

spark.readStream 创建了一个 Spark 结构化流读取器，以便从数据源中读取实时数据。

.format("json") 指定了数据源的格式，这里是 JSON 。这告诉 Spark 结构化流将从 JSON 文件中读取数据。

.schema(jsonSchema) 使用之前定义的 jsonSchema 指定的数据结构。这确保了 Spark 正确地解释和处理数据。

.option("maxFilesPerTrigger", 1) 设置了每个触发操作处理的文件数。这里将每个触发操作限制为处理一个文件，以实现实时数据处理的微批处理模式。

.load(inputPath) 指定了要读取数据的文件夹路径，即之前定义的 inputPath 变量。这告诉 Spark 结构化流在该路径中寻找新的 JSON 数据文件。

读取 JSON 数据流的这个部分是实时数据处理流程的关键起点。它告诉 Spark 如何从指定的数据源中读取数据，并确定数据的结构。一旦设置完成，Spark 结构化流将会实时监视该路径，捕获新的数据，并在每个触发操作中处理这些数据。

该部分是关于如何定义对实时数据流进行处理的逻辑的步骤。在这里，可以根据需要执行各种数据处理和转换操作。

streamingDF.select("*") 是一个简单的选择操作，它选择了数据流中的所有字段。这是一种最简单的处理逻辑，可以用来将原始数据直接传递到后续的输出操作。

然而，这只是一个示例的起点。根据应用程序的需求，可以在这里定义更复杂的处理逻辑，例如过滤、映射、聚合、连接其他数据源等。 Spark 结构化流提供了丰富的数据处理操作，可以帮助实现各种实时数据处理需求。

这一部分是实时数据流处理的核心，可以根据具体的应用场景来编写适当的数据处理逻辑。

创建了一个输出操作，将数据输出到控制台。

query 是一个流查询对象，它代表了正在执行的实时查询任务。

processedDF.writeStream 创建了一个输出操作，以便将实时处理的数据写出。

.outputMode("append") 指定了输出模式。在这里，输出模式设置为 "append" ，表示新数据将追加到输出结果中。其他可选的输出模式包括 "complete" 和 "update" ，具体选择取决于应用程序的需求。

.format("console") 指定了输出的格式。在这里，输出格式设置为 "console" ，这将在控制台上打印实时处理的数据。还可以选择其他输出格式，如 "parquet"、"csv"、"json" 等，根据需要将数据写入不同的目标。

.start() 启动流查询，使其开始运行。一旦启动，Spark 结构化流将开始监视数据流，处理数据，并将结果输出到指定的输出目标。

负责启动实时数据流查询，以便实时处理数据流并等待其终止。

一旦流查询启动，它将开始监视数据流、执行实时数据处理逻辑，并将处理后的数据写入指定的输出目标。awaitTermination 方法的调用将使应用程序等待流查询终止，通常在应用程序执行完所有处理后或手动终止查询时才会退出。

awaitTermination 方法中传递了一个超时参数。这表示等待查询终止的最大时间（以秒为单位）。

query.stop() 是用于停止正在运行的 Spark 结构化流查询的方法。这个方法会立即终止查询，无论查询的运行状态如何。一旦调用 query.stop() ，查询将停止，不再处理新的数据，并且查询任务将被终止。

这种方法在需要手动停止查询的情况下非常有用，例如在应用程序完成任务后或发生错误时。它可以用来主动终止查询，以确保查询不会一直运行下去。一旦查询被停止，awaitTermination 方法会返回，应用程序可以继续执行其他代码或退出。

在调用 query.stop() 之后，查询将无法重新启动，如果需要重新运行查询，需要创建一个新的查询实例并启动它。

根据提示，在右侧编辑器 Begin-End 内补充代码，编程要求如下：

无

旨在对中国移动支付数据进行分析，重点关注不同支付渠道的支付金额，以及成功和失败交易的比例，同时深入研究失败交易的原因。

为了完成本关任务，你需要掌握：

中国移动支付领域涉及大量的支付交易数据。本实验旨在分析中国移动支付数据，重点关注不同渠道的支付金额以及成功和失败交易的比例，同时深入研究失败交易的原因。

分析不同支付渠道的支付金额是一项关键任务，可以帮助我们深入了解在中国移动支付领域，哪些渠道在吸引最多交易和资金流入方面表现出色。这种分析可以揭示消费者偏好、支付渠道的效益，以及在支付生态系统中分配资金的方式。

首先，导入所需的模块，包括 SparkSession 和 col 函数，以便在 Spark 中进行数据分析。 您创建了一个名为 "PaymentAnalysis" 的 Spark 应用程序，然后初始化了一个 SparkSession ，其名称为 "PaymentAnalysis" 。

使用 spark.read.json() 方法从指定路径的 JSON 文件（"cmcc.json" ）中读取数据，并将其创建为一个 DataFrame ，命名为 df 。读取 JSON 数据并创建 DataFrame 。

使用 DataFrame 的 groupBy 和 agg 方法，根据 "channelCode" 列对数据进行分组，同时对 "chargefee" 列进行求和，以及对 "channelCode" 列进行计数。

使用 withColumnRenamed 方法为生成的列更改名称，将 "sum(chargefee)" 列重命名为 "total_payment_in_cents"，将 "count(channelCode)" 列重命名为 "recharge_count" 。

然后，通过除以100将支付金额单位转换为元，并创建一个新的列 "total_payment_in_yuan"。

最后，使用 show 方法显示了转换后的数据，以查看每个渠道的充值数量和总支付金额（以元为单位）。

程序运行结果如下：

程序结果提供了关于不同渠道的支付情况的分析。以下是对结果的分析：

使用 DataFrame 的 filter 操作，分别统计成功和失败的支付数量。col("bussinessRst") == "0000" 表示成功的支付记录，而 col("bussinessRst") != "0000" 表示失败的支付记录。

使用 DataFrame 的 count 方法计算总的支付数据记录的数量。

通过将成功数和失败数除以总数，计算成功和失败的支付比例。结果以百分比形式表示。

使用 print 语句将成功和失败的支付比例打印出来，以百分比形式表示。同时，使用 spark.stop() 停止 SparkSession ，释放资源并终止 Spark 应用程序。

程序运行结果如下：

根据提示，在右侧编辑器 Begin-End 内补充代码，编程要求如下：

根据 DataFrame df 中的 "bussinessRst" 列过滤出值不等于 "0000" 的行，并将结果保存在名为 failed_payments 的新 DataFrame 中。

从 failed_payments DataFrame 中选择（.select()）以下字段 "bussinessRst", "channelCode", "chargefee", "bussinessRst", 和 "retMsg" 并将结果保存在名为 selected_data 的新 DataFrame 中。

显示 selected_data DataFrame 中的数据，确保不截断列内容（ truncate=False ），以便查看完整的列数据。

无

本关任务：分析充值请求的起始时间（ startReqTime ）和结束时间（ endReqTime ）以了解请求的持续时间。同时，检查不同服务名称（ serviceName ）的充值请求的数量。

为了完成本关任务，你需要掌握：充值请求的持续时间。

分析充值请求的起始时间（ startReqTime ）和结束时间（ endReqTime ）以了解请求的持续时间在多个方面具有重要作用。首先，通过计算每个充值请求的持续时间，我们可以对充值系统的性能进行评估。短持续时间通常表明系统响应速度快，而较长的持续时间可能暗示性能问题或延迟。这有助于识别和解决潜在的性能瓶颈，从而提高系统的效率和响应速度。

其次，分析持续时间还有助于异常检测。异常的持续时间可能是请求超时、系统故障或其他错误的迹象。通过及时识别这些异常，可以采取措施来维护系统的可靠性，确保充值请求的顺畅进行。

此外，了解请求的持续时间对用户体验至关重要。用户通常更喜欢快速的交易体验，而较长的持续时间可能会降低用户满意度。因此，通过优化请求持续时间，可以提高用户的满意度和忠诚度。

导入必要的库，并创建了一个名为 "RechargeRequestAnalysis" 的 SparkSession 对象，以便初始化 Spark 应用程序。

使用 Spark 从 JSON 文件中读取数据并将其加载到名为 "df" 的 DataFrame 中。

此配置用于将 Spark 的日期时间解析行为设置为 "LEGACY"，以确保与 Spark 3.0 之前的版本相兼容。

将时间戳列 "startReqTime" 和 "endReqTime" 转换为 Timestamp 类型，以便进行后续时间计算。

计算了充值请求的持续时间（秒）并将其存储在名为 "requestDuration" 的新列中。

从 DataFrame 中选择列 "serviceName" 、"startReqTime" 、"endReqTime" 和 "requestDuration" 并在控制台上显示这些信息，以便分析充值请求的持续时间。

使用 DataFrame 的 agg 函数找出最长和最短的充值请求持续时间，并将结果存储在变量 "max_duration" 和 "min_duration" 中。

打印出最长和最短操作时间，以便您了解充值请求的时间分布情况。停止 SparkSession 并释放相关资源，终止 Spark 应用程序的执行。

程序运行结果如下：

运行结果显示了不同服务名称（serviceName）的充值请求信息，包括请求的起始时间（startReqTime）、结束时间（endReqTime）以及请求的持续时间（requestDuration）。

从这些结果中，我们可以初步了解不同服务名称的充值请求的持续时间情况，以及可能存在的数据质量问题，例如缺失值或未记录的信息。要更深入地分析这些数据，可能需要进一步处理和清洗数据，以确保准确性和完整性。

充值请求的最长操作时间为 31 秒，而最短操作时间为 0 秒。这意味着在数据集中存在一些充值请求需要较长的时间来完成，而另一些几乎是瞬时完成的。

这种分析可以帮助我们识别充值请求的性能特征，可能有助于系统优化。特别是对于长时间的充值请求，可以进一步调查其原因，以确保系统的性能和可用性。另外，对于瞬时完成的请求，也可以进一步研究其性质，以优化资源利用和响应时间。这些信息对于系统运营和性能改进非常重要。

根据提示，在右侧编辑器 Begin-End 内补充代码，统计不同服务名称（ serviceName ）的充值请求的数量。编程要求如下：

使用 DataFrame 的 groupBy 方法按照服务名称字段( "serviceName" )进行分组。然后，使用 count 方法计算每个服务的请求数量并将结果存储在 service_request_count 变量中。

使用 show 方法来展示不同服务的充值请求数量。

无

 本关任务：旨在分析客户IP地址在数据中的出现频率。统计每个唯一 IP 地址出现的次数，以便了解哪些 IP 地址最常出现。

为了完成本关任务，你需要掌握：客户端IP活跃性。

用户行为分析是一项关键的数据分析任务，它有助于企业深入了解其客户或用户，从而制定更智慧的战略决策。首先，根据客户端 IP 地址，可以分析不同IP地址的活跃性和地理位置分布。客户端 IP 地址可用于识别用户，了解哪些 IP 地址的访问频率高，哪些可能属于爬虫或恶意用户。

客户端 IP 地址是访问您的网站或服务的用户的唯一标识之一。通过分析不同 IP 地址的活跃性，可以了解哪些 IP 地址访问频率较高，哪些可能是爬虫或恶意用户。通过使用访问频率、平均会话时长等指标来评估活跃性。

从 PySpark 中导入所需的模块和函数，并创建一个 SparkSession ，这是使用 Spark 功能的入口点。以及从 "cmcc.json" 文件中读取 JSON 数据并加载到名为 json_df 的 DataFrame中。

设置 timeParserPolicy 的配置选项为 "LEGACY" ，它影响日期和时间解析的行为。同时将 DataFrame 中的时间戳列（ "startReqTime" 和 "endReqTime" ）转换为以秒为单位的时间戳。

该行代码的作用是向 DataFrame json_df 中添加一个名为 "sessionDuration" 的新列，该列用于存储会话持续时间（以秒为单位）。这个会话持续时间是通过从 "endTimestamp" 列中的时间戳减去 "startTimestamp" 列中的时间戳来计算得出的。

json_df : 这是包含 JSON 数据的 Spark DataFrame。

.withColumn("sessionDuration", ...) : 这是一个 DataFrame 的方法，用于向 DataFrame 添加一个新列，其中 "sessionDuration" 是新列的名称。

(col("endTimestamp") - col("startTimestamp")) : 这是一个表达式，它计算了 "endTimestamp" 列中的时间戳减去 "startTimestamp" 列中的时间戳。这个表达式的结果是会话的持续时间（以秒为单位）。

通过统计每个不同 IP 地址在数据集中出现的次数，然后按照出现次数从高到低排序，以了解哪些IP 地址出现频率最高。

json_df : 这是包含了 JSON 数据的 Spark DataFrame ，它是分析的数据源。

.groupBy("clientIp") : 这是一个 DataFrame 的方法，它将数据按照 "clientIp" 列的值进行分组。这意味着数据将根据不同的 IP 地址进行分组。

.agg(count("*").alias("count")) : 这是聚合操作，它计算每个 IP 地址出现的次数。count("*") 计算每个分组中的行数，也就是每个 IP 地址的出现次数。.alias("count") 将结果的列名更改为 "count"，以便更好地理解结果。

.orderBy("count", ascending=False) : 这部分对结果进行排序，按照 "count" 列的值，以降序（从最高到最低）的方式排序。这将使最频繁出现的 IP 地址在顶部。

程序运行结果如下：

代码的运行结果显示了不同 IP 地址在数据集中的出现频率，并按出现次数降序排列。其中

clientIp 列： 显示了不同的 IP 地址。每一行代表一个唯一的 IP 地址。

count 列： 显示了每个 IP 地址在数据集中出现的次数。出现次数从高到低排列。

根据这个结果，可以得出以下一些观察和分析：

"175.10.113.181"是出现次数最多的 IP 地址，共出现487次。 "223.240.218.49"是第二常见的 IP 地址，出现447次。

average_duration = json_df.groupBy("clientIp").agg(expr("avg(sessionDuration * 1000)").alias("avgSessionDurationMillis"))

average_duration = average_duration.orderBy("avgSessionDurationMillis", ascending=True)

average_duration : 这是包含了每个不同 IP 地址的平均会话时长的 Spark DataFrame。

orderBy("avgSessionDurationMillis", ascending=True) : 这部分对结果进行排序，按照 "avgSessionDurationMillis" 列的值，以升序（从最小到最大）的方式排序。这将使平均会话时长最短的 IP 地址出现在结果的顶部。

程序运行结果如下：

根据统计结果显示为 NULL ，这通常表示在计算平均会话时长时，没有匹配的数据可供计算。这可能是因为没有满足条件的数据行，导致无法计算平均会话时长。

根据数据缺失值统计结果，发现 "startReqTime" 列没有缺失值，但 "endReqTime" 列存在着 27,277 条缺失值数据。这种数据缺失导致在计算平均会话时长时出现错误。因此，在进行统计分析之前，需要对数据进行预处理，以处理缺失值，确保数据的完整性和准确性。

数据缺失值统计结果如下：

根据提示，在右侧编辑器 Begin-End 内补充代码，编程要求如下：

使用 groupBy 函数对 DataFrame json_df 按照 "clientIp" 列的唯一值进行分组。存储结果在名为 grouped_ip 的新 DataFrame 中。

使用 agg 函数对分组后的数据执行聚合操作。count("*") 计算每个分组中的数据行数，即每个 IP 地址的访问频次。使用 alias("count") 为聚合结果创建一个新列，列名为 "count"，用于存储访问频次统计结果。存储结果在名为 ip_count 的新 DataFrame 中。

使用 orderBy 函数对 ip_count 中的数据按照 "count" 列的值进行排序。使用 ascending=False 按降序排序，以将访问频次最高的 IP 地址排在前面。存储结果在名为 sorted_ip_frequency 的新 DataFrame 中。

无

本关任务：对移动充值相关数据进行统计，特别是针对不同省份。

为了完成本关任务，你需要掌握：不同省份移动充值。

不同省份移动充值分析在电信行业中具有重要作用。首先，它为电信运营商提供了有力的业务决策支持。通过深入了解不同省份的移动充值模式和趋势，运营商可以更明智地分配资源、精确定位市场需求以及制定更具竞争力的资费计划。这有助于提高业务的效益和竞争力。

其次，分析不同省份的充值数据可以用于优化营销策略。了解不同省份的客户充值行为和偏好使得运营商能够更好地制定营销策略，包括制定有针对性的优惠活动、定位广告以及提供个性化服务。

首先初始化一个 SparkSession 。 SparkSession 是 Spark 功能的入口点，用于创建数据框、读取数据以及对数据执行各种操作。

使用 spark.read.json 方法从指定路径读取 JSON 文件，并将数据加载到名为 df 的数据框中。

从 DataFrame 中筛选出 "chargefee" 列中的那些可以被成功转换为整数的行，同时将不符合整数格式的值或包含 null 值的行过滤掉。

col("chargefee") ：使用 col 函数从 DataFrame 中选择名为 "chargefee" 的列，即数据框中包含充值费用的列。

.cast(IntegerType()) ：使用 cast 方法将选定列的数据类型转换为整数。这是为了确保 "chargefee" 列中的值都可以转换为整数。

.isNotNull() ：isNotNull 函数用于检查是否某列中的值不为 null ，如果为 null ，则过滤掉该行数据。

对 DataFrame df 进行分组和汇总操作，以计算每个省份的充值费用总和。

df.groupBy("provinceCode") ：使用 groupBy 方法，将 DataFrame df 按照 "provinceCode" 列的值进行分组。创建了一个以不同省份代码作为分组键的分组对象，用于后续的聚合操作。

.agg(sum("chargefee").alias("total_chargefee")) ：在上面的分组之后，使用 agg 方法对每个分组执行聚合操作。这里的聚合操作有两部分：

sum("chargefee") ：这部分计算了每个分组内 "chargefee" 列的总和。使用 PySpark 内置的 sum 函数完成的，它会对每个分组内的 "chargefee" 值进行求和。

.alias("total_chargefee") ：alias 方法用于为聚合结果列指定别名，即将聚合结果列的名称更改为 "total_chargefee" 。这是为了更好地描述这一列的内容，即每个省份的总充值费用。

定义一个字典（ province_mapping ），将省份代码映射到省份名称。此映射用于向 result 数据框添加一个新列，即 "provinceName" 。

通过循环遍历字典项，对于每个省份代码，根据相应的代码更新 result数据框中的 "provinceName" 列。

将省份代码映射到省份名称的主要原因是提高数据处理的可读性、可维护性和准确性。在大规模数据分析中，使用代码代表省份有多个优点。首先，这减少了潜在的数据不一致性，因为省份名称可能包含各种特殊字符、大小写不一致或拼写错误。而省份代码通常是标准化的、唯一的标识符，更容易阅读和匹配。

其次，省份代码映射减少了数据输入错误的风险。手动输入省份名称时，拼写错误或大小写不一致是常见问题，但代码映射可以降低这些问题的可能性，因为代码是事先定义的，通常由系统自动生成。

程序运行结果如下：

provinceCode 列：这一列显示了各个省份的代码。每个省份都有一个唯一的代码，用于标识不同的地区。

total_chargefee 列：这一列表示每个省份的总充值费用。总充值费用是在数据分析过程中计算的，它表示每个省份的移动网络充值金额。

金额单位为分

provinceName 列：这一列是根据之前定义的 province_mapping 字典中的映射将省份代码转换为省份名称的结果。这使得结果更容易理解和阅读。

根据提示，在右侧编辑器 Begin-End 内补充代码，编程要求如下：

使用 withColumn 方法，选择 "chargefee" 列，并将其新的数据类型设置为整数类型( IntegerType() )，然后将结果存储回 "chargefee" 列。

使用 filter 方法，创建一个筛选条件，使用 cast 将 "chargefee" 列的数据类型转换为整数类型，然后使用 isNotNull 检查是否为不为 null（即有效整数值），最后将结果存储回 "df" 。

使用 groupBy 方法，按照 "provinceCode" 列进行分组。使用 agg 方法，计算每个分组内的 "chargefee" 列的总和，并将结果以别名 "total_chargefee" 存储在 "result" 数据框中。

无

至此，所有内容都完成辣。如果存在任何问题欢迎大佬指教🥰！