转换逻辑为什么要写测试
ETL 转换逻辑的错误大多不是任务崩溃，而是静默算错：金额口径改了一位、join 少了一层过滤、空值被当作 0 参与求和。这类问题上线后要靠对账才能暴露，排查成本很高。把转换逻辑当作代码来对待——用脱敏样本数据做测试，对边界值逐条断言——是拦截这类低频但致命错误最便宜的手段。
基本信息
| 适用范围 | SQL/Python/Spark 转换逻辑、维表关联、口径变更 |
|---|---|
| 样本来源 | 生产脱敏抽样，或手工构造边界用例 |
| 断言对象 | 行数、主键唯一、字段非空、金额非负、枚举范围 |
| 运行方式 | pytest + pandas 本地跑，或在数仓内跑稽核 SQL |
典型使用场景
- 修改指标口径时，先跑测试确认历史口径未被带偏
- join 逻辑调整（维表版本切换）后，验证未匹配率是否可接受
- 金额、税率、折扣等计算列变更，用边界值锁定结果
- 上线前回归：用最近一期真实数据跑一遍，对比行数与关键指标
- 跨库同步后校验：源端与目标端主键集合是否一致
pytest 断言示例
pytest 断言示例
import pandas as pd
def load(): # 每个用例独立加载样本，失败时好定位
return pd.read_parquet('/tmp/sample_dwd_order.parquet')
def test_amount_not_negative():
df = load()
assert (df['amount'] >= 0).all(), '存在负金额' # 退款冲正需单独建模，不能混进本表
def test_order_id_unique():
df = load()
assert df['order_id'].is_unique, 'order_id 有重复' # 主键唯一是下游去重的前提
def test_no_null_status():
df = load()
assert df['status'].notna().all(), '状态为空会让枚举统计失真'边界用例清单
| 用例 | 输入特征 | 期望表现 |
|---|---|---|
| 空值 | 关键字段为 NULL 或空串 | 按口径置默认值或丢弃，不能进分母 |
| 重复键 | 同一主键出现两行 | 去重后唯一，或直接触发告警 |
| 负值与零值 | 金额为负、数量为 0 | 按业务规则处理，不静默参与求和 |
| 超长字符串 | 超出目标字段长度 | 截断或报错，不能撑爆写入 |
| 时间边界 | 跨天、跨年、时区偏移的时间戳 | 归属日期按约定时区计算 |
把测试样本与断言纳入代码库，转换逻辑改动后先跑测试再提交，能拦住大部分低级口径错误；样本要小而覆盖边界，不要用整表数据跑，否则本地跑不动也没人愿意维护。
断言通过不等于业务正确：测试锁定的只是你写下的口径。口径本身理解错了，测试只会让错误稳定复现，关键指标的最终确认仍要和业务方对齐。