Spark SQL 数据分析实战的 6 个高频场景（附数据清洗案例）

作为 Apache Spark 的核心组件，Spark SQL 提供了强大的结构化数据处理能力，特别适合大数据分析场景。它通过 SQL 语法或 DataFrame API 实现高效查询，支持分布式计算，能处理 TB 级数据。下面我将逐步介绍 Spark SQL 在实战中的 6 个高频场景（基于常见行业应用，如电商、金融和日志分析），并附上一个完整的数据清洗案例。所有内容基于真实数据工程实践，确保可靠性。

高频场景 1: 数据聚合与汇总

这是最基础且高频的场景，用于计算指标如总销售额、平均值等。Spark SQL 使用 GROUP BY 和聚合函数（如 SUM()、AVG()）实现。例如，在电商中分析每日销售额：

• 应用示例：计算每个产品的总销量，公式可表示为 $\text{总销量} = \sum_{i=1}^{n} \text{sales}_i$。
• 代码片段（使用 PySpark）:

from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("Aggregation").getOrCreate()
df = spark.read.csv("sales_data.csv", header=True, inferSchema=True)
result = df.groupBy("product_id").agg({"sales": "sum"}).withColumnRenamed("sum(sales)", "total_sales")
result.show()

高频场景 2: 数据过滤与筛选

用于快速提取子集，例如筛选特定时间段或条件的数据。Spark SQL 的 WHERE 子句高效处理海量数据。在金融风控中，常用于识别高风险交易：

• 应用示例：筛选交易金额大于阈值的记录，公式为 $\text{amount} > 1000$。
• 代码片段:

high_risk_transactions = df.filter("amount > 1000 AND status = 'pending'")
high_risk_transactions.show()

高频场景 3: 数据连接（JOIN 操作）

合并多个数据源，如用户表和订单表。Spark SQL 支持各种 JOIN 类型（如 INNER JOIN、LEFT JOIN），在数据集成中必不可少。例如，电商中关联用户信息和购买记录：

• 应用示例：JOIN 操作可表示为 $\text{Users} \bowtie_{\text{user_id}} \text{Orders}$。
• 代码片段:

users_df = spark.read.csv("users.csv", header=True)
orders_df = spark.read.csv("orders.csv", header=True)
joined_df = users_df.join(orders_df, users_df.user_id == orders_df.user_id, "inner")
joined_df.show()

高频场景 4: 窗口函数分析

用于复杂分析如排名、累计计算或时间序列处理。Spark SQL 的窗口函数（如 ROW_NUMBER()、LAG()）在用户行为分析中高频出现。例如，计算用户购买排名：

• 应用示例：窗口函数可定义为 $\text{RANK() OVER (PARTITION BY user_id ORDER BY purchase_date DESC)}$。
• 代码片段:

from pyspark.sql.window import Window
from pyspark.sql.functions import rank
window_spec = Window.partitionBy("user_id").orderBy(df["purchase_date"].desc())
df_with_rank = df.withColumn("rank", rank().over(window_spec))
df_with_rank.show()

高频场景 5: 用户定义函数（UDFs）

处理自定义逻辑，如复杂字符串处理或数学计算。Spark SQL 支持 UDFs 扩展功能，在数据清洗和特征工程中常见。例如，自定义函数计算字符串长度：

• 应用示例：UDF 可定义为 $f(x) = \text{len}(x)$。
• 代码片段:

from pyspark.sql.functions import udf
from pyspark.sql.types import IntegerType
length_udf = udf(lambda x: len(x), IntegerType())
df_with_length = df.withColumn("name_length", length_udf(df["name"]))
df_with_length.show()

高频场景 6: 数据探索与可视化准备

为下游可视化（如 Tableau 或 Matplotlib）准备数据，包括采样、统计描述等。Spark SQL 的 describe() 或 sample() 函数高效支持探索性分析。例如，生成数据摘要用于报表：

• 应用示例：计算数值列的统计量，如均值 $\mu = \frac{\sum x_i}{n}$。
• 代码片段:

summary = df.describe(["age", "income"])  # 输出 count, mean, stddev 等
summary.show()
sampled_data = df.sample(fraction=0.1)  # 采样用于可视化

附：数据清洗案例（电商数据清洗）

数据清洗是数据分析的基础步骤，Spark SQL 擅长处理缺失值、去重、类型转换等。以下是一个完整案例，基于模拟电商数据集（包含用户交易记录），目标清洗后输出干净数据。

案例背景:

• 数据集：ecommerce_data.csv，包含列：user_id, transaction_id, amount, product, date, status。
• 常见问题：缺失值（如 amount 为 null）、重复记录、错误数据类型（如 date 格式不一致）、无效状态（如 status 非标准值）。
• 清洗目标：移除无效行、填充缺失值、统一格式，确保数据质量。

清洗步骤和代码:

1. 加载数据并初步检查:

   from pyspark.sql import SparkSession
   spark = SparkSession.builder.appName("DataCleaning").getOrCreate()
   df = spark.read.csv("ecommerce_data.csv", header=True, inferSchema=True)
   print("原始数据行数:", df.count())  # 假设原始有 10000 行
   df.printSchema()  # 检查数据类型
   

2. 处理缺失值:

   • 填充 amount 缺失值为 0（基于业务逻辑，假设缺失表示无交易）。
   • 删除 user_id 或 transaction_id 缺失的行（关键列不能为空）。

   from pyspark.sql.functions import when, col
   df_clean = df.na.fill({"amount": 0})  # 填充 amount 缺失
   df_clean = df_clean.dropna(subset=["user_id", "transaction_id"])  # 删除关键列缺失行
   

3. 去重处理:

   • 基于 transaction_id 移除重复记录。

   df_clean = df_clean.dropDuplicates(["transaction_id"])
   

4. 数据类型转换和格式化:

   • 转换 date 列为标准日期类型。
   • 过滤无效 status 值（只保留 'completed' 或 'cancelled'）。

   from pyspark.sql.functions import to_date
   df_clean = df_clean.withColumn("date", to_date(col("date"), "yyyy-MM-dd"))  # 统一日期格式
   df_clean = df_clean.filter(col("status").isin(["completed", "cancelled"]))  # 只保留有效状态
   

5. 最终输出和验证:

   print("清洗后数据行数:", df_clean.count())  # 假设减少到 9500 行
   df_clean.show(5)  # 展示清洗后样本
   df_clean.write.csv("cleaned_ecommerce_data", mode="overwrite")  # 保存清洗数据
   

案例总结:

• 清洗后数据质量提升：移除无效记录约 5%，缺失值处理完整，格式统一。
• Spark SQL 优势：分布式处理高效（在集群上秒级完成 TB 数据清洗），代码简洁。
• 实际应用：此案例可直接用于电商分析流水线，支持后续场景如聚合或 JOIN。

通过以上 6 个高频场景和清洗案例，您能快速上手 Spark SQL 实战。如果有特定数据集或需求，我可以进一步优化示例！