基于Great Expectations优化AWS行计数与列鲜度校验结果展示

我在AWS环境中通过以下流程做数据校验：

• 用Great Expectations（GE）实现行计数（expect_table_row_count_to_be_between）和自定义列鲜度（custom_expect_column_to_be_fresh）校验
• 通过check_config.json配置校验规则
• 用Terraform创建Glue任务执行校验，结果写入DynamoDB

当前DynamoDB仅存储了校验通过/失败的状态，缺少关键实际数据：比如行计数的实际数值、列鲜度校验对应的表最后修改时间，无法直观判断校验失败的具体原因。

当前check_config.json配置如下：

{
    "table": "table1",
    "checks": [
        {
            "check": "custom_expect_column_to_be_fresh",
            "parameters": {
                "columns": [
                    "column1"
                ],
                "strftime_format": "%Y-%m-%d",
                "threshold_days": 0,
                "threshold_hours": 10
            }
        },
        {
            "check": "expect_table_row_count_to_be_between",
            "result_format" : "COMPLETE",
            "include_config": "True",
            "parameters": {
                "min_value": 1,
                "max_value": 100000
            },
            "alarm" : {
                "threshold": 100,
                "period": 3600
            }
        }
    ]
}

解决方案 #

要在DynamoDB中存储实际校验值，需要从GE结果配置和Glue任务数据写入逻辑两方面调整：

1. 调整GE校验配置，获取完整结果 #

确保所有校验规则都设置result_format": "COMPLETE"，这样GE会返回包含实际计算值的完整结果对象。修改后的check_config.json如下：

{
    "table": "table1",
    "checks": [
        {
            "check": "custom_expect_column_to_be_fresh",
            "result_format": "COMPLETE", // 新增：要求返回完整结果
            "include_config": "True",
            "parameters": {
                "columns": [
                    "column1"
                ],
                "strftime_format": "%Y-%m-%d",
                "threshold_days": 0,
                "threshold_hours": 10
            }
        },
        {
            "check": "expect_table_row_count_to_be_between",
            "result_format" : "COMPLETE",
            "include_config": "True",
            "parameters": {
                "min_value": 1,
                "max_value": 100000
            },
            "alarm" : {
                "threshold": 100,
                "period": 3600
            }
        }
    ]
}

2. 修改Glue任务逻辑，提取并写入实际值 #

GE的COMPLETE格式结果中，result字段包含实际计算值，需要在Glue任务中提取这些值后再写入DynamoDB。

示例Glue任务代码片段（处理结果部分）：

import boto3
from datetime import datetime

# 假设已经通过GE执行校验得到results列表
for check_result in results:
    check_name = check_result["expectation_config"]["expectation_type"]
    status = check_result["success"]
    actual_value = None

    # 提取行计数实际值
    if check_name == "expect_table_row_count_to_be_between":
        actual_value = check_result["result"]["observed_value"]
    # 提取鲜度校验的最后修改时间
    elif check_name == "custom_expect_column_to_be_fresh":
        # 需确保自定义expectation返回了该字段
        actual_value = check_result["result"].get("last_modified_time")

    # 构造DynamoDB条目
    item = {
        "table_name": {"S": "table1"},
        "check_type": {"S": check_name},
        "status": {"S": "PASS" if status else "FAIL"},
        "actual_value": {"N" if isinstance(actual_value, int) else "S": str(actual_value)},
        "timestamp": {"S": datetime.utcnow().isoformat()}
    }

    # 写入DynamoDB
    dynamodb = boto3.client("dynamodb")
    dynamodb.put_item(TableName="your-validation-results-table", Item=item)

3. 确保自定义鲜度校验返回实际时间 #

如果custom_expect_column_to_be_fresh是自定义实现，需要在返回结果中明确包含最后修改时间：

from great_expectations.expectations.expectation import ColumnMapExpectation

class CustomExpectColumnToBeFresh(ColumnMapExpectation):
    # 省略其他必要实现代码
    def _validate(self, configuration, metrics, runtime_configuration=None):
        # 逻辑：获取表/列的最后修改时间
        last_modified_time = self._fetch_last_modified_time(metrics)
        # 校验是否符合鲜度要求
        success = self._check_freshness(last_modified_time, configuration)

        return {
            "success": success,
            "result": {
                "observed_value": last_modified_time,
                "last_modified_time": last_modified_time  # 明确返回实际时间
            }
        }

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