数据质量问题往往不是计算错误，而是脏数据悄悄进入了后续流程：订单金额变成负数、必填字段为空、日期格式混乱，或者某批数据的行数突然大幅下降。Pointblank 为 Python 提供了一种声明式验证方式，让数据检查从散落在脚本里的 if 判断，变成可以审阅、执行和重复运行的验证计划。
这类验证流程通常包含三个动作：声明数据必须满足的规则，定位通过或失败的行，再把规则保存为 YAML，供批处理和 CI 流程重新执行。
把质量规则写成验证计划
Pointblank 的核心思路是围绕数据集构建 Validate 对象，然后链式添加检查。例如，下面的计划要求订单编号不为空、金额必须处于合理范围，并检查订单状态是否属于允许集合。
安装依赖：
python -m pip install pointblank pandas
下面的示例可以直接运行。示例中的 API 名称可能会随 Pointblank 版本演进，运行时应以当前版本文档为准；验证规则本身可以直接改造成项目中的订单、用户或日志数据检查。
from __future__ import annotations
import pandas as pd
import pointblank as pb
orders = pd.DataFrame(
{
"order_id": ["A100", "A101", None, "A103"],
"amount": [25.0, -3.0, 18.5, 120.0],
"status": ["paid", "paid", "pending", "unknown"],
}
)
validation = (
pb.Validate(data=orders)
.col_vals_not_null(columns="order_id")
.col_vals_between(columns="amount", left=0, right=100)
.col_vals_in_set(columns="status", set=["pending", "paid", "cancelled"])
)
# 执行验证并输出报告
validation.interrogate()
print(validation)
这里的重点不是把所有业务逻辑都塞进一个大函数，而是让每条规则都拥有清晰的语义：order_id 必须存在，amount 必须在区间内，status 必须来自受控集合。规则数量增加后，验证计划仍然可以按字段或业务主题组织。
通过行与失败行分开处理
验证报告适合回答“这批数据是否通过”，但生产任务还需要回答“哪些行失败了”。实际项目中可以让 Pointblank 负责声明和报告规则，同时使用相同的谓词生成通过数据和失败数据。这样既保留了验证框架的可读性，也能把坏行送到隔离表或人工修复队列。
下面的代码展示了一个完整的拆分模式：
from __future__ import annotations
import pandas as pd
import pointblank as pb
orders = pd.DataFrame(
{
"order_id": ["A100", "A101", None, "A103"],
"amount": [25.0, -3.0, 18.5, 120.0],
"status": ["paid", "paid", "pending", "unknown"],
}
)
validation = (
pb.Validate(data=orders)
.col_vals_not_null(columns="order_id")
.col_vals_between(columns="amount", left=0, right=100)
.col_vals_in_set(columns="status", set=["pending", "paid", "cancelled"])
)
validation.interrogate()
# 与验证计划保持一致的行级条件，用于生成数据产品
row_is_valid = (
orders["order_id"].notna()
& orders["amount"].between(0, 100)
& orders["status"].isin({"pending", "paid", "cancelled"})
)
clean_rows = orders.loc[row_is_valid].copy()
failing_rows = orders.loc[~row_is_valid].copy()
failing_rows["validation_error"] = "order_id、amount 或 status 未通过检查"
clean_rows.to_parquet("orders_clean.parquet", index=False)
failing_rows.to_parquet("orders_failed.parquet", index=False)
print(f"clean={len(clean_rows)}, failed={len(failing_rows)}")
这个模式有一个实际边界：如果同一条规则在 Pointblank 和拆分逻辑中分别手写，长期维护可能产生漂移。可以把列名、阈值和允许值提取到配置对象，再由验证计划和行过滤逻辑共同消费。对于更复杂的项目，也可以优先使用 Pointblank 当前版本提供的数据提取接口，而不是重复实现规则。
把验证计划放进 YAML
当检查规则需要在开发、测试和生产环境重复运行时，YAML 比硬编码在脚本里的链式调用更容易审阅和版本控制。下面是一份项目级验证计划示例。它描述了数据源、检查集合以及失败数据的输出位置。
# validation_plan.yaml
name: orders_quality
source:
path: data/orders.parquet
format: parquet
checks:
- type: not_null
column: order_id
- type: between
column: amount
left: 0
right: 100
- type: in_set
column: status
values: [pending, paid, cancelled]
outputs:
clean: output/orders_clean.parquet
failed: output/orders_failed.parquet
可以用一个很薄的加载器把 YAML 映射到 Pointblank 验证计划。这个示例采用显式映射，避免直接执行配置中的任意 Python 代码：
python -m pip install pointblank pandas pyyaml pyarrow
from __future__ import annotations
from pathlib import Path
import pandas as pd
import pointblank as pb
import yaml
def build_validation(data: pd.DataFrame, checks: list[dict]) -> pb.Validate:
plan = pb.Validate(data=data)
for check in checks:
kind = check["type"]
column = check["column"]
if kind == "not_null":
plan = plan.col_vals_not_null(columns=column)
elif kind == "between":
plan = plan.col_vals_between(
columns=column,
left=check["left"],
right=check["right"],
)
elif kind == "in_set":
plan = plan.col_vals_in_set(
columns=column,
set=check["values"],
)
else:
raise ValueError(f"Unsupported check type: {kind}")
return plan
config = yaml.safe_load(Path("validation_plan.yaml").read_text())
data = pd.read_parquet(config["source"]["path"])
validation = build_validation(data, config["checks"])
validation.interrogate()
print(validation)
YAML 并不自动解决规则版本管理问题。建议为配置文件和 Pointblank 版本建立明确的依赖约束，并在 CI 中使用一小份固定样本运行验证。这样，阈值变化、允许值变化或库升级造成的行为变化，都能在合并前暴露出来。
落地时的检查清单
- 把“字段不能为空”“数值范围”“枚举集合”等规则写成可读的声明式检查。
- 验证失败时保留原始行、规则名称和批次标识，方便追溯。
- 将通过数据和失败数据输出到不同位置，避免脏行继续进入下游任务。
- 对 YAML 验证计划做 schema 校验，只允许预先定义的检查类型。
- 在 CI 和生产调度中使用同一份验证计划，减少环境之间的规则差异。
- 把阈值、允许集合和 Pointblank 版本纳入代码评审与变更记录。
Pointblank 更适合作为数据质量层，而不是替代完整的数据契约、业务校验或异常监控。合理的组合方式是：用 Pointblank 声明并报告质量规则，用数据管道负责隔离失败行，再用 YAML 和 CI 保证这套规则可以稳定重跑。