如何使用Great Expectations将数据校验批次拆分为通过和失败的两个数据流

完全可以实现你的需求，不需要因为部分数据校验失败中断整个ETL流程，Great Expectations本身支持行级校验结果查询，你可以直接基于校验结果拆分出干净数据和异常数据，分别写入目标表或者DataFrame。 #

核心逻辑 #

Great Expectations运行完校验后返回的ValidationResult对象中，会包含所有未通过Expectation规则的行标识信息，直接基于这些标识过滤源数据即可完成拆分。 #

两种常用场景的实现方式 #

场景1：源数据已加载为Pandas DataFrame #

• 正常执行数据校验，获取校验结果对象
• 从校验结果中提取所有未通过校验的行索引，合并去重后得到异常行索引集合
• 直接基于索引拆分DataFrame，示例代码如下：

# 从校验结果中提取所有异常行索引
unexpected_index_set = set()
for res in validation_result["results"]:
    if not res["success"] and "unexpected_index_list" in res["result"]:
        unexpected_index_set.update(res["result"]["unexpected_index_list"])

# 拆分数据
bad_df = df.loc[list(unexpected_index_set)]
good_df = df.drop(list(unexpected_index_set), axis=0)

场景2：直接对接Postgres数据源、无需全量拉取到本地 #

如果数据量较大不想全量加载到内存，可以在校验完成后直接在Postgres侧完成数据拆分：

• 首先从校验结果中提取所有未通过校验的行主键值，合并去重得到异常主键集合
• 构造SQL直接在数据库层完成数据分流，示例SQL如下：

-- 合格数据写入目标表
INSERT INTO 合格目标表 SELECT * FROM 源表 WHERE 主键列 NOT IN (异常主键值列表);

-- 异常数据写入异常表
INSERT INTO 异常目标表 SELECT * FROM 源表 WHERE 主键列 IN (异常主键值列表);

注意事项 #

• 配置Great Expectations校验时，将fail_on_validation_error参数设置为False，即可避免校验失败时自动中断ETL流程
• 可以给异常表新增failed_rule、batch_id等扩展字段，分别存储该行未通过的规则名称、对应校验批次ID，方便后续问题排查
• 如果你的Expectation规则包含表级校验（比如总行数校验），可以单独处理这类规则的校验结果，不需要关联行级数据拆分逻辑

内容的提问来源于stack exchange，提问作者MariaMadalina