Spark：DataFrame 与 SQL 的数据契约
前置：Python函数、CSV字段和SQL分组。五列订单、batch.py与aggregate.sql见完整订单工程；本页解释其字段校验和日期规则。本文固定Apache Spark/PySpark 3.5.6版本族，运行前应核对所在集群版本；不是对4.x的兼容承诺。本轮只写作与静态解析，示例未执行，不启动Spark、JVM或外部服务。
Spark 显式 schema
Spark 显式 schema 把 CSV 原始列定义为字符串，再做可审计的类型转换；这能保留坏金额原文，而不是依赖样本推断。DataFrame 是带命名列、分布在执行器上的关系，不是 Python 进程里的列表。Python 使用 DataFrame API，没有 Scala/Java 的强类型 Dataset API。schema 描述读取结构，可空字段仍须业务校验。
本例五列分别是事件ID、客户、带偏移时间、金额、状态。先按字符串读取，再 trim、try_cast(amount AS DECIMAL(12,2))、to_timestamp。Decimal 避免二进制浮点直接参与钱款求和，但超出两位小数的输入会按类型规则舍入；若业务要求拒绝，须另加原文字法检查。FAILFAST只负责解析器认为损坏的记录，不能代替负金额、空客户、未知状态校验。输入表头顺序错误必须失败，不能把客户当金额静默读取。
Spark CSV不把所有多/少token行都认作corrupted。本例额外用csv.reader预检本地单文件的精确表头和每行五字段，防止多余列被忽略；这只是输入格式检查，不代替真实Spark计算。生产分片源不能用driver全量扫描，应另设分布式质量关口和输入快照。
Spark 空值与聚合
Spark 空值与聚合必须按三值逻辑判断：NULL = NULL不是true，WHERE只保留true。检查缺失用isNull；需要两边空也算相等的连接用SQL的<=>或eqNullSafe，不应随手把所有NULL填0。COUNT(*)统计行，COUNT(amount)忽略空值；SUM忽略空值，但全空组返回NULL，不等于业务已确认收入0。
问题：三行金额10、NULL、0，COUNT(*)=3、COUNT(amount)=2、SUM(amount)=10。若第二行是坏金额，它应进入隔离而不是在SUM中消失。本例先隔离金额转换为空的记录，才汇总PAID；这样报表不靠聚合默认行为掩盖输入缺陷。客户为空时也不能合成一个“未知客户”组后交付。
Spark SQL 与 DataFrame
Spark SQL 与 DataFrame 可表达同一关系计算，共用执行引擎。createOrReplaceTempView注册会话临时视图，不等于把数据写进持久数据库；会话停止后名称不再可用。代码将accepted_orders交给SQL，按UTC日期与客户分组，用COUNT和SUM汇总；用DataFrame的groupBy/agg写法可得到同样的关系结果。
列名与SQL字符串都应来自固定契约；不要把不可信输入拼成SQL表达式。explain能辅助核对列裁剪、过滤和Exchange，不能证明业务去重策略正确。SQL输出没有ORDER BY时不承诺顺序，即使上游CSV曾按客户排列。
Spark 时间戳日界
Spark 时间戳日界由会话时区参与日期提取，输入偏移决定时间点。固定spark.sql.session.timeZone为UTC，2026-01-01T23:30:00-02:00代表UTC次日01:30，因此收入归1月2日。偏移不是可以直接删掉的字符串装饰；对已经带偏移并解析成时刻的值，再无条件to_utc_timestamp可能造成二次位移。
示例拒绝无偏移、格式不符的时间，避免夏令时重叠时刻的隐式解释。若改为Asia/Shanghai，e1会跨到1月2日，alice当天收入从20变30。业务若需要商户本地日，应显式保存商户时区并设计归日规则，而不是让机器默认时区决定财务统计。
资料与版本核对
英文以官方3.5.6固定版本文档为准；中文作者资料用于术语与入门路径，不据旧版教程推断3.5新增API。核对日期：2026-10-17。
继续阅读
- 分区与流语义
- 批处理完整工程
- 分区与流语义