Spark：订单批处理完整工程与推导

前置：会读Python函数、SQL分组和文件路径。本文固定Apache Spark/PySpark 3.5.6版本族，运行前应核对所在集群版本；不是对4.x的兼容承诺。本轮只写作与静态解析，示例未执行，不启动Spark、JVM或外部服务。

Spark 订单批处理工程

Spark 订单批处理工程输入11行CSV，输出UTC日客户汇总与坏行隔离目录。实现使用真实PySpark API与独立SQL文件，金额用Decimal；没有用Python列表模拟分布式计算。完整依赖、配置、输入、预期表与README在下文同源展开。状态是未执行：本轮只做AST、JSON与CSV解析，未导入pyspark，也未启动JVM。

业务顺序是读取、规范化、字段校验、排除取消事件、折叠完全相同事件、隔离同ID冲突、汇总。取消但格式损坏的行仍隔离。两个输出目录不具有联合原子性；COMPLETE最后写入，仅辅助辨认练习是否写完，不能替代事务提交协议。

Spark 订单结果推导

Spark 订单结果推导从11行逐项扣除：4行无效、1行取消、1行完全重复、2行同ID冲突，接受3个事件。金额总和35；UTC下1月1日alice=10，1月2日alice=20、bob=5。e3带-02:00偏移，归UTC次日。rejected应有6条：4个invalid_field、2个conflicting_event_id。

具体题目：把CSV倒序后哪项结果会变？答案：业务汇总与隔离数量不变，但文件顺序、分片名及被保留的等价原文字段可能变化。accepted汇总仅用规范化业务字段，不依赖原文字段。新增e2金额21会使该ID整组隔离，alice的1月2日行消失；这验证冲突策略与任意first不同。

参考答案：为何不能用 first 解决冲突

同一事件ID的两个金额代表业务歧义；没有版本号就无法确定哪个正确。整组隔离使下游收入可追溯，而任意first会随执行顺序变化。

Spark 批任务恢复清理

Spark 批任务恢复清理以输入快照和新目录重跑为单位。输出路径存在时先报错，避免覆盖；若totals写完但rejected失败，没有COMPLETE就按未完成处理。先保存错误与输入，再选新目录重算，不能盲目追加导致重复收入。真实生产需版本化输入、幂等发布、事务表或受控manifest协议。

清理只删除本例自己创建且已确认不需保留的输出与虚拟环境；不要删除共享JDK、原始CSV、共享checkpoint或其他作业输出。README给出安装、自跑、对照、失败和清理步骤，命令是未来读者的操作文本，本轮没有执行。停止SparkSession并不能撤销已经写出的目录。

完整源码、配置、输入与说明

所有文件与仓库工程同源include，未执行。README包含前置、命令、人工预期、失败与清理。

requirements.txt

pyspark==3.5.6

config.json

{
  "app_name": "p2-spark-orders",
  "master": "local[2]",
  "timezone": "UTC",
  "shuffle_partitions": 4
}

orders.csv

event_id,customer,occurred_at,amount,status
e1,alice,2026-01-01T23:30:00+00:00,10.00,PAID
e1,alice,2026-01-01T23:30:00+00:00,10.00,PAID
e2,alice,2026-01-02T08:00:00+00:00,20.00,PAID
e3,bob,2026-01-01T23:30:00-02:00,5.00,PAID
e4,,2026-01-02T09:00:00+00:00,8.00,PAID
e5,bob,2026-01-02T09:00:00+00:00,nope,PAID
e6,bob,2026-01-02T09:00:00+00:00,7.00,PAID
e6,bob,2026-01-02T09:00:00+00:00,8.00,PAID
e7,bob,not-a-time,9.00,PAID
e8,bob,2026-01-02T09:00:00+00:00,6.00,CANCELLED
e9,bob,2026-01-02T09:00:00+00:00,-1.00,PAID

expected.csv

day,customer,orders,revenue
2026-01-01,alice,1,10.00
2026-01-02,alice,1,20.00
2026-01-02,bob,1,5.00

aggregate.sql

SELECT CAST(event_ts AS DATE) AS day, customer,
       COUNT(*) AS orders, CAST(SUM(amount) AS DECIMAL(18,2)) AS revenue
FROM accepted_orders
GROUP BY CAST(event_ts AS DATE), customer

batch.py

"""Real PySpark batch example; authored and AST-checked, NOT executed."""
import argparse
import csv
import json
from pathlib import Path

from pyspark.sql import SparkSession, functions as F, types as T


def run(input_path, output_path, config_path):
    base = Path(__file__).resolve().parent
    config = json.loads(Path(config_path).read_text(encoding="utf-8"))
    # Reserve a new destination; a failed partial run must be inspected first.
    output = Path(output_path).resolve()
    if output.exists():
        raise FileExistsError("Output already exists; inspect it and choose a new path")
    # Local single-file teaching input only. Spark CSV FAILFAST alone does not
    # reject every row with extra/missing tokens; check our narrower contract.
    with Path(input_path).open(encoding="utf-8", newline="") as stream:
        rows = csv.reader(stream, strict=True)
        header = ["event_id", "customer", "occurred_at", "amount", "status"]
        if next(rows, None) != header:
            raise ValueError("Expected the exact five-column header")
        for line_number, row in enumerate(rows, 2):
            if len(row) != len(header):
                raise ValueError(f"CSV record {line_number}: expected five fields")
    schema = T.StructType([
        T.StructField(name, T.StringType(), True)
        for name in ["event_id", "customer", "occurred_at", "amount", "status"]
    ])
    spark = (SparkSession.builder.appName(config["app_name"])
             .master(config["master"])
             .config("spark.sql.session.timeZone", config["timezone"])
             .config("spark.sql.shuffle.partitions", str(config["shuffle_partitions"]))
             .config("spark.sql.ansi.enabled", "false")
             .config("spark.sql.legacy.timeParserPolicy", "CORRECTED")
             .getOrCreate())
    try:
        # Input stays immutable after preflight; parser errors still fail actions.
        raw = (spark.read.schema(schema).option("header", "true")
               .option("enforceSchema", "false").option("mode", "FAILFAST")
               .csv(str(Path(input_path).resolve())))
        parsed = (raw.select(*[F.trim(F.col(c)).alias(c) for c in raw.columns])
                  .withColumn("amount_raw", F.col("amount"))
                  .withColumn("amount", F.expr("try_cast(amount AS DECIMAL(12,2))"))
                  .withColumn("event_ts", F.to_timestamp("occurred_at", "yyyy-MM-dd'T'HH:mm:ssXXX")))
        invalid = (F.col("event_id").isNull() | (F.col("event_id") == "")
                   | F.col("customer").isNull() | (F.col("customer") == "")
                   | F.col("event_ts").isNull() | F.col("amount").isNull()
                   | (F.col("amount") < 0)
                   | F.col("status").isNull()
                   | ~F.col("status").isin("PAID", "CANCELLED"))
        bad = parsed.filter(invalid).withColumn("reason", F.lit("invalid_field"))
        # Deduplicate normalized business fields, never choose an arbitrary first row.
        paid = parsed.filter(~invalid & (F.col("status") == "PAID"))
        distinct = paid.dropDuplicates(["event_id", "customer", "event_ts", "amount", "status"])
        conflicts = (distinct.groupBy("event_id").count()
                     .filter(F.col("count") > 1).select("event_id"))
        conflicting = distinct.join(conflicts, "event_id", "inner")
        accepted = distinct.join(conflicts, "event_id", "left_anti")
        rejected = bad.unionByName(conflicting.withColumn("reason", F.lit("conflicting_event_id")))
        accepted.createOrReplaceTempView("accepted_orders")
        result = spark.sql((base / "aggregate.sql").read_text(encoding="utf-8"))
        # show() is a bounded action; no collect() of unbounded input.
        result.orderBy("day", "customer").show(20, truncate=False)
        result.explain(mode="formatted")
        result.write.mode("errorifexists").option("header", "true").csv(str(output / "totals"))
        rejected.write.mode("errorifexists").json(str(output / "rejected"))
        # Written last; two writes are not a multi-directory transaction.
        (output / "COMPLETE").write_text("Both outputs written; verify contents before publishing.\n", encoding="utf-8")
    finally:
        spark.stop()


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--input", default=str(Path(__file__).with_name("orders.csv")))
    parser.add_argument("--config", default=str(Path(__file__).with_name("config.json")))
    parser.add_argument("--output", required=True)
    args = parser.parse_args()
    run(args.input, args.output, args.config)

README.md

# PySpark 3.5.6：UTC 订单日汇总

状态：**未执行**。本轮未安装依赖、未启动 Python 示例、Spark、JVM 或服务；只解析源码 AST、JSON、CSV 和文档引用。下面命令是读者将来在独立练习环境自行执行的步骤，不是成功日志。

## 前置和契约

建议 Linux/macOS 或 WSL2，Python 3.11、JDK 17，正确设置 JAVA_HOME。Spark 3.5.6 官方支持 Python >=3.8、Java 8（至少8u371）/11/17；此包固定 pyspark==3.5.6，不把 Spark 4.x 当兼容版本。本包仅本地 local[2]，没有集群、Hive、Kafka、账号与网络源。pip 安装步骤需要读者联网；仓库作者未执行它。标准库外仅 PySpark；不需要 pandas/Arrow。

输入是 UTF-8、逗号分隔、带精确五列表头 CSV：event_id/customer/occurred_at/amount/status，原始 schema 全是可空 StringType。字符串 trim 后，时间必须含显式偏移，格式 yyyy-MM-dd'T'HH:mm:ssXXX；金额转换为 Decimal(12,2)，使用 try_cast 避免坏数字直接中断。这个契约允许 Spark 的 decimal 四舍五入；若业务禁止超过两位小数，必须在转换前增加词法校验。CSV 解析器 FAILFAST 负责结构错误；空字段、无效时间、负金额和未知状态是业务无效行，输出 reason=invalid_field。schema 可空是读取边界，不是有效性保证。

## 完整自跑步骤

在仓库根目录打开终端。先人工确认 java/python 版本与 JAVA_HOME，不要在生产机器直接练习。以下 bash 命令会安装 PySpark 并启动 JVM，只在读者明确允许后运行：

```bash
python3.11 -m venv examples/p2-spark/.venv
examples/p2-spark/.venv/bin/python -m pip install -r examples/p2-spark/requirements.txt
examples/p2-spark/.venv/bin/python examples/p2-spark/batch.py --output examples/p2-spark/out-run1
```

Windows 原生 Python 环境若已有正确 Java/Hadoop 本地文件系统支持，可以将解释器替换为 `.venv/Scripts/python.exe`；推荐 WSL2 避免把本机路径兼容问题误判为 SQL 错误。不要混用系统 Spark 和另一版本 pip PySpark。

预期控制台有三行汇总与 formatted 计划（实际 stage、AQE 与执行时间依环境变化）。输出 totals 是**目录**，含任意多个 part 文件和 _SUCCESS；不可假设 part 文件名、数量、行顺序。将各 CSV 数据文件按 day/customer 排序后与 expected.csv 人工对照；rejected 下 JSON 应有四个 invalid_field 行及两个 conflicting_event_id 行；COMPLETE 必须存在。marker 只是本地练习完成标志，不是业务发布事务。该包没有运行验证结果。

## 人工推导

原始11行。e4空customer、e5坏金额、e7坏时间、e9负金额共4行无效；e8取消不计收入。剩余6行中 e1 的完全重复折叠为1行；e6两个不同金额属于冲突，将整组2行隔离，最后接受e1/e2/e3共3个事件。UTC下 e1 是1月1日、alice收入10；e2是1月2日、alice收入20；e3虽原文写1月1日23:30，偏移-02:00换算UTC是1月2日01:30，因此bob当天收入5。合计35，不把重复的10或冲突的7/8加入。

本例只在单批快照内去重，跨两次输入的重复不被自动识别；dropDuplicates是业务字段等价，不是 exactly-once sink。改成 dropDuplicates(["event_id"]) 会随机保留冲突版本，因而不采用。合法性检查先于状态过滤，取消但损坏的记录也应隔离。

## 失败、恢复与清理

再次指定同一 out-run1，预期在创建 Spark 前报 FileExistsError；因此不会覆盖旧结果。输入不存在、表头错位、字段数不等于5或CSV引号结构错误，由本地csv.reader预检在创建Spark前失败；金额 nope 则不属于结构错误，进入隔离。Spark CSV的少/多token记录并不总被定义为corrupted，因此不能只依赖FAILFAST保证行宽。预检只适用于本地单文件教学输入，不适合driver扫描生产级大文件；扩展到远程分片必须重新设计分布式质量校验。预检后到所有action完成前输入必须保持不可变。失败可能发生在 totals 完成而 rejected 未完成之间：没有 COMPLETE 时把整个目录视为未完成，先保存错误日志与输入快照，使用新的 out-run2 从头重跑。不可只补写一半目录，也不可宣称两个目录原子提交。发布到真实存储需另设提交协议或支持事务的表格式。

清理前确认输出位于本例目录且不需要留存，逐项删除练习创建的 out-run1/out-run2 与 .venv；不要删除输入、源码、全局 Java 或共享 checkpoint。示例不创建 streaming checkpoint，不需要停外部服务。SparkSession 在 finally 中 stop，但硬杀进程的残留须由读者确认。

## 练习

1. 将config的timezone改为Asia/Shanghai，会怎样？e1变为1月2日07:30，alice的1月2日合计30，bob仍5。时间戳代表时刻，日界由会话时区决定。
2. 交换CSV行序会怎样？聚合值不变，part行序可能改变；不能拿文件字节相等当正确性标准。
3. 再加一条e2金额21会怎样？e2整组冲突被隔离，alice的1月2日汇总消失；这比任意first保留可解释。
4. 为什么不用collect核对全量？它把结果拉到driver，数据规模增加会耗尽driver内存；有限show不代表全量正确，生产核对应用分布式对账。

资料与版本核对

英文以官方3.5.6固定版本文档为准；中文作者资料用于术语与入门路径，不据旧版教程推断3.5新增API。核对日期：2026-10-17。

• PySpark安装约束
• CSV读取选项

返回 Spark 专题 · 继续学习路线