数据校验最容易失控的地方，不是不会写 if，而是校验规则散落在 ETL 脚本、Notebook 和定时任务里，失败后也很难回答两个问题：哪些行出了问题？这套规则能不能在下一次任务中原样重跑？
Pointblank 提供了一种更集中的做法：用声明式验证计划描述数据质量要求，执行后查看验证结果，并把通过行与失败行分开处理。对于需要反复运行的批处理任务，还可以把计划放进 YAML，让校验逻辑脱离业务代码单独维护。
把数据质量规则写成验证计划
一个典型的订单数据集可能要求：订单号不能为空，金额必须大于零，订单状态只能取有限的几个值。与其在多个处理步骤中重复写条件，不如把这些要求组织成一条 Pointblank 验证链。
下面的示例使用 Pandas 创建数据，并通过 Pointblank 声明校验规则。运行前安装依赖：
python -m pip install pandas pointblank
可复制运行的示例：
import pandas as pd
import pointblank as pb
orders = pd.DataFrame(
{
"order_id": ["A001", "A002", None, "A004"],
"amount": [120.0, -5.0, 80.0, 42.0],
"status": ["paid", "pending", "paid", "unknown"],
}
)
validation = (
pb.Validate(data=orders)
.col_exists(columns=["order_id", "amount", "status"])
.col_vals_not_null(columns="order_id")
.col_vals_gt(columns="amount", value=0)
.col_vals_in_set(
columns="status",
set=["paid", "pending", "cancelled"],
)
.interrogate()
)
print(validation)
这里的每个验证步骤都表达一个明确的质量契约：
col_exists检查输入表是否包含预期字段。col_vals_not_null防止关键标识缺失。col_vals_gt检查数值下界。col_vals_in_set限制枚举字段的合法取值。interrogate执行整条验证计划并生成结果。
实际项目中，列名、允许的状态集合和数值阈值应该从业务契约或配置文件中来，而不是散落在脚本中的魔法值。
失败行不要直接丢掉
验证失败不等于数据没有价值。批处理系统通常需要把正常数据继续送入下游，同时把失败数据保存到隔离区，供修复、告警或人工复核。
Pointblank 负责定义并执行质量检查；如果应用需要精确地拆分行，可以结合相同规则使用 Pandas 构造布尔掩码。这样做的好处是数据分流逻辑清晰、输出结果可直接写入文件或数据库。
import pandas as pd
orders = pd.DataFrame(
{
"order_id": ["A001", "A002", None, "A004"],
"amount": [120.0, -5.0, 80.0, 42.0],
"status": ["paid", "pending", "paid", "unknown"],
}
)
valid_mask = (
orders["order_id"].notna()
& orders["amount"].gt(0)
& orders["status"].isin(["paid", "pending", "cancelled"])
)
clean_rows = orders.loc[valid_mask].copy()
failing_rows = orders.loc[~valid_mask].copy()
clean_rows.to_parquet("orders_clean.parquet", index=False)
failing_rows.to_parquet("orders_rejected.parquet", index=False)
print(f"clean={len(clean_rows)}, rejected={len(failing_rows)}")
这里有一个重要边界：不要让 Pointblank 的验证结果和分流条件各自维护一套、逐渐产生差异。可以把规则参数集中到一个配置对象，再分别生成 Pointblank 验证计划和分流掩码；或者在团队约定中明确，Pointblank 是质量报告的权威来源，Pandas 掩码只负责输出数据分区。
失败数据最好至少附带以下信息：批次号、源文件或表名、校验失败原因、发现时间和原始行标识。单独保存一份没有上下文的坏数据，后续很难定位上游问题。
用 YAML 让验证计划可重跑
当规则需要由数据工程师、分析师和业务人员共同维护时，YAML 比把条件硬编码在 Python 中更容易审查。可以把 YAML 作为团队自己的验证计划格式，再由 Python 读取配置并构造 Pointblank 计划。
例如建立 orders_validation.yml：
columns:
required:
- order_id
- amount
- status
rules:
order_id:
not_null: true
amount:
greater_than: 0
status:
allowed:
- paid
- pending
- cancelled
下面的代码展示一种轻量的配置驱动方式。它不是对所有 Pointblank 版本 YAML 接口的固定封装，项目应根据当前安装版本的 YAML API 调整加载部分；核心思路是保持规则配置与执行代码分离。
from pathlib import Path
import pandas as pd
import pointblank as pb
import yaml
def build_validation(data: pd.DataFrame, config: dict):
validation = pb.Validate(data=data)
required = config["columns"]["required"]
validation = validation.col_exists(columns=required)
rules = config["rules"]
if rules["order_id"].get("not_null"):
validation = validation.col_vals_not_null(columns="order_id")
if "greater_than" in rules["amount"]:
validation = validation.col_vals_gt(
columns="amount",
value=rules["amount"]["greater_than"],
)
if "allowed" in rules["status"]:
validation = validation.col_vals_in_set(
columns="status",
set=rules["status"]["allowed"],
)
return validation.interrogate()
config = yaml.safe_load(Path("orders_validation.yml").read_text())
orders = pd.read_csv("orders.csv")
result = build_validation(orders, config)
print(result)
安装 YAML 解析依赖：
python -m pip install pyyaml
python validate_orders.py
如果当前 Pointblank 版本支持直接从 YAML 加载验证计划，也可以采用其原生接口，并保留同样的配置版本管理方式。无论采用哪种加载路径，CI 或调度系统都应该固定依赖版本，并在每次运行中记录 YAML 文件版本、数据批次和验证摘要。
在数据管道中设置失败策略
验证计划真正有价值的地方，在于它会影响管道决策。不要只在日志里打印“校验失败”，而应该明确不同规则的处理方式：
- 结构性错误，例如缺少
order_id，通常应立即终止批次。 - 少量坏行，例如金额为负，可以隔离失败行并继续处理通过行。
- 统计性异常，例如空值比例突然升高，可以告警但暂不阻断，具体取决于业务容忍度。
可以在任务入口处加入一个简单的门禁逻辑。下面的 validation 结果对象具体属性可能随 Pointblank 版本变化，实际使用时应以当前版本文档和返回对象为准；示例重点是把“验证”和“管道决策”分成两个步骤。
import sys
result = build_validation(orders, config)
# 根据项目采用的 Pointblank 版本替换为对应的失败状态判断。
validation_failed = result.get("all_passed") is False if isinstance(result, dict) else False
if validation_failed:
print("validation failed; stop this batch", file=sys.stderr)
sys.exit(1)
print("validation passed; continue downstream processing")
生产环境中建议把验证摘要写入监控系统，至少包含通过率、失败规则、失败行数和批次标识。这样数据质量问题才能从一次性的脚本输出，变成可追踪的运行指标。
采用前的检查清单
Pointblank 适合把数据质量规则集中表达、执行和审查，但它不会自动决定什么是“正确数据”。落地时可以检查以下事项：
- 先定义关键列、合法范围和允许的异常比例。
- 为每条规则确定失败后的动作：终止、隔离还是告警。
- 让失败行保留原因和批次上下文，避免静默丢弃。
- 将验证计划放入版本控制，随数据模型变更一起评审。
- 固定 Pointblank、Pandas 和 YAML 解析器版本，避免返回结构或接口变化影响调度任务。
- 用一批包含正常值、空值、边界值和非法值的样本测试验证计划。
当规则数量增加时，声明式验证的价值会越来越明显：规则有名字、有执行结果，也能在不同批次之间重复运行。真正需要谨慎的是失败策略和配置版本，而不是把所有检查都堆进一条很长的验证链。