PySpark 实战

本章目标：能用 PySpark 读订单 CSV、定义 schema、清洗订单、算出月度 GMV 和品类 TopN，并用 DuckDB 对账。 前置知识：读完 Spark 基础；会写 GROUP BY、JOIN、窗口函数。 预计用时：3 小时。

1. 为什么需要它

生产里大多数 Spark 任务是 SQL，但复杂清洗、调用 Python 库、把多个数据源拼成流水线时，DataFrame API 更顺手。它和 SQL 最终走同一个 Catalyst 优化器，不存在“谁更高级”。

本章用 data/sample/ 下的两份小 CSV 走一遍真实流程：读数据、定 schema、清洗、算指标、对账、写分区表。数据虽然只有 12 行，代码结构和生产任务一致。验收数字如下：

注意：完全数据集和真实脏数据的完整处理在独立的项目线里做。本章只保留能讲清 PySpark 用法的字段，避免把工具课变成项目复述。

2. 核心概念

2.1 SparkSession：一个程序一个入口

SparkSession 是 Spark 3.x 的统一入口，封装了 SparkContext、SQL 和 catalog：

from pyspark.sql import SparkSession

spark = (SparkSession.builder
         .appName("pyspark_gmv")
         .master("local[*]")
         .config("spark.sql.shuffle.partitions", "8")
         .getOrCreate())

appName 在 Spark UI 和 YARN 上可见；local[*] 用本机所有核，学习用；getOrCreate() 已有会话就复用；结束前 spark.stop()。本地小数据把 shuffle 分区设成 8，避免默认 200 产生大量空任务；这不是生产参数。

2.2 读取 CSV 与 schema

最省事的读法是 option("header", True).option("inferSchema", True).csv(path)。inferSchema 会多扫一遍数据推断类型，方便探索，但生产里更推荐显式 schema。下面用 DDL 字符串定义，写起来比 StructField 列表短：

from pyspark.sql.types import StructType

orders_schema = StructType.fromDDL(
    "order_id STRING, status STRING, city STRING, "
    "amount DOUBLE, order_date STRING")
items_schema = StructType.fromDDL(
    "order_id STRING, product_id STRING, category STRING, "
    "price DOUBLE, freight DOUBLE")

orders = (spark.read.option("header", True)
          .schema(orders_schema)
          .csv("data/sample/orders_demo.csv"))
items = (spark.read.option("header", True)
         .schema(items_schema)
         .csv("data/sample/order_items_demo.csv"))

示例数据的 order_date 在 CSV 里是字符串。ODS 先保持字符串，DWD 再用 to_date / to_timestamp 转换，这样原始值可追溯。显式 schema 类型稳定，不会因为某批数据有异常字符串就推断成 string，还能少扫一遍、明确字段含义。常用选项有 header、sep、encoding、nullValue、mode（PERMISSIVE / DROPMALFORMED / FAILFAST）和 schema。

注意：订单表的 amount 是订单级金额，明细表的 price + freight 才是商品级金额。两者只能选一个口径算 GMV，不能相加，否则金额翻倍。本章统一用明细口径。

2.3 DataFrame 常用 API

from pyspark.sql import functions as F

df.select("order_id", F.col("price").alias("单价"))
df.filter((F.col("status") != "canceled") & F.col("price").isNotNull())
df.withColumn("amount", F.col("price") + F.col("freight"))
df.groupBy("category").agg(F.count("*").alias("cnt"),
                           F.round(F.sum("amount"), 2).alias("gmv"))
a.join(b, on="order_id", how="left")
df.orderBy(F.col("gmv").desc()).limit(10)

注意：join 前先想清楚两边的粒度。订单表一行一单，明细表一行一商品；直接 join 后行数会按明细膨胀。评价和支付也是一单多行，必须先聚合到订单粒度再 join，否则 GMV 和评分都会被放大。

2.4 Spark SQL 临时视图

detail.createOrReplaceTempView("dwd_detail")
spark.sql("""
    SELECT date_format(order_date, 'yyyy-MM') AS month,
           ROUND(SUM(item_amount_with_freight), 2) AS gmv
    FROM dwd_detail
    GROUP BY 1 ORDER BY 1
""").show(20, truncate=False)

临时视图只在当前 SparkSession 有效，程序结束就消失，不写元数据、不落表。createGlobalTempView 可跨会话，访问要加 global_temp.；需要持久化则用 saveAsTable 写入 Metastore。

注意：Spark SQL 和 Hive SQL 不完全一样。Spark 日期格式化用 date_format(col, 'yyyy-MM')，DuckDB 用 strftime(col, '%Y-%m')；Spark 里常用 coalesce 替代 nvl。跨引擎写 SQL 时，日期函数和隐式类型转换是两大坑。

2.5 清洗与口径

示例数据能覆盖的清洗点不多，但每一条都对应生产里的一类问题：

清洗顺序建议：订单先排除 canceled 和 unavailable；明细 join 有效订单；评价先聚合再 join；商品缺品类补 unknown；时间字符串统一转 DATE。不要为了“看起来干净”而删掉业务空值，比如未签收订单的签收时间。

2.6 分区、repartition 与 coalesce

分区数决定并行度：太少机器闲着、单任务过大易 OOM；太多调度开销大、输出小文件多。

写 Parquet 分区表：

spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")
out = detail.withColumn("dt", F.date_format("order_date", "yyyy-MM"))
(out.repartition(8, "dt").write.mode("overwrite")
     .partitionBy("dt").parquet("data/warehouse/dwd_order_detail"))

partitionBy("dt") 生成 dt=2024-01/ 这样的目录，查询带 dt 条件时只读命中目录，这叫分区裁剪；读回后看 explain 里的 PartitionFilters 可验证。dynamic 覆盖只重写本次涉及的分区；默认静态模式会覆盖整张表，补数时特别危险。

注意：Windows 本地写 Parquet 可能报 winutils.exe 相关错误，这是缺 Hadoop 本地库，不是代码错误。只读练习可继续；要写文件用 WSL2、Docker，或参考 project/hive/README.md。不要为一个本地写文件折腾一下午。

3. 动手示例

3.1 完整脚本

先确认 java -version 和 python -c "import pyspark; print(pyspark.__version__)" 都正常，再在项目根目录新建 scripts/spark_gmv.py，写入下面的完整脚本。

# -*- coding: utf-8 -*-
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import StructType

spark = (SparkSession.builder
         .appName("sales_gmv")
         .master("local[*]")
         .config("spark.sql.shuffle.partitions", "8")
         .getOrCreate())
spark.sparkContext.setLogLevel("ERROR")

RAW = "data/sample/"
orders_schema = StructType.fromDDL(
    "order_id STRING, status STRING, city STRING, "
    "amount DOUBLE, order_date STRING")
items_schema = StructType.fromDDL(
    "order_id STRING, product_id STRING, category STRING, "
    "price DOUBLE, freight DOUBLE")

def read_csv(path, schema):
    return spark.read.option("header", True).schema(schema).csv(RAW + path)

orders = read_csv("orders_demo.csv", orders_schema)
items = read_csv("order_items_demo.csv", items_schema)
orders.printSchema()

valid_orders = (orders
                .filter(~F.col("status").isin("canceled", "unavailable"))
                .withColumn("order_date", F.to_date("order_date")))
detail = (items
          .join(valid_orders.select("order_id", "order_date",
                                    "city", "status"),
                "order_id", "inner")
          .withColumn("item_amount", F.col("price"))
          .withColumn("item_amount_with_freight",
                      F.col("price") + F.col("freight"))
          .withColumn("category",
                      F.coalesce("category", F.lit("unknown"))))

monthly = (detail
           .withColumn("month", F.date_format("order_date", "yyyy-MM"))
           .groupBy("month")
           .agg(F.round(F.sum("item_amount_with_freight"), 2).alias("gmv"))
           .orderBy("month"))
monthly.show(20, truncate=False)

overview = detail.agg(
    F.countDistinct("order_id").alias("order_cnt"),
    F.round(F.sum("item_amount_with_freight"), 2).alias("gmv"),
    F.round(F.sum("item_amount"), 2).alias("goods_gmv"),
    F.round(F.sum("item_amount_with_freight")
            / F.countDistinct("order_id"), 2).alias("aov"),
)
overview.show()

category_gmv = (detail
                .groupBy("category")
                .agg(F.round(F.sum("item_amount_with_freight"), 2).alias("gmv"),
                     F.countDistinct("order_id").alias("order_cnt"))
                .orderBy(F.col("gmv").desc()))
category_gmv.show(10, truncate=False)
spark.stop()

运行 python scripts/spark_gmv.py。printSchema 应看到 order_date 最初是 string，DWD 里才转成 date。

3.2 月度 GMV 和总览

示例数据全部落在 2024-01：

overview 输出：

订单数是 9，不是 12。原始订单里有 2 个取消单和 1 个不可用单，排除后剩 9 个；每个有效订单都有明细，所以订单粒度和明细粒度在这一份数据里刚好相同。换成完整数据后，两者可能差几单，写报告要说明用的是哪个口径。

3.3 品类 TopN

category_gmv 使用含运费 GMV：

三个品类加起来正好是 1,509.00。品类 TopN 用 orderBy(...).limit(N)；每个品类内取 TopN 商品要用窗口函数，见练习 5。

3.4 与 DuckDB 对账

在项目根目录执行下面的脚本，用 DuckDB 按同一口径重算：

python - <<'PY'
import duckdb

con = duckdb.connect()
con.execute("""
CREATE VIEW orders AS
SELECT * FROM read_csv_auto('data/sample/orders_demo.csv', header=true);
CREATE VIEW items AS
SELECT * FROM read_csv_auto('data/sample/order_items_demo.csv', header=true);
""")
print(con.sql("""
    SELECT COUNT(DISTINCT o.order_id) AS order_cnt,
           ROUND(SUM(i.price + i.freight), 2) AS gmv,
           ROUND(SUM(i.price), 2) AS gmv_nofreight,
           ROUND(SUM(i.price + i.freight)
                 / COUNT(DISTINCT o.order_id), 2) AS aov
    FROM items i
    JOIN orders o USING (order_id)
    WHERE o.status NOT IN ('canceled', 'unavailable')
""").df())
PY

对账结果：

面试高频：两个引擎结果对不上，先查口径，不要先怀疑引擎。依次核对：订单状态是否都排除 canceled/unavailable？评价和支付是否先聚合再 join？GMV 是否都含运费？时间窗口是否一致？join 前后行数是否一致？日期类型和时区是否一致？这六步能解决大部分差异。

4. 常见坑

5. 练习

练习 1：商品品类统计

读 items，按 category 统计商品数和平均价格，按平均价格降序打印。

items = read_csv("order_items_demo.csv", items_schema)
(items.groupBy("category")
      .agg(F.countDistinct("product_id").alias("product_cnt"),
           F.round(F.avg("price"), 2).alias("avg_price"))
      .orderBy(F.col("avg_price").desc()).show())

练习 2：join 前为什么要确认粒度

detail 一行是一个订单商品，评价表一行是一条评价。两者按 order_id join 后，行数会变吗？什么情况下 GMV 会被算大？

练习 3：统计示例数据质量

分别统计重复 order_id、取消/不可用订单、缺品类明细、有效明细行数，并说明怎么处理。

print("重复 order_id:",
      orders.count() - orders.select("order_id").distinct().count())  # 0
print("取消/不可用订单:",
      orders.filter(F.col("status").isin("canceled", "unavailable")).count())  # 3
print("缺品类明细:",
      items.filter(F.col("category").isNull()).count())  # 0
print("有效明细:",
      items.join(orders.filter(~F.col("status").isin("canceled", "unavailable")),
                 "order_id").count())  # 9

练习 4：用 Spark SQL 算 2024-01 的 GMV

detail.createOrReplaceTempView("dwd_detail")
spark.sql("""
    SELECT ROUND(SUM(item_amount_with_freight), 2) AS gmv,
           COUNT(DISTINCT order_id) AS order_cnt
    FROM dwd_detail
    WHERE order_date >= DATE '2024-01-01'
      AND order_date <  DATE '2024-02-01'
""").show()

练习 5：每个品类 GMV Top3 商品

from pyspark.sql.window import Window
product_gmv = (detail
               .groupBy("category", "product_id")
               .agg(F.round(F.sum("item_amount_with_freight"), 2).alias("gmv")))
w = Window.partitionBy("category").orderBy(F.col("gmv").desc())
(product_gmv.withColumn("rn", F.row_number().over(w))
 .filter(F.col("rn") <= 3).orderBy("category", "rn")
 .show(30, truncate=False))

练习 6：写分区表并验证裁剪

把 detail 按 dt 写成 Parquet，读回后只查 2024-01，输出行数并确认分区裁剪。

spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")
out = detail.withColumn("dt", F.date_format("order_date", "yyyy-MM"))
(out.repartition(8, "dt").write.mode("overwrite")
     .partitionBy("dt").parquet("data/warehouse/dwd_order_detail"))
back = spark.read.parquet("data/warehouse/dwd_order_detail")
print(back.filter(F.col("dt") == "2024-01").count())   # 9
back.filter(F.col("dt") == "2024-01").explain(True)

6. 面试延伸

面试题：DataFrame 和 RDD 怎么选？

是什么：RDD 是无 schema 的分布式对象集合；DataFrame 是带 schema 的分布式表，能走 Catalyst 优化和 Tungsten 执行。

解决什么问题：结构化数据用 DataFrame 更省事、更快，优化器能做列裁剪、谓词下推、join 重排；RDD 适合非结构化数据和需要精细控制分区的场景。

我的实操边界：我在本地用 DataFrame 做过清洗和聚合，也用 RDD 写过 WordCount；没有做过 RDD 级别的内存和分区调优。

面试题：临时视图和 Hive 表有什么区别？

是什么：临时视图是当前 SparkSession 内的命名查询，只存在内存里；Hive 表有 Metastore 元数据和存储上的数据，跨会话、跨引擎可见。

解决什么问题：临时视图适合一个作业内部复用 DataFrame；Hive 表适合多个作业共享、需要权限和血缘管理的稳定资产。

我的实操边界：我在本地用 createOrReplaceTempView 串过多步 SQL，也看过 Hive 风格建表和分区；没有在生产 Metastore 上建过表，也没处理过权限和元数据同步。

面试题：Spark 和 DuckDB 算同一个指标，结果对不上，怎么排查？

是什么：两个引擎语法和默认行为不同，但统一口径后应该得到相同结果。对不上通常是口径、类型或 join 粒度出了问题。

解决什么问题：对账是数据开发的核心动作。先统一过滤条件、去重规则、金额公式和日期边界，再逐层比对行数和金额，才能定位差异。

我的实操边界：我在示例数据和项目线上做过 Spark 与 DuckDB 的 GMV、AOV 对账，能按“订单状态、评价/支付聚合、运费口径、时间窗口、join 行数、日期类型”六步排查；没有处理过生产级跨引擎数据漂移。

7. 本章验收

• 能创建和关闭 SparkSession，并解释 local[*] 的含义
• 能显式定义订单 CSV schema，并说出它比 inferSchema 稳在哪里
• 能写出 select、filter、withColumn、groupBy、agg、join、orderBy、limit 的实际用法
• 能注册临时视图，并用 Spark SQL 算出月度 GMV
• 能说出至少 4 类清洗点和处理方式，不会把业务空值当脏数据
• 能跑出 GMV 1,509.00、AOV 167.67，并与 DuckDB 结果对齐
• 能解释 repartition 和 coalesce 的区别，能写按 dt 分区的 Parquet 并验证分区裁剪
• 完成 6 道练习，至少 4 道能独立写出答案

下一章

Hive 基础：把 SQL 跑在 HDFS 上，理解表、元数据与分区。