Apache Hamilton 装饰器与数据校验详解

目标：掌握 Apache Hamilton 三个高频装饰器的实战用法——@parameterize（一个函数展开为多个节点）、@extract_columns（把 DataFrame 列拆成独立节点）、@check_output（节点输出数据质量校验），并通过综合运行实录看清三者叠加后的 DAG 形态。全文在 Ubuntu 22.04（Python 3.10.12、apache-hamilton 1.90.0）环境实测通过，每个装饰器都给出「说明、实测代码、实测输出、要点」的完整闭环，五处真实踩坑全部附原始报错回显与修复方法。

一、环境信息

本文全部代码与命令回显均在以下环境实测取得，复现时保持相近版本即可获得一致结果。

操作系统：Ubuntu 22.04（x86_64）
Python：3.10.12（虚拟环境 /opt/venv-hamilton，pip 26.2.1）
Apache Hamilton：apache-hamilton 1.90.0（依赖 numpy 2.2.6、pandas 2.3.3、typing_extensions 4.16.0、typing_inspect 0.9.0；另装 packaging 26.3 兜底，见第二节）
示例目录：/root（装饰器模块与入口脚本均放在该目录下执行）

验证环境就绪：

/opt/venv-hamilton/bin/python -c "from importlib.metadata import version; print(version('apache-hamilton'))"

输出：

1.90.0

/opt/venv-hamilton/bin/python -c "import numpy; print(numpy.__version__)"

输出：

2.2.6

二、前置条件

Apache Hamilton 已安装（安装全流程见《Ubuntu 22.04 安装 Apache Hamilton 指南》）。
packaging 兜底安装（本篇实测确认的硬前置）：1.90.0 的 pandas 插件模块需要 packaging 库，但 apache-hamilton 与 pandas 的依赖声明均未包含它，属上游打包问题。凡用到 DataFrame 类型感知的装饰器（本篇的 @extract_columns 即典型），必须先补装，否则导入阶段直接报错（完整报错见 4.2 节）。
一个工作目录存放示例脚本，本文使用 /root。

packaging 的实测安装回显：

/opt/venv-hamilton/bin/pip install packaging

输出（实测节选）：

Downloading packaging-26.3-py3-none-any.whl (129 kB)
Installing collected packages: packaging
Successfully installed packaging-26.3

影响面实测结论：不装 packaging 且保持默认扩展自动加载时，基础数据流仍能跑（自动加载器静默跳过加载失败的插件），但 @extract_columns 一类依赖 DataFrame 类型注册的功能会失败；装上 packaging 后，无论是否关闭自动加载，本篇全部示例正常。

三、@parameterize：一个函数展开为多个节点

本节按「机制说明、踩坑实录、实测代码、节点验证、要点小结」的闭环展开，先讲清展开机制，再还原第一手报错与修复过程。

3.1 机制说明

@parameterize 把同一个函数按不同参数取值展开成多个独立节点：装饰器字典的键是展开后的节点名，值是「参数名到取值」的映射。适合「同一套逻辑、多种口径」的场景——比如求均值时分别要绝对值口径与带符号口径，无需写两个函数。

3.2 踩坑实录：取值必须用 value() 或 source() 包裹

第一版直接写裸值（如 {"abs": True}），运行即报错：

hamilton.function_modifiers.base.InvalidDecoratorException: @parameterize must specify a dependency type -- either source() or value().The following are not allowed: [True, False].

原因：@parameterize 的参数取值不允许裸字面量，必须显式声明类型——固定常量用 value() 包裹，引用其他节点的输出用 source() 包裹（报错原文给出的两种合法方式）。修复：{"abs": value(True)}，并从 hamilton.function_modifiers 导入 value。

3.3 实测代码

my_decorators.py 中对应的函数定义：

from hamilton.function_modifiers import parameterize, value


@parameterize(
    spend_mean_abs={"abs": value(True)},
    spend_mean_signed={"abs": value(False)},
)
def spend_mean_variant(spend: pd.Series, abs: bool) -> float:
    """Mean of spend, optionally taking the absolute value first."""
    if abs:
        return spend.abs().mean()
    return spend.mean()

3.4 节点验证

运行后节点清单中出现了两个展开节点（清单片段）：

  spend_mean_abs  (type: <class 'float'>)
  spend_mean_signed  (type: <class 'float'>)

实测输入 spend = pd.Series([10, 20, 30, 40]) 时两节点均输出 25.0——本例数据全为正数，绝对值口径与带符号口径的均值相同，属于正常结果；两者作为独立节点已可在 execute 的请求列表中单独选用。

3.5 要点小结

要点	说明
取值包裹	常量用 `value(常量)`，引用上游节点用 `source(节点名)`，禁止裸字面量
展开命名	字典键即节点名，语义化命名便于下游引用
节点独立性	展开后的每个节点可独立请求、独立校验

四、@extract_columns：把 DataFrame 列拆成独立节点

本节同样按完整闭环展开；这个装饰器的排障价值最高——两处真实报错（类型注册为空、packaging 缺失）都在它身上暴露，逐层定位过程见 4.2 节。

4.1 机制说明

@extract_columns("列名一", "列名二") 修饰一个返回 DataFrame 的函数，把指定列拆成同名独立节点：下游函数可以直接以列名为参数名引用某一列，而不必手工从 DataFrame 里取列。原函数本身仍是节点，两种粒度并存。

4.2 踩坑实录：Registered types are {} 与 packaging 缺失

在关闭扩展自动加载（registry.disable_autoload()）的环境里，被 @extract_columns 修饰的函数在导入阶段即报错：

NotImplementedError: Cannot get column type for [<class 'pandas.core.frame.DataFrame'>]. Registered types are {}

随后还有一条：

hamilton.function_modifiers.base.InvalidDecoratorException: Error <function raw_df at 0x74e7a3cd2ef0> does not output a type we know about. Is it a dataframe type we support?

排障定位分两层。第一层：disable_autoload() 关闭了扩展自动加载，DataFrame 类型没有注册，需手动加载 pandas 后端扩展：

from hamilton.plugins import pandas_extensions

第二层：手动加载后又暴露出新的报错（traceback 原文）：

File "/opt/venv-hamilton/lib/python3.10/site-packages/hamilton/plugins/pandas_extensions.py", line 27, in <module>
    from packaging.version import Version
ModuleNotFoundError: No module named 'packaging'

即第二节所述的上游依赖声明缺失，pip install packaging 后两层问题同时消除。另有一条实测发现值得记录：不关闭自动加载且未装 packaging 时，@extract_columns 同样报 Registered types are {}——说明该情况下自动加载器也是静默失败，pandas 后端从未注册成功，packaging 兜底安装是必选项。

4.3 实测代码

my_decorators.py 中对应的函数定义：

from hamilton.function_modifiers import extract_columns


@extract_columns("date", "value")
def raw_df() -> pd.DataFrame:
    """A two-column dataframe that @extract_columns explodes into separate DAG nodes."""
    return pd.DataFrame(
        {
            "date": pd.date_range("2022-01-01", periods=4, freq="D"),
            "value": [1.0, 2.0, 3.0, 4.0],
        }
    )

4.4 节点验证

运行后节点清单中出现一拆三的效果（清单片段）：

  raw_df  (type: <class 'pandas.core.frame.DataFrame'>)
  date  (type: <class 'pandas.core.series.Series'>)
  value  (type: <class 'pandas.core.series.Series'>)

对回显的解读：raw_df 原函数保留为 DataFrame 节点，date 与 value 两列各自成为 Series 节点；第六节综合实录的执行结果里，date 与 value 两节点的输出与 raw_df 中对应列逐值一致。

4.5 要点小结

要点	说明
拆列即拆节点	装饰器参数里的每个列名都成为独立节点，可单独请求
原节点保留	返回 DataFrame 的原函数仍在 DAG 中，两种粒度并存
环境前置	关闭自动加载时手动加载 pandas_extensions，且必须 `pip install packaging` 兜底

五、@check_output：节点输出数据质量校验

本节围绕「声明式校验」展开：先给机制，再还原两个参数层面的真实踩坑（dtype 参数不支持、data_type 写成序列形态），最后看框架自动生成的校验节点行为。

5.1 机制说明

@check_output 为节点输出声明校验规则，框架自动生成对应的校验节点并接管输出：校验通过时数据正常流向下游，失败时默认产生告警而不中断流程（告警模式可配置）。可校验维度包括数据类型、取值范围等，均以声明式参数给出。

5.2 踩坑实录一：Series 输出不支持 dtype 校验参数

第一版写成 @check_output(data_type=pd.Series, dtype="float64", range=(0, 100))，运行报错：

ValueError: No registered subclass of BaseDefaultValidator is available for arg: dtype and type <class 'pandas.core.series.Series'>. This either means (a) this arg-type contribution isn't supported or (b) this has not been added yet (but should be). In the case of (b), we welcome contributions. Get started at github.com/dagworks-inc/hamilton.

随后还有一条：

ValueError: Could not resolve validators for @check_output for function [value_pct]. Please check that `target_` is set correctly if you're using that argument.

原因：对 pd.Series 类型的输出，当前版本没有注册 dtype 维度的校验器。修复：删掉 dtype="float64" 参数，保留 data_type 与 range。

5.3 踩坑实录二：data_type 校验的是元素 dtype 而非 Series 本身

删掉 dtype 参数后能运行，但出现校验告警：

[value_pct:data_type_validator] validator failed. Message was: Requires subclass of datatype: <class 'pandas.core.series.Series'>. Got datatype: <class 'numpy.float64'>. This is not a match.. Diagnostic information is: {'required_dtype': <class 'pandas.core.series.Series'>, 'actual_dtype': <class 'numpy.float64'>}.

原因：对 Series 输出，data_type 校验的是元素的 dtype（逐元素比对 numpy.float64 一类），而不是 Series 序列本身；写 data_type=pd.Series 等于要求元素也是 Series，必然失配。修复：按元素类型写 data_type=np.float64，重跑后告警消失。

5.4 实测代码

my_decorators.py 中对应的函数定义：

import numpy as np

from hamilton.function_modifiers import check_output


@check_output(
    data_type=np.float64,
    range=(0, 100),
)
def value_pct(value: pd.Series) -> pd.Series:
    """value as a percent of its own max, kept within [0, 100]."""
    return value / value.max() * 100.0

5.5 节点行为验证

运行后节点清单中出现了校验器接管输出的三个痕迹（清单片段）：

  value_pct_data_type_validator  (type: <class 'hamilton.data_quality.base.ValidationResult'>)
  value_pct_range_validator  (type: <class 'hamilton.data_quality.base.ValidationResult'>)
  value_pct  (type: <class 'pandas.core.series.Series'>)
  value_pct_raw  (type: <class 'pandas.core.series.Series'>)

对回显的解读：两个 ValidationResult 类型的校验节点是框架按 data_type 与 range 两条规则自动追加的；原函数被改名为 value_pct_raw（原始输出），校验通过后的结果以 value_pct 节点提供给下游——下游依赖写 value_pct 即拿到校验后的数据。

5.6 要点小结

要点	说明
Series 输出的 data_type 写元素类型	写 `np.float64` 一类元素 dtype，不要写 `pd.Series`
不支持的参数会直接报错	dtype 参数对 Series 输出无注册校验器，删除即可
校验节点自动化	每条规则生成一个 ValidationResult 节点，原函数改名带 _raw 后缀
默认告警不中断	校验失败默认告警，流程可继续，适合先观察再收紧策略

六、三装饰器综合运行实录

把三个装饰器放进同一个模块，一次运行看清叠加效果。以下为 /root/my_decorators.py 全文：

import pandas as pd
import numpy as np

from hamilton.function_modifiers import check_output, extract_columns, parameterize, value


@parameterize(
    spend_mean_abs={"abs": value(True)},
    spend_mean_signed={"abs": value(False)},
)
def spend_mean_variant(spend: pd.Series, abs: bool) -> float:
    """Mean of spend, optionally taking the absolute value first."""
    if abs:
        return spend.abs().mean()
    return spend.mean()


@extract_columns("date", "value")
def raw_df() -> pd.DataFrame:
    """A two-column dataframe that @extract_columns explodes into separate DAG nodes."""
    return pd.DataFrame(
        {
            "date": pd.date_range("2022-01-01", periods=4, freq="D"),
            "value": [1.0, 2.0, 3.0, 4.0],
        }
    )


@check_output(
    data_type=np.float64,
    range=(0, 100),
)
def value_pct(value: pd.Series) -> pd.Series:
    """value as a percent of its own max, kept within [0, 100]."""
    return value / value.max() * 100.0

配套入口脚本 /root/my_decorators_script.py 全文（本次不挂结果装配器，execute 直接返回 dict）：

import pandas as pd

from hamilton import registry

registry.disable_autoload()
# autoload is off, so we load the pandas backend extension ourselves
from hamilton.plugins import pandas_extensions  # noqa: E402,F401

from hamilton import driver  # noqa: E402

import my_decorators  # noqa: E402

if __name__ == "__main__":
    dr = (
        driver.Builder()
        .with_config({})
        .with_modules(my_decorators)
        .build()  # no result adapter -> execute() returns a dict of {node: value}
    )

    print("=== DAG nodes: dr.list_available_variables() ===")
    for var in dr.list_available_variables():
        print(f"  {var.name}  (type: {var.type})")

    result = dr.execute(
        ["spend_mean_abs", "spend_mean_signed", "date", "value", "value_pct"],
        inputs={"spend": pd.Series([10, 20, 30, 40])},
    )
    print("=== execute result ===")
    for k, v in result.items():
        print(f"  {k}: {v!r}")

运行：

cd /root && /opt/venv-hamilton/bin/python my_decorators_script.py

输出（完整，最终成功版）：

=== DAG nodes: dr.list_available_variables() ===
  raw_df  (type: <class 'pandas.core.frame.DataFrame'>)
  date  (type: <class 'pandas.core.series.Series'>)
  value  (type: <class 'pandas.core.series.Series'>)
  spend_mean_abs  (type: <class 'float'>)
  spend_mean_signed  (type: <class 'float'>)
  value_pct_data_type_validator  (type: <class 'hamilton.data_quality.base.ValidationResult'>)
  value_pct_range_validator  (type: <class 'hamilton.data_quality.base.ValidationResult'>)
  value_pct  (type: <class 'pandas.core.series.Series'>)
  value_pct_raw  (type: <class 'pandas.core.series.Series'>)
  spend  (type: <class 'pandas.core.series.Series'>)
=== execute result ===
  spend_mean_abs: np.float64(25.0)
  spend_mean_signed: np.float64(25.0)
  date: 0   2022-01-01
1   2022-01-02
2   2022-01-03
3   2022-01-04
Name: date, dtype: datetime64[ns]
  value: 0    1.0
1    2.0
2    3.0
3    4.0
Name: value, dtype: float64
  value_pct: 0     25.0
1     50.0
2     75.0
3    100.0
Name: value, dtype: float64

对运行结果逐项解读：

节点	类型	行为来源与数值核对
spend_mean_abs、spend_mean_signed	float	@parameterize 展开；输入全为正数，两口径均值同为 25.0（(10+20+30+40)/4）
raw_df、date、value	DataFrame、Series、Series	@extract_columns 拆列；date 为 4 天日期序列，value 为 1.0 至 4.0
value_pct_raw、value_pct_data_type_validator、value_pct_range_validator、value_pct	Series、两个 ValidationResult、Series	@check_output 接管；value_pct 逐值为自身最大值的百分比：1/4=25.0、2/4=50.0、3/4=75.0、4/4=100.0，全部落在 (0, 100) 区间内，两条校验均通过
spend	Series	外部输入节点

补充说明：本次 execute 返回 dict（未挂 adapter），键为节点名、值为节点输出；校验节点未出现在请求列表中，但校验逻辑已在 value_pct 输出前执行。三个装饰器的行为在节点清单中全部可追溯，说明装饰器是声明式的节点级增强，不改变「函数即节点」的基本模型。

七、排障速查表

报错关键词	触发场景	根因	解决
must specify a dependency type -- either source() or value()	@parameterize 用裸字面量取值	装饰器要求显式声明取值类型	用 `value(常量)` 或 `source(节点名)` 包裹
Registered types are {}	使用 DataFrame 类型感知装饰器	pandas 后端未注册（关闭自动加载，或自动加载静默失败）	手动 `from hamilton.plugins import pandas_extensions`
No module named 'packaging'	加载 pandas_extensions 时	上游未声明 packaging 依赖	`pip install packaging`
No registered subclass of BaseDefaultValidator is available for arg: dtype	@check_output 对 Series 输出写 dtype 参数	该类型不支持 dtype 校验参数	删除 dtype 参数，保留 data_type 与 range
validator failed. Message was: Requires subclass of datatype	data_type 写成 pd.Series	data_type 对 Series 校验的是元素 dtype	改写为元素类型，如 `np.float64`

对上表的解读：五条全部来自本篇实测的第一手报错，按「报错关键词对号入座」即可直接处置；其中 packaging 缺失是 1.90.0 的上游问题，建议在项目依赖清单里显式加入 packaging，一劳永逸。

八、参考链接

Apache Hamilton 官方文档：https://hamilton.apache.org/
官方快速上手 Your First Dataflow：https://hamilton.apache.org/get-started/your-first-dataflow/
PyPI 项目页（apache-hamilton）：https://pypi.org/project/apache-hamilton/
项目主页（pip show Home-page 实测值）：https://www.tryhamilton.dev/
GitHub 仓库（Apache 孵化）：https://github.com/apache/hamilton