Great Expectations声明式数据验证

import json # 声明式期望套件示例：AI训练数据质量规则suite = {    "suite_name": "training_data_quality",    "expectations": [        {            "expectation_type": "expect_column_values_to_not_be_null",            "kwargs": {"column": "text"}        },        {            "expectation_type": "expect_column_values_to_be_between",            "kwargs": {"column": "text_length", "min_value": 10, "max_value": 10000}        },        {            "expectation_type": "expect_column_values_to_be_unique",            "kwargs": {"column": "doc_id"}        }    ]}print(json.dumps(suite, indent=2, ensure_ascii=False))

from collections import defaultdict class ExpectationValidator:    """简化版期望验证器：执行期望套件并汇总结果"""     def __init__(self, suite):        self.suite = suite        self._handlers = {            "expect_column_values_to_not_be_null": self._check_not_null,            "expect_column_values_to_be_between": self._check_between,            "expect_column_values_to_be_unique": self._check_unique,        }     def validate(self, dataframe):        results = []        for exp in self.suite["expectations"]:            handler = self._handlers.get(exp["expectation_type"])            if handler:                results.append(handler(dataframe, exp["kwargs"]))        return results     def _check_not_null(self, df, kwargs):        col = kwargs["column"]        null_count = sum(1 for r in df if r.get(col) is None)        return {"success": null_count == 0, "observed": {"null_count": null_count}}     def _check_between(self, df, kwargs):        col = kwargs["column"]        vals = [r[col] for r in df if r.get(col) is not None]        out = sum(1 for v in vals if v < kwargs["min_value"] or v > kwargs["max_value"])        return {"success": out == 0, "observed": {"out_of_range": out}}     def _check_unique(self, df, kwargs):        col = kwargs["column"]        vals = [r[col] for r in df if r.get(col) is not None]        return {"success": len(vals) == len(set(vals)), "observed": {"duplicates": len(vals) - len(set(vals))}}

class Checkpoint:    """简化版Checkpoint：封装套件、数据源和验证动作"""     def __init__(self, name, suite_name, datasource, actions=None):        self.name = name        self.suite_name = suite_name        self.datasource = datasource        self.actions = actions or []     def run(self, suites, data_fetcher):        suite = suites[self.suite_name]        data = data_fetcher(self.datasource)        validator = ExpectationValidator(suite)        results = validator.validate(data)        success = all(r["success"] for r in results)        for action in self.actions:            action(results, success)        return {"success": success, "results": results}  def alert_action(results, success):    if not success:        failed = [r for r in results if not r["success"]]        print(f"[ALERT] 验证失败，{len(failed)}条期望未通过")  def store_action(results, success):    print(f"[STORE] 验证结果已存储，状态: {'通过' if success else '失败'}")

import jsonfrom collections import defaultdictfrom datetime import datetime  class Expectation:    """期望规则基类：声明式定义数据质量验证规则"""     def __init__(self, expectation_type, column, **kwargs):        self.expectation_type = expectation_type        self.column = column        self.kwargs = kwargs     def to_dict(self):        return {            "expectation_type": self.expectation_type,            "kwargs": {"column": self.column, **self.kwargs}        }  class ExpectationSuite:    """期望套件：管理一组期望规则的声明式配置"""     def __init__(self, name):        self.name = name        self.expectations = []     def add_expectation(self, expectation):        self.expectations.append(expectation)        return self     def to_dict(self):        return {            "suite_name": self.name,            "expectations": [e.to_dict() for e in self.expectations]        }  class ValidationResult:    """验证结果：记录每条期望的执行结果"""     def __init__(self, expectation_type, column, success, observed):        self.expectation_type = expectation_type        self.column = column        self.success = success        self.observed = observed        self.timestamp = datetime.now().isoformat()  class Checkpoint:    """Checkpoint：封装套件、数据源和验证动作的可调度单元"""     def __init__(self, name, suite, actions=None):        self.name = name        self.suite = suite        self.actions = actions or []     def run(self, data):        results = self._validate(data)        overall_success = all(r.success for r in results)        for action in self.actions:            action(results, overall_success)        return {"success": overall_success, "results": results}     def _validate(self, data):        results = []        for exp in self.suite.expectations:            handler = VALIDATORS.get(exp.expectation_type)            if handler:                success, observed = handler(data, exp)                results.append(ValidationResult(                    exp.expectation_type, exp.column, success, observed                ))        return results  def validate_not_null(data, exp):    null_count = sum(1 for r in data if r.get(exp.column) is None)    return null_count == 0, {"null_count": null_count}  def validate_between(data, exp):    vals = [r[exp.column] for r in data if r.get(exp.column) is not None]    out_of_range = sum(1 for v in vals                        if v < exp.kwargs["min_value"] or v > exp.kwargs["max_value"])    return out_of_range == 0, {"out_of_range": out_of_range}  def validate_unique(data, exp):    vals = [r[exp.column] for r in data if r.get(exp.column) is not None]    return len(vals) == len(set(vals)), {"duplicates": len(vals) - len(set(vals))}  VALIDATORS = {    "expect_column_values_to_not_be_null": validate_not_null,    "expect_column_values_to_be_between": validate_between,    "expect_column_values_to_be_unique": validate_unique,}  def alert_action(results, success):    if not success:        failed = [r for r in results if not r.success]        print(f"[ALERT] 验证失败: {len(failed)} 条期望未通过")  def store_action(results, success):    print(f"[STORE] 结果已存储, 状态: {'通过' if success else '失败'}")  # 构建期望套件suite = ExpectationSuite("training_data_quality")suite.add_expectation(Expectation("expect_column_values_to_not_be_null", "text"))suite.add_expectation(Expectation("expect_column_values_to_be_between", "text_length", min_value=10, max_value=10000))suite.add_expectation(Expectation("expect_column_values_to_be_unique", "doc_id")) # 模拟训练数据data = [    {"doc_id": 1, "text": "大模型训练数据", "text_length": 7},    {"doc_id": 2, "text": "数据质量验证", "text_length": 6},    {"doc_id": 3, "text": None, "text_length": 0},] # 创建并运行Checkpointcheckpoint = Checkpoint("training_gate", suite, [store_action, alert_action])report = checkpoint.run(data)print(json.dumps({"success": report["success"], "total": len(report["results"])}, ensure_ascii=False))