Apache Hamilton 数据管道实战详解

目标：从零跑通一个真实可复现的 Apache Hamilton 数据管道——编写函数模块、用 Builder 构建 Driver、注入输入数据执行、核对节点清单与计算结果。全文在 Ubuntu 22.04（Python 3.10.12、apache-hamilton 1.90.0）环境实测通过，示例取自官方 quickstart，所有代码与回显均为真实运行结果。

一、环境信息

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

操作系统：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）
示例目录：/root（函数模块与入口脚本均放在该目录下执行）

验证 Python 与 Apache Hamilton 就绪状态：

python3 -V

输出：

Python 3.10.12

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

输出（返回的是元组）：

(1, 90, 0)

二、前置条件

Apache Hamilton 已安装并可导入（安装全流程见《Ubuntu 22.04 安装 Apache Hamilton 指南》，本文不重复安装步骤）。
pandas 可用（随 apache-hamilton 依赖自动安装，实测版本 2.3.3）。
一个工作目录用于存放函数模块与入口脚本，本文使用 /root，脚本在该目录下直接执行。

验证 pandas 与 hamilton 同时可用：

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

输出：

2.3.3

三、核心 API 概览

实战只围绕两大对象展开：定义数据流的「函数模块」，与负责组装执行的「Driver」。当前官方文档统一使用 builder API 构造 Driver（旧式 driver.Driver(config, module) 直接构造已不出现）。核心 API 清单如下：

API	所属	作用	本文实测用法
`driver.Builder()`	hamilton.driver	创建 Builder，链式配置后构建 Driver	`driver.Builder().with_config({}).with_modules(my_functions).with_adapters(base.PandasDataFrameResult()).build()`
`with_config({})`	Builder	传入配置字典，配合 @config.when 等装饰器切换节点实现	空配置 `{}`
`with_modules(module)`	Builder	指定从哪些 Python 模块收集函数定义	传入 my_functions 模块
`with_adapters(adapter)`	Builder	指定结果装配器，决定 execute 返回形态	`base.PandasDataFrameResult()`
`execute(final_vars, inputs=输入字典)`	Driver	执行 DAG，返回所请求输出的计算结果	传入 4 个输出列名与输入字典
`list_available_variables()`	Driver	列出 DAG 全部节点名与类型	循环打印节点清单

对上表的解读：final_vars 是想要的结果列表，Driver 只围绕这些输出的依赖子图计算；inputs 是外部输入数据字典，键名必须与函数参数名一致。挂载 PandasDataFrameResult 时 execute 返回 pandas DataFrame，不挂载 adapter 时返回 dict（键为节点名、值为节点输出）。

四、集成示例：官方 quickstart 数据流

示例来源为官方文档 Your First Dataflow 页面，业务含义是「营销花费与注册用户的指标管道」：对每周花费做 3 周滚动均值，再除以注册数得到获客成本，并演示标量节点（均值、标准差）与序列节点的混合编排。

4.1 函数模块

创建 /root/my_functions.py，共 6 个函数（全文收录，函数名即节点名、参数即上游依赖、类型注解即数据形态）：

import pandas as pd


def avg_3wk_spend(spend: pd.Series) -> pd.Series:
    """Rolling 3 week average of spend."""
    return spend.rolling(3).mean()


def acquisition_cost(avg_3wk_spend: pd.Series, signups: pd.Series) -> pd.Series:
    """The cost per signup in relation to a rolling average of spend."""
    return avg_3wk_spend / signups


def spend_mean(spend: pd.Series) -> float:
    """Shows function creating a scalar. In this case it computes the mean of the entire column."""
    return spend.mean()


def spend_zero_mean(spend: pd.Series, spend_mean: float) -> pd.Series:
    """Shows function that takes a scalar. In this case to zero mean spend."""
    return spend - spend_mean


def spend_std_dev(spend: pd.Series) -> float:
    """Function that computes the standard deviation of the spend column."""
    return spend.std()


def spend_zero_mean_unit_variance(
    spend_zero_mean: pd.Series, spend_std_dev: float
) -> pd.Series:
    """Function showing one way to make spend have zero mean and unit variance."""
    return spend_zero_mean / spend_std_dev

模块要点：avg_3wk_spend 依赖输入节点 spend；acquisition_cost 同时依赖 avg_3wk_spend 与输入节点 signups；spend_zero_mean 展示了「序列节点依赖标量节点」的写法（依赖 spend_mean 的 float 输出）；spend_zero_mean_unit_variance 再进一步依赖另一个标量 spend_std_dev。整个模块没有任何框架侵入代码，就是普通的 pandas 函数。

4.2 Driver 入口脚本

创建 /root/my_script.py（全文收录），构造一个 6 周粒度的时间序列输入，构建 Driver 后先打印节点清单再执行：

import logging
import sys

import pandas as pd

# We add this to speed up running things if you have a lot in your python environment.
from hamilton import registry

registry.disable_autoload()

from hamilton import base, driver

import my_functions  # we import the module here!

logger = logging.getLogger(__name__)
logging.basicConfig(stream=sys.stdout)

if __name__ == "__main__":
    # Instantiate a common spine for your pipeline
    index = pd.date_range("2022-01-01", periods=6, freq="w")
    initial_columns = {
        "signups": pd.Series([1, 10, 50, 100, 200, 400], index=index),
        "spend": pd.Series([10, 10, 20, 40, 40, 50], index=index),
    }
    dr = (
        driver.Builder()
        .with_config({})  # we don't have any configuration or invariant data for this example.
        .with_modules(my_functions)  # we need to tell hamilton where to load function definitions from
        .with_adapters(base.PandasDataFrameResult())  # we want a pandas dataframe as output
        .build()
    )

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

    output_columns = [
        "spend",
        "signups",
        "avg_3wk_spend",
        "acquisition_cost",
    ]
    df = dr.execute(output_columns, inputs=initial_columns)
    print("=== execute result ===")
    print(df)

脚本要点：registry.disable_autoload() 关闭扩展自动加载以加快启动（关闭后本例仍能正常运行，因为 quickstart 不依赖 DataFrame 类型注册；若要用 @extract_columns 等类型感知装饰器，需手动加载 pandas 后端扩展，见注意事项）；initial_columns 的两个键 signups 与 spend 正是函数模块里的参数名。

4.3 运行

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

输出（完整，一次跑通）：

/root/my_script.py:20: FutureWarning: 'w' is deprecated and will be removed in a future version, please use 'W' instead.
  index = pd.date_range("2022-01-01", periods=6, freq="w")
=== DAG nodes: dr.list_available_variables() ===
  acquisition_cost  (type: <class 'pandas.core.series.Series'>)
  avg_3wk_spend  (type: <class 'pandas.core.series.Series'>)
  spend_mean  (type: <class 'float'>)
  spend_std_dev  (type: <class 'float'>)
  spend_zero_mean  (type: <class 'pandas.core.series.Series'>)
  spend_zero_mean_unit_variance  (type: <class 'pandas.core.series.Series'>)
  signups  (type: <class 'pandas.core.series.Series'>)
  spend  (type: <class 'pandas.core.series.Series'>)
=== execute result ===
            spend  signups  avg_3wk_spend  acquisition_cost
2022-01-02     10        1            NaN               NaN
2022-01-09     10       10            NaN               NaN
2022-01-16     20       50      13.333333          0.266667
2022-01-23     40      100      23.333333          0.233333
2022-01-30     40      200      33.333333          0.166667
2022-02-06     50      400      43.333333          0.108333

⚠️ 首行的 FutureWarning 来自 pandas 2.3.3 对小写 freq="w" 的弃用提示（官方示例原样使用小写）。写新代码应使用大写 "W"；本文如实保留官方原样写法与对应警告，便于读者对号入座。

五、执行结果验证

跑通只是第一步，这一节做三层核对：节点清单、数值、返回形态。

5.1 节点清单核对

回显中 list_available_variables() 列出 8 个节点，逐一对照来源：

节点名	类型	来源
acquisition_cost	pd.Series	函数 `acquisition_cost`
avg_3wk_spend	pd.Series	函数 `avg_3wk_spend`
spend_mean	float	函数 `spend_mean`
spend_std_dev	float	函数 `spend_std_dev`
spend_zero_mean	pd.Series	函数 `spend_zero_mean`
spend_zero_mean_unit_variance	pd.Series	函数 `spend_zero_mean_unit_variance`
signups	pd.Series	外部输入（initial_columns 注入）
spend	pd.Series	外部输入（initial_columns 注入）

对上表的解读：6 个函数节点与 my_functions.py 的 6 个函数一一对应，2 个输入节点与 inputs 字典的两个键一一对应——函数即节点、参数即依赖在节点清单上得到了直接验证。

5.2 数值人工核对

取回显结果表逐格验算：

avg_3wk_spend 前两行为 NaN：滚动窗口为 3 期，前两期数据不足，符合 rolling(3).mean() 的语义；第 3 行 (10+10+20)/3 = 13.333333，与回显一致。
acquisition_cost 第 3 行 = 13.333333/50 = 0.266667，与回显一致；第 4 行 23.333333/100 = 0.233333、第 5 行 33.333333/200 = 0.166667、第 6 行 43.333333/400 = 0.108333，全部吻合。
结果表只有请求的 4 列（spend、signups、avg_3wk_spend、acquisition_cost），spend_mean 等其余函数节点虽然也在 DAG 中，但因未被请求而未计算输出——验证了 Driver 只计算所请求输出的依赖子图。

💡 技术拓展：官方 quickstart 页面展示的示例输出中 acquisition_cost 列是 10.000、1.000、0.400 一类的数值，与官方自己贴出的代码（avg_3wk_spend / signups）算出的结果不符——那是官方页面残留的旧版代码产物。以本文真实运行为准即可，遇到官方文档示例输出对不上时，优先相信本地实测。

5.3 返回形态确认

本例通过 with_adapters(base.PandasDataFrameResult()) 挂载了结果装配器，回显的 execute result 是带日期索引的 pandas DataFrame，列顺序与请求顺序一致。若不挂载 adapter，execute() 直接返回 dict（键为节点名、值为节点输出），适合逐节点取用或对接自定义存储的场合；两种形态如何选择取决于下游消费方式——表格交付用 adapter，逐节点程序化消费用 dict。

六、注意事项

实测梳理出以下注意点，逐条落实可以避开绝大多数运行期问题：

freq 参数用大写：pd.date_range 的 freq="w" 在 pandas 2.3.3 触发 FutureWarning（回显见 4.3 节），新代码统一写 freq="W"。
inputs 键名必须与函数参数名严格一致：Driver 在执行前会校验输入完备性，缺失时会直接报错并列出缺失项及受影响节点，不会带病运行；输入键写成别名将触发同样的校验失败。
disable_autoload 与类型感知功能：registry.disable_autoload() 后 DataFrame 类型注册表为空，使用 @extract_columns 等装饰器需要手动 from hamilton.plugins import pandas_extensions，且实测需先 pip install packaging 兜底（上游依赖声明缺失），完整排障见《Apache Hamilton 装饰器与数据校验详解》。
只算依赖子图：请求越少计算越少，生产上按需声明 final_vars，避免把整个 DAG 全量算一遍。
版本行为差异：hamilton.__version__ 返回元组 (1, 90, 0) 而非字符串；需要字符串时用 importlib.metadata.version('apache-hamilton')（回显见第一节）。
官方示例输出可能陈旧：官方页面示例输出与现行代码的真实运行结果存在不一致（见 5.2 节技术拓展），落地时一律以本地实测为准。

七、参考链接

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