4. 开发第一个自定义 Expectation：校验列中所有值均为正数
好，我们正式开始动手写代码。
前面几章我们把 Great Expectations 的架构、插件体系、以及 Expectation 的生命周期都捋了一遍。说实话，理论讲再多，不如亲手写一个来得实在。我个人习惯是，学一个新框架，先照着文档抄一个最小的例子跑通，然后再去理解它为什么这么设计。
这一章我们就来干这件事——写一个自定义 Expectation，功能很简单：校验某一列的所有值是否都大于 0。你别看它简单，麻雀虽小五脏俱全。写完这个，后面那些复杂的、带自定义 Metrics 的、甚至需要跑 Spark 的 Expectation，你都能举一反三。
4.1 先想清楚我们要做什么
在敲键盘之前，我们先明确一下目标。
这个 Expectation 叫 expect_column_values_to_be_positive。它的输入是一列数据，输出是一个 Expectation Validation Result。如果列里所有值都大于 0，就通过；但凡有一个 ≤ 0 的，就不通过。
你可能会问：「Great Expectations 不是已经有 expect_column_values_to_be_between 了吗？把下限设成 0 不就行了？」
嗯，理论上可以。但我在项目中遇到过几次这种情况：业务方要求「金额必须为正数」，但数据里偶尔会出现 0 或者负数。如果用 expect_column_values_to_be_between，你得手动把下限设成 0.000001，还得处理边界问题。更关键的是，语义不清晰。你写一个 expect_column_values_to_be_positive，任何人一看名字就知道你在检查什么。这就是自定义 Expectation 的价值——把业务规则显式化。
核心原则：自定义 Expectation 不是为了炫技，而是为了把领域知识编码成可复用、可共享的数据质量规则。
4.2 项目结构准备
Great Expectations 的自定义插件有固定的目录结构。我建议你按照官方推荐的方式来组织代码，这样后续升级框架版本时不容易出问题。
在你的 Great Expectations 项目根目录下，创建如下结构：
great_expectations/
├── plugins/
│ └── expectations/
│ └── expect_column_values_to_be_positive.py
└── ... 其他目录
plugins/expectations/ 这个路径是 Great Expectations 默认会扫描的插件目录。你把自定义 Expectation 放进去，框架启动时就会自动加载。
小技巧：如果你用的是 Great Expectations Cloud 或者部署在容器里，记得把 plugins 目录挂载进去。我曾经因为忘记挂载，排查了半天为什么自定义 Expectation 找不到……
4.3 编写 Expectation 类
现在打开 expect_column_values_to_be_positive.py，开始写代码。
一个自定义 Expectation 的核心，是继承 ExpectColumnValuesToBeBetween 或者更通用的 ColumnMapExpectation。我们这里用 ColumnMapExpectation，因为它天然适合「对每一行做判断」的场景。
from great_expectations.expectations.expectation import ColumnMapExpectation
from great_expectations.expectations.metrics import column_condition_partial
class ExpectColumnValuesToBePositive(ColumnMapExpectation):
"""Expect all values in a column to be positive (greater than 0)."""
# 1. 定义 metric 的计算逻辑
@column_condition_partial(engine="pandas")
def _pandas(cls, column, **kwargs):
return column > 0
# 2. 定义期望的元数据
map_metric = "_pandas"
success_keys = tuple()
default_kwarg_values = {}
# 3. 定义库信息（用于文档和 UI 展示）
library_metadata = {
"maturity": "experimental",
"tags": ["custom", "positive-values"],
"contributors": ["your_name"],
}
代码不长，但有几个关键点我得强调一下。
4.3.1 @column_condition_partial 装饰器
这个装饰器告诉 Great Expectations：我这个 Expectation 的 Metric 计算逻辑，是对每一列数据应用一个条件判断。你传入一个 Pandas Series，返回一个布尔 Series。框架会自动帮你处理聚合、统计、生成结果对象这些事情。
如果你需要支持 Spark 或者 SQLAlchemy，可以再加一个 @column_condition_partial(engine="spark") 或者 @column_condition_partial(engine="sqlite")。不过我们这一章先只做 Pandas，把流程跑通再说。
4.3.2 map_metric 和 success_keys
map_metric 指向你定义的 Metric 方法名。这里我们写的是 "_pandas"，所以 map_metric = "_pandas"。
success_keys 是一个元组，用来声明这个 Expectation 有哪些参数会影响「成功」的判断。我们这个例子没有任何额外参数，所以是空元组。如果你要写一个 expect_column_values_to_be_positive_with_threshold，允许用户指定「允许有多少比例的非正数」，那 success_keys 里就要加上 "mostly"。
注意：success_keys 和 default_kwarg_values 必须匹配。如果你在 success_keys 里声明了 "mostly"，但 default_kwarg_values 里没有给默认值，框架会报错。我刚开始写的时候踩过这个坑，排查了半天。
4.4 注册并测试
代码写完了，怎么让它生效？
Great Expectations 会自动扫描 plugins/expectations/ 目录下的所有 Python 文件，并加载其中继承自 Expectation 的类。你只需要重启你的 Great Expectations 进程（或者重新加载 Notebook 内核），就可以在 expect_* 的自动补全里看到它了。
我们来写一个简单的测试：
import pandas as pd
from great_expectations.dataset import PandasDataset
df = pd.DataFrame({"amount": [100, 200, 300, -50, 0]})
ds = PandasDataset(df)
result = ds.expect_column_values_to_be_positive("amount")
print(result)
输出应该是：
{
"success": False,
"result": {
"element_count": 5,
"unexpected_count": 2,
"unexpected_percent": 40.0,
"unexpected_list": [-50, 0]
}
}
看到 success: False，说明我们的 Expectation 生效了。它准确地识别出了 -50 和 0 这两个非正数。
调试技巧：如果你发现自定义 Expectation 没有被加载，先检查一下 great_expectations --version 和你的 Python 环境。有时候是因为虚拟环境没切对，插件目录没在正确的项目下。我习惯在项目根目录下运行 python -c "from great_expectations.expectations import Expectation; print(Expectation.__subclasses__())" 来确认自定义类是否被注册成功。
4.5 理解背后的执行流程
光会写还不够，你得知道它跑起来的时候内部发生了什么。我画了一张图，帮你理清这个流程。
你看，整个流程其实很清晰。开发者只需要关注「条件逻辑」那一块——也就是 column > 0 这行代码。剩下的解析、聚合、结果生成，框架全帮你干了。这就是 Great Expectations 的设计哲学：把复杂留给自己，把简单留给用户。
4.6 进阶：添加参数支持
上面那个例子是最简版本。实际项目中，你很可能需要支持 mostly 参数——允许一定比例的数据不满足条件。比如「允许 5% 的金额为 0 或负数」，这在业务上很常见。
我们来升级一下：
from great_expectations.expectations.expectation import ColumnMapExpectation
from great_expectations.expectations.metrics import column_condition_partial
class ExpectColumnValuesToBePositive(ColumnMapExpectation):
@column_condition_partial(engine="pandas")
def _pandas(cls, column, **kwargs):
return column > 0
map_metric = "_pandas"
# 添加 mostly 参数支持
success_keys = ("mostly",)
default_kwarg_values = {
"mostly": 1.0, # 默认必须全部为正数
}
library_metadata = {
"maturity": "experimental",
"tags": ["custom", "positive-values"],
"contributors": ["your_name"],
}
改动很小，就两处：
success_keys加上了"mostly"default_kwarg_values给了默认值1.0（100% 必须通过）
现在你可以这样调用：
result = ds.expect_column_values_to_be_positive("amount", mostly=0.95)
如果 100 行数据里有 3 行是负数，mostly=0.95 意味着最多允许 5% 的异常，3% 在范围内，所以 success 会是 True。
经验之谈：我在做金融风控数据质量监控时，经常用 mostly 参数来设置「容忍度」。因为生产环境的数据永远不可能 100% 干净，设置一个合理的容忍度，可以避免因为零星脏数据导致整个 pipeline 报警中断。
4.7 常见问题与避坑
写自定义 Expectation 的过程中，有几个坑我帮你提前踩了。
| 问题 | 原因 | 解决方案 |
|---|---|---|
| 自定义 Expectation 找不到 | 插件目录不在框架扫描路径中 | 检查 plugins/expectations/ 是否存在，重启进程 |
| Metric 返回类型不对 | @column_condition_partial 期望返回布尔 Series |
确保你的条件表达式返回的是 pd.Series(bool) |
success_keys 与 default_kwarg_values 不匹配 |
框架在合并参数时发现缺失默认值 | 检查两个字段的 key 是否完全一致 |
| 升级框架版本后报错 | API 有变动，比如 column_condition_partial 的导入路径变了 |
查看官方 Changelog，更新导入路径 |
我曾经在升级 Great Expectations 0.13 到 0.14 时，发现 column_condition_partial 的导入路径变了，导致所有自定义 Expectation 全部失效。那次教训让我养成了一个习惯：每次升级框架前，先在测试环境跑一遍所有自定义 Expectation 的单元测试。
4.8 小结
这一章我们走完了自定义 Expectation 的完整流程：从项目结构、代码编写、注册测试，到理解执行流程和添加参数支持。你写出来的 expect_column_values_to_be_positive 虽然简单，但它包含了自定义 Expectation 的所有核心要素。
说白了，你只要掌握了这个套路，后面无论是写复杂的统计型 Expectation，还是对接 Spark 引擎，本质上都是在做同样的事情——定义 Metric 逻辑，声明参数，然后让框架帮你处理剩下的。