Great Expectations — 企业级数据质量验证框架¶

# pip 安装
pip install great-expectations         # 当前版本 1.x（GX Core）

# 验证
python -c "import great_expectations as gx; print(gx.__version__)"

import great_expectations as gx        # 导入
import pandas as pd

# 加载数据
df = pd.DataFrame({
    "age":    [25, 30, -1, 200, 45],   # 含异常值
    "email":  ["a@b.com", "bad", "c@d.com", None, "e@f.com"],
    "salary": [5000, 6000, 7000, 8000, None],
})

# 创建 GX 数据源（Ephemeral，适合快速测试）
context = gx.get_context()             # 获取上下文（内存模式）

# 从 pandas 创建数据资产
ds = context.data_sources.add_pandas("my_pandas")
asset = ds.add_dataframe_asset("patient_data")
batch_def = asset.add_batch_definition_whole_dataframe("batch")

batch = batch_def.get_batch(batch_parameters={"dataframe": df})

# 创建期望套件
suite = context.suites.add(gx.ExpectationSuite(name="patient_checks"))

# 添加各种期望规则
suite.add_expectation(gx.expectations.ExpectColumnValuesToBeBetween(
    column="age",               # 检查 age 列
    min_value=0,                # 最小值 0
    max_value=120,              # 最大值 120
))

suite.add_expectation(gx.expectations.ExpectColumnValuesToMatchRegex(
    column="email",             # 检查 email 列
    regex=r"^[^@]+@[^@]+\.[^@]+$",  # 邮箱格式正则
))

suite.add_expectation(gx.expectations.ExpectColumnValuesToNotBeNull(
    column="salary",            # salary 不能为空
))

suite.add_expectation(gx.expectations.ExpectColumnToExist(
    column="age",               # 列必须存在
))

# 创建验证定义
validation_def = context.validation_definitions.add(
    gx.ValidationDefinition(
        name       = "patient_validation",
        data       = batch_def,
        suite      = suite,
    )
)

# 执行验证
results = validation_def.run(batch_parameters={"dataframe": df})

# 查看结果
print(results.success)         # True/False：整体是否通过
for r in results.results:
    print(r["expectation_config"]["expectation_type"],
          "✓" if r["success"] else "✗",
          r.get("result", {}).get("unexpected_count", 0), "个异常值")

import great_expectations as gx
import pandas as pd

# 菌群丰度表验证
otu_df = pd.read_csv("otu_table.csv", index_col=0)

context = gx.get_context()
ds      = context.data_sources.add_pandas("bioinf")
asset   = ds.add_dataframe_asset("otu_table")
batch_def = asset.add_batch_definition_whole_dataframe("bd")

suite = context.suites.add(gx.ExpectationSuite(name="otu_checks"))

# OTU 表的质量规则
suite.add_expectation(gx.expectations.ExpectTableRowCountToBeBetween(
    min_value=10, max_value=10000        # 至少 10 个样本
))
suite.add_expectation(gx.expectations.ExpectColumnValuesToBeBetween(
    column="Shannon",                    # Shannon 多样性指数
    min_value=0.0, max_value=10.0
))
suite.add_expectation(gx.expectations.ExpectColumnValuesToNotBeNull(
    column="sample_id"                   # 样本 ID 不能缺失
))

vd = context.validation_definitions.add(
    gx.ValidationDefinition(name="otu_vd", data=batch_def, suite=suite)
)
results = vd.run(batch_parameters={"dataframe": otu_df})
print(f"数据质量检查：{'通过' if results.success else '失败'}")