apache/hamilton · 上手攻略

• 仓库：apache/hamilton
• 链接：https://github.com/apache/hamilton
• 分类：数据流 DAG 框架 / Python 编排 / Feature Engineering
• 作者：spark
• 更新：2026-07-20

是什么

Apache Hamilton（incubating）是一个轻量的 Python 库，用"Python 函数即节点"的方式构建数据转换 DAG。它的核心思路是：你写普通的 Python 函数，函数参数的名字 = 它依赖的上游节点名，Hamilton 自动把这些函数拼成 DAG、并按依赖顺序执行。

举例：

# my_dag.py
import pandas as pd

def signups() -> pd.DataFrame:
    return load_signups()

def avg_age(signups: pd.DataFrame) -> float:
    return signups["age"].mean()

def report_text(avg_age: float) -> str:
    return f"平均年龄 {avg_age:.1f} 岁"

Hamilton 看到 avg_age(signups=...)、report_text(avg_age=...)，自动构建出 signups → avg_age → report_text 的有向无环图，调用时只问 driver 要 report_text，它会从最上游开始递归求值。这种"参数名 = 依赖"的约定是 Hamilton 区别于 Airflow / Prefect / Dagster 的关键——你完全不用画图、不用写 YAML、不用写 operator。

项目出身：Stitch Fix 内部工具（2019 起），后来独立成 DAGWorks 公司，2024 年进入 Apache 孵化器。生产用户列表里能看到 Stitch Fix、IBM、Adobe、Opendoor、UK Government Digital Services、Lexis Nexis、Federal Reserve Board、Joby Aviation 等。

License：Apache-2.0。Python 3.8+。

解决什么问题

数据团队写 ETL / ML 特征工程 / LLM 应用时，几乎都会撞到同一种代码腐烂：

• 一个 notebook 写到底，函数之间靠"上面 cell 的变量"隐式传递，挪到生产模块时一脸懵。
• 上百个特征散在多个文件里，新人无法回答"A 特征依赖哪些上游"。
• 改了某个特征，没有可视化告诉你它会影响哪些下游指标。
• 想把同一份 DAG 跑到 notebook 本地、Airflow 任务、FastAPI 服务三处，要写三套胶水。

Hamilton 把这些问题一次性解决：

• DAG 是显式的、可视化的：dr.visualize_execution([...]) / hamilton ui 直接出图。
• 模块化拼接：把 DAG 拆成多个 .py 文件，driver 同时加载，靠函数名跨文件依赖。
• 自文档化：函数名 + docstring + type hint 就是节点元数据，UI 自动生成 catalog。
• 一处定义，多处跑：Hamilton driver 不绑任何 runtime，notebook、Airflow、Dagster、FastAPI、Ray、Metaflow 都能复用同一份 DAG 定义。
• 测试友好：每个节点 = 一个普通函数，单元测试不需要启 DAG。

简言之，Hamilton 是 Python 世界的 dbt：dbt 用 SQL 定义资产转换，Hamilton 用 Python 函数做同样的事。

快速安装

最小化

pip install apache-hamilton

带可视化（需要系统装 Graphviz）

# macOS
brew install graphviz

pip install "apache-hamilton[visualization]"

带 UI + SDK 适配器（团队协作 + lineage tracking）

pip install "apache-hamilton[ui,sdk]"

跑起来后访问 hamilton.apache.org/hamilton-ui/ui/，首次进入建账号 + 新建 project，project_id=1 即可。

在线试玩

不想本地装：访问 https://www.tryhamilton.dev/，浏览器内运行官方 hello-world。

核心用法

1) 定义 DAG

my_dag.py：

import pandas as pd

def signups() -> pd.DataFrame:
    return pd.read_csv("signups.csv")

def spend(signups: pd.DataFrame) -> pd.Series:
    return signups.groupby("channel")["spend"].sum()

def avg_spend_per_channel(spend: pd.Series) -> pd.DataFrame:
    return spend.reset_index(name="total_spend")

def top_channel(avg_spend_per_channel: pd.DataFrame) -> str:
    row = avg_spend_per_channel.sort_values("total_spend", ascending=False).iloc[0]
    return row["channel"]

注意每个函数的参数名 = 它依赖的节点名。

2) 跑 DAG

from hamilton import driver

dr = driver.Builder().with_modules(my_dag).build()
result = dr.execute(["top_channel", "avg_spend_per_channel"])
print(result)

Hamilton 会自动按依赖顺序执行，最后只把 top_channel 和 avg_spend_per_channel 的结果返回。

3) 可视化

dr.visualize_execution(["top_channel"], ..., render_kwargs={"format": "png"})

依赖图直接渲染到 notebook 或文件。

4) 跟 UI 联动（团队 catalog + execution tracking）

from hamilton import driver
from hamilton_sdk.adapters import HamiltonTracker

tracker = HamiltonTracker(
    username="my_username",
    project_id=1,
    dag_name="hello_world",
)

dr = (
    driver.Builder()
    .with_modules(my_dag)
    .with_adapters(tracker)  # 注册 tracker
    .build()
)

dr.execute(["top_channel"])

打开 hamilton ui（或者远程 Hamilton UI），新建的 DAG 会出现在 catalog 里，每次 execute 都会写入 lineage + 结果摘要。

5) 函数修饰器（核心杀手锏）

Hamilton 提供一组 @config.* 和 @check_* 修饰器，让 DAG 在不同环境表现不同：

from hamilton.function_modifiers import config

@config.when(environment="dev")
def sample_data(raw: pd.DataFrame) -> pd.DataFrame:
    return raw.head(100)

@config.when(environment="prod")
def sample_data(raw: pd.DataFrame) -> pd.DataFrame:
    return raw

调用时：

dr.execute(["feature_x"], inputs={"environment": "prod"})

无需 if/else 即可切换行为。其他常用修饰器：

• @check_output：验证节点输出（dataframe shape、schema、范围）。
• @parameterize：用一份函数定义派生多个节点。
• @extract_fields：把 dataclass 自动展开成多个下游节点。
• @load_from：从外部系统（cache、对象存储）注入上游节点。

6) 多模块拼接

# features/user.py、features/order.py、features/text.py ...
dr = driver.Builder().with_modules(features.user, features.order, features.text).build()

跨文件依赖靠函数名匹配，零胶水。

7) 跟其它执行器协作

Hamilton 自己不做 orchestration，但提供官方适配器：

• Airflow：把 Hamilton DAG 包成 Airflow task。
• FastAPI：在 request handler 里直接 dr.execute([...])，单次请求的 lineage 也能被 UI 抓到。
• Ray / Dask：远程执行节点，适合重特征工程。
• Metaflow / Dagster：作为 step 内容。

典型适用场景

• ML 特征工程：几百个特征散在多个 notebook，整合成 Hamilton DAG 后，train.py / serve.py 复用同一份定义。
• LLM 应用预处理：把 prompt 构造、retrieval、rerank、token 计数、prompt 模板等都定义成 Hamilton 节点，用 dr.execute(["final_prompt"]) 一步拿到结果，配合 Hamilton UI 看 prompt 漂移。
• RAG 数据管线：文档加载 → 切分 → embedding → 索引，每步是个节点，schema check 一行修饰器加上。
• BI 数据资产：dbt 管 SQL，Hamilton 管 Python（pandas / polars / Ibis），组合用。
• 跨团队数据契约：定义好模块化的 features/ 包，下游消费者只 from features.user import * 然后 driver.execute(...)。

坑与注意

• 函数参数名即契约：重命名函数参数会破坏依赖关系，IDE rename refactor 要小心；团队要约定参数命名规范。
• 循环依赖会报错：Hamilton 拒绝构建有环 DAG，这是 feature 不是 bug；如果业务需要循环，README 明确推荐用 Burr（Apache 旗下姊妹项目）。
• Graphviz 必须系统装：光 pip install graphviz 不够，macOS/Linux 需要 brew install graphviz 或 apt-get install graphviz，否则可视化失败。
• 不是 orchestrator：Hamilton 不调度、不重试、不报警，需要 Airflow/Prefect/Dagster 之类做 orchestration。README 原话："Hamilton is not an orchestrator, nor a feature store"。
• Apache 孵化器状态：项目还在 Apache Incubator，没有顶级项目 (TLPs) 背书，但代码本身从 2019 年就在生产用。
• 类型注解不是强制的：但不写 type hint 会在 UI catalog 里少很多 metadata，建议团队 lint 强制。
• 大节点不适合 Hamilton：单个函数如果耗 30 分钟且要 checkpoint，建议用 Airflow + Hamilton 组合而不是 Hamilton 单干。

与同类对比

Hamilton 在"Python 数据转换"这个细分上几乎没有对手：dagster / prefect 偏 orchestration 复杂度过高，dbt 不支持 Python，Metaflow 偏 ML 训练且绑定 AWS 较深。如果团队的数据资产大部分是 Python（pandas / polars），Hamilton 是最贴切的选项。

一句话推荐

任何"几百个 Python 函数拼数据、想看依赖图、想一处定义到处跑"的场景，Hamilton 都比手撸 + Airflow 更值得——尤其当你的数据资产已经在 notebook 里散落多年。

不确定处

• apache-hamilton[ui,sdk] 的最新 extras 列表来自 README，未对照 pyproject.toml 逐项验证；安装前建议看 https://hamilton.apache.org/getting-started/installation/。
• Hamilton UI 的 "username + project_id=1" 是初次启动流程，文档可能因版本更新调整；启动后按 UI 提示为准。
• 用户列表中的 Joby Aviation、UK Gov 等案例，团队规模和具体使用方式 README 未给出，无法判断"小团队能否复刻"。
• Apache 孵化状态截至 README 抓取时为 incubating，未来可能晋升 TLP，引用方式无变化但社区治理会逐步规范化。