如何在Great Expectations中创建自定义期望验证多唯一id_client（基于id_product）

需求说明 #

需要验证DataFrame中，基于指定分组列（如id_product），目标列（如id_client）是否存在至少一个分组包含多个唯一值，以此检测数据不一致情况。

数据示例 #

import pandas as pd

df = pd.DataFrame({
    'id_product': [1, 1, 2, 2, 2, 3, 3],
    'id_client': [101, 102, 201, 202, 203, 301, 301]
})

自定义期望实现 #

Great Expectations的自定义期望需继承Expectation类，并实现核心验证逻辑。以下是适配Pandas引擎的完整实现：

from great_expectations.core.expectation_configuration import ExpectationConfiguration
from great_expectations.expectations.expectation import Expectation
from great_expectations.execution_engine import PandasExecutionEngine

class ExpectGroupedColumnToHaveMultipleUniqueValues(Expectation):
    """验证指定分组列对应的目标列是否存在至少一个分组包含多个唯一值"""
    
    # 自定义期望标识与参数配置
    expectation_type = "expect_grouped_column_to_have_multiple_unique_values"
    success_keys = ("group_by_column", "target_column")
    default_kwarg_values = {}

    def _validate(self, configuration: ExpectationConfiguration, metrics, runtime_configuration=None):
        # 获取配置参数
        group_by_col = configuration.kwargs["group_by_column"]
        target_col = configuration.kwargs["target_column"]
        execution_engine = self.execution_engine

        # 仅支持Pandas引擎（可扩展至Spark等其他引擎）
        if isinstance(execution_engine, PandasExecutionEngine):
            df = execution_engine.engine.dataframe

            # 复用Pandas验证逻辑
            unique_pairs = df.groupby([group_by_col, target_col]).size().reset_index()
            group_unique_count = unique_pairs.groupby(group_by_col).size().reset_index(name='count')
            
            # 判断是否存在符合条件的分组
            has_multiple_unique = any(group_unique_count['count'] > 1)

            # 返回验证结果
            return {
                "success": has_multiple_unique,
                "result": {
                    "observed_value": has_multiple_unique,
                    "details": {
                        "groups_with_multiple_unique": group_unique_count[group_unique_count['count']>1][group_by_col].tolist(),
                        "total_groups_checked": len(group_unique_count)
                    }
                }
            }
        else:
            raise NotImplementedError("当前仅支持PandasExecutionEngine")

使用自定义期望 #

代码方式添加期望 #

在你的Great Expectations Suite中直接调用：

# 假设已初始化suite对象
suite.add_expectation(
    ExpectGroupedColumnToHaveMultipleUniqueValues(
        group_by_column="id_product",
        target_column="id_client"
    )
)

YAML配置方式 #

若使用数据文档配置，可在suite的YAML文件中添加：

expectations:
  - name: expect_grouped_column_to_have_multiple_unique_values
    kwargs:
      group_by_column: id_product
      target_column: id_client

验证逻辑说明 #

• 先按group_by_column和target_column分组，统计唯一组合
• 再按group_by_column分组，统计每个分组下的唯一值数量
• 检查是否存在数量大于1的分组，以此判断验证是否成功
• 返回结果包含成功状态及详细信息（符合条件的分组列表、总检查分组数）

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