Proposal: Hamilton Provider for Apache Airflow

[Dev-iL]

Motivation

Hamilton and Airflow are complementary tools that operate at different levels:

• Airflow orchestrates tasks (Python, SQL, Bash, etc.) and ensures they complete.
• Hamilton is a data transformation micro-framework where Python functions define DAG nodes.

Today, using Hamilton inside Airflow requires manual boilerplate in every task: importing modules, creating the Driver, executing it, and handling results. The existing examples/airflow directory demonstrates this pattern. A provider package would eliminate this boilerplate and give Hamilton first-class status in the Airflow ecosystem.

This proposal targets Airflow 3 only. Airflow providers are moving away from Airflow 2 support, and an Airflow 3-only provider avoids compatibility shims, direct metadata database assumptions, and old XCom guidance that no longer reflects configured XCom backends.

Package Identity

	Community provider (v1)	Official Apache provider (future)
PyPI name	airflow-provider-hamilton	apache-airflow-providers-apache-hamilton
Python package	hamilton_provider	airflow.providers.apache.hamilton
Lives in	Hamilton monorepo (submodule)	Airflow monorepo

Starting as a community provider is lower friction -- no Airflow PR process, release on Hamilton's own cadence. Can migrate to an official Apache provider later if there is demand.

Package Structure

hamilton_provider/
    __init__.py
    get_provider_info.py
    lineage.py                     # OpenLineage helpers + Hamilton metadata collector
    operators/
        __init__.py
        hamilton.py                # HamiltonOperator + shared execution helpers
    decorators/
        __init__.py
        hamilton.py                # _HamiltonDecoratedOperator + hamilton_task()

No types.py dataclass is proposed for v1. The provider should expose Airflow-native operator/decorator arguments and normal TaskFlow function arguments instead of requiring a custom HamiltonTask return object.

Entry Point Registration

pyproject.toml:

[project.entry-points."apache_airflow_provider"]
provider_info = "hamilton_provider.get_provider_info:get_provider_info"

get_provider_info.py:

def get_provider_info():
    return {
        "package-name": "airflow-provider-hamilton",
        "name": "Apache Hamilton",
        "description": "Apache Hamilton integration for Airflow",
        "operators": [
            {
                "integration-name": "Apache Hamilton",
                "python-modules": ["hamilton_provider.operators.hamilton"],
            }
        ],
        "task-decorators": [
            {
                "name": "hamilton",
                "class-name": "hamilton_provider.decorators.hamilton.hamilton_task",
            }
        ],
    }

---

API Surface

API 1: HamiltonOperator

A declarative operator for Hamilton modules that handle their own data loading or materialization. inputs, config, and overrides are templateable dictionaries. Dict-shaped Jinja templates require render_template_as_native_obj=True on the DAG; otherwise rendered values may arrive as strings and must either be valid JSON objects or fail with a clear AirflowException.

Usage

from airflow.sdk import DAG
from hamilton_provider.operators.hamilton import HamiltonOperator

with DAG(..., render_template_as_native_obj=True):
    extract = SomeUpstreamOperator(task_id="extract", ...)

    transform = HamiltonOperator(
        task_id="transform",
        modules=["my_project.prepare_data", "my_project.feature_logic"],
        config={"development_flag": True},
        inputs={"csv_path": "{{ ti.xcom_pull('extract') }}"},
        final_vars=["features_path", "row_count"],
    )

    extract >> transform

Implementation Sketch

class HamiltonOperator(BaseOperator):
    template_fields = ("inputs", "config", "overrides")
    template_fields_renderers = {
        "inputs": "json",
        "config": "json",
        "overrides": "json",
    }

    def __init__(
        self,
        *,
        modules: list[str],
        final_vars: list[str],
        config: dict[str, Any] | str | None = None,
        inputs: dict[str, Any] | str | None = None,
        overrides: dict[str, Any] | str | None = None,
        adapters: list[Any] | None = None,
        materializers: list[Any] | None = None,
        allow_module_overrides: bool = False,
        result_handler: Callable[[Any], Any] | None = None,
        **kwargs,
    ):
        super().__init__(**kwargs)
        self.modules = modules
        self.final_vars = final_vars
        self.config = config or {}
        self.inputs = inputs or {}
        self.overrides = overrides or {}
        self.adapters = adapters or []
        self.materializers = materializers or []
        self.allow_module_overrides = allow_module_overrides
        self.result_handler = result_handler
        self._lineage_collector = None

    def execute(self, context):
        result, collector = run_hamilton(
            modules=self.modules,
            config=coerce_json_object(self.config, "config"),
            inputs=coerce_json_object(self.inputs, "inputs"),
            overrides=coerce_json_object(self.overrides, "overrides"),
            final_vars=self.final_vars,
            adapters=self.adapters,
            materializers=self.materializers,
            allow_module_overrides=self.allow_module_overrides,
        )
        self._lineage_collector = collector

        if self.result_handler:
            return self.result_handler(result)
        return result

Key properties:

• template_fields = ("inputs", "config", "overrides") -- Airflow renders these before execute() runs.
• template_fields_renderers tells the Airflow UI these fields are JSON-like.
• coerce_json_object() accepts dictionaries as-is, parses JSON object strings, and raises AirflowException for non-object strings with guidance to enable render_template_as_native_obj=True.
• result_handler transforms output before Airflow handles XCom.
• All Hamilton imports are deferred to execution so DAG parsing stays fast.
• The operator does not query or write Airflow metadata database tables directly.

When to use

The operator is designed for static Airflow tasks where Hamilton modules are known at DAG parse time and runtime variation is mostly through config, inputs, and overrides. It works well for modules using Hamilton data loaders/savers, materializers, or Airflow XCom-backed inputs.

---

API 2: @task.hamilton

A TaskFlow decorator for cases needing Python input preparation, dynamic runtime values, or Builder-level control while still fitting normal Airflow TaskFlow semantics. The decorated function receives individual Airflow task arguments and returns a dictionary of additional Hamilton inputs, or None if no runtime inputs are needed.

Hamilton execution options live on the decorator/operator kwargs, not in a custom dataclass return value. This keeps the callable compatible with mapped tasks and with the way Airflow users expect TaskFlow functions to behave.

Usage

from airflow.sdk import dag, task

@dag(..., render_template_as_native_obj=True)
def my_pipeline():
    @task
    def extract():
        return "/data/raw/2024-01-01.csv"

    @task.hamilton(
        modules=["my_project.prepare_data", "my_project.feature_logic"],
        config={"development_flag": True},
        final_vars=["features_path"],
    )
    def transform(csv_path: str) -> dict[str, object]:
        import pandas as pd

        raw_df = pd.read_csv(csv_path)
        return {"raw_df": raw_df}

    transform(extract())

Implementation Sketch

class _HamiltonDecoratedOperator(DecoratedOperator, PythonOperator):
    custom_operator_name = "@task.hamilton"
    template_fields = (
        *PythonOperator.template_fields,
        "hamilton_config",
        "hamilton_inputs",
        "hamilton_overrides",
    )
    template_fields_renderers = {
        **PythonOperator.template_fields_renderers,
        "hamilton_config": "json",
        "hamilton_inputs": "json",
        "hamilton_overrides": "json",
    }

    def __init__(
        self,
        *,
        python_callable: Callable,
        modules: list[str],
        final_vars: list[str],
        config: dict[str, Any] | str | None = None,
        inputs: dict[str, Any] | str | None = None,
        overrides: dict[str, Any] | str | None = None,
        adapters: list[Any] | None = None,
        materializers: list[Any] | None = None,
        allow_module_overrides: bool = False,
        result_handler: Callable[[Any], Any] | None = None,
        op_args: Sequence | None = None,
        op_kwargs: dict | None = None,
        **kwargs,
    ) -> None:
        if inspect.iscoroutinefunction(python_callable):
            raise AirflowException("@task.hamilton does not support async callables in v1")

        kwargs_to_upstream = {
            "python_callable": python_callable,
            "op_args": op_args,
            "op_kwargs": op_kwargs,
        }
        super().__init__(
            kwargs_to_upstream=kwargs_to_upstream,
            python_callable=python_callable,
            op_args=op_args,
            op_kwargs=op_kwargs,
            **kwargs,
        )
        self.modules = modules
        self.final_vars = final_vars
        self.hamilton_config = config or {}
        self.hamilton_inputs = inputs or {}
        self.hamilton_overrides = overrides or {}
        self.adapters = adapters or []
        self.materializers = materializers or []
        self.allow_module_overrides = allow_module_overrides
        self.result_handler = result_handler
        self._lineage_collector = None

    def execute_callable(self) -> Any:
        runtime_inputs = super().execute_callable()
        if runtime_inputs is None:
            runtime_inputs = {}
        if not isinstance(runtime_inputs, dict):
            raise AirflowException("@task.hamilton callable must return dict[str, Any] or None")

        inputs = {
            **coerce_json_object(self.hamilton_inputs, "inputs"),
            **runtime_inputs,
        }
        result, collector = run_hamilton(
            modules=self.modules,
            config=coerce_json_object(self.hamilton_config, "config"),
            inputs=inputs,
            overrides=coerce_json_object(self.hamilton_overrides, "overrides"),
            final_vars=self.final_vars,
            adapters=self.adapters,
            materializers=self.materializers,
            allow_module_overrides=self.allow_module_overrides,
        )
        self._lineage_collector = collector

        if self.result_handler:
            return self.result_handler(result)
        return result

The decorator factory follows Airflow provider patterns:

def hamilton_task(
    python_callable: Callable | None = None,
    multiple_outputs: bool | None = None,
    **kwargs,
) -> TaskDecorator:
    return task_decorator_factory(
        python_callable=python_callable,
        multiple_outputs=multiple_outputs,
        decorated_operator_class=_HamiltonDecoratedOperator,
        **kwargs,
    )

Execution Model

The decorated operator should follow the same pattern as Airflow's provider decorators: inherit from DecoratedOperator and the concrete underlying operator, pass kwargs_to_upstream, and customize the point where the Python callable result is available. The implementation should be tested against Airflow 3 rather than justified by a hand-written MRO description.

Async/deferrable execution is out of scope for v1. Hamilton has async driver behavior, but mixing that with Airflow's async/deferrable task lifecycle needs a separate design. v1 should fail fast for coroutine callables instead of accidentally running them through the synchronous driver path.

---

Shared Hamilton Execution

Both APIs should use one internal helper so operator and decorator behavior does not drift:

def run_hamilton(
    *,
    modules: list[str],
    config: dict[str, Any],
    inputs: dict[str, Any],
    overrides: dict[str, Any],
    final_vars: list[str],
    adapters: list[Any],
    materializers: list[Any],
    allow_module_overrides: bool,
) -> tuple[dict[str, Any], HamiltonLineageCollector]:
    import importlib
    from hamilton import driver

    imported_modules = [importlib.import_module(module) for module in modules]
    lineage_collector = HamiltonLineageCollector()

    builder = (
        driver.Builder()
        .with_config(config)
        .with_modules(*imported_modules)
        .with_adapters(lineage_collector, *adapters)
    )
    if materializers:
        builder.with_materializers(*materializers)
    if allow_module_overrides:
        builder.allow_module_overrides()

    result = builder.build().execute(
        final_vars=final_vars,
        inputs=inputs,
        overrides=overrides,
    )
    return result, lineage_collector

modules are strings for DAG serialization and parse-time lightness. If future users need module objects for advanced local use, that can be added explicitly later; v1 should keep the provider serialization-friendly.

---

OpenLineage

OpenLineage support is required for v1. Hamilton already exposes lineage-capable metadata through materializers and data loader/saver decorators; the Airflow provider should translate that metadata into Airflow's OpenLineage integration.

Provider behavior

• Add a HamiltonLineageCollector lifecycle adapter that records Hamilton loader/saver node metadata during execution.
• Implement get_openlineage_facets_on_complete(self, task_instance) on HamiltonOperator and _HamiltonDecoratedOperator.
• Lazily import Airflow OpenLineage classes from the OpenLineage provider inside the lineage method.
• Convert collected file metadata and SQL metadata into OperatorLineage(inputs=[...], outputs=[...]).
• Return empty OperatorLineage() when no loader/saver/materializer metadata was collected.
• Log extraction failures, but do not fail a successful Hamilton task only because optional lineage extraction failed.

Implementation Sketch

class HamiltonOpenLineageMixin:
    def get_openlineage_facets_on_complete(self, task_instance):
        from airflow.providers.openlineage.extractors import OperatorLineage

        if self._lineage_collector is None:
            return OperatorLineage()
        return self._lineage_collector.to_operator_lineage()

The collector should rely on Hamilton metadata produced by @load_from, @save_to, @dataloader, @datasaver, and materializers. The docs should show that users get the best lineage when Hamilton performs or describes the I/O boundary instead of hiding all I/O inside arbitrary Python code.

---

XCom Strategy

Current Airflow Reality

The provider should not claim that XCom values must always be small. With the default database-backed XCom implementation, large values remain a poor fit. With a configured custom XCom backend, especially object-storage backends from Airflow Common IO patterns, deployments can route large XCom values to external storage and keep only references or metadata in the Airflow database.

The provider should therefore avoid prescribing a single persistence pattern:

Pattern	Mechanism	When to use
Airflow-managed XCom	Return Hamilton dict[str, Any]; Airflow's configured XCom backend stores or routes it.	Teams with an object-storage XCom backend or small scalar results
Hamilton-managed persistence	Use @save_to, @datasaver, or materializers and return metadata/path-like values.	Durable datasets, cross-system sharing, lineage-rich workflows
Result shaping	Use result_handler before Airflow handles XCom.	Compatibility, custom serialization, or selecting a subset of results

Recommended Durable Dataset Pattern

Hamilton modules can use @save_to.* decorators to persist durable outputs. This is a data product and lineage choice, not a workaround for XCom in every deployment.

from hamilton.function_modifiers import load_from, save_to, source

@load_from.csv(path=source("csv_path"))
def raw_data(data: pd.DataFrame) -> pd.DataFrame:
    return data

def features(raw_data: pd.DataFrame) -> pd.DataFrame:
    ...

@save_to.parquet(path=source("output_path"), output_name_="save_features")
def features_to_parquet(features: pd.DataFrame) -> pd.DataFrame:
    return features

HamiltonOperator(
    task_id="transform",
    modules=["my_project.prepare_data"],
    inputs={
        "csv_path": "{{ ti.xcom_pull('extract') }}",
        "output_path": "/data/features/{{ ds }}.parquet",
    },
    final_vars=["save_features"],
)

Custom XCom Backend Scope

Do not build a Hamilton-specific custom XCom backend in v1. Users should configure Airflow's XCom backend for their deployment. The provider documentation should link to current Airflow/Astronomer guidance on custom XCom backend strategies and object-storage routing.

---

Operator vs. Decorator Decision Guide

Criterion	HamiltonOperator	@task.hamilton
Static Hamilton modules	Yes	Yes
Runtime Python input preparation	No	Yes
Airflow mapped task compatibility	Standard operator mapping	Function args remain individually mappable
Builder-level control	Yes, via operator kwargs	Yes, via decorator kwargs
Best for	Static pipelines, ops-owned DAGs	TaskFlow DAGs, prepared inputs, data-science-owned task bodies

---

Tests and Documentation

Tests and docs are mandatory for the first PR.

Unit Tests

• Provider info registers the operator and task decorator.
• HamiltonOperator imports modules at execution time and returns dict[str, Any].
• HamiltonOperator applies config, inputs, overrides, adapters, materializers, and result_handler.
• JSON object strings are accepted by coerce_json_object; non-object strings raise AirflowException with render_template_as_native_obj=True guidance.
• _HamiltonDecoratedOperator accepts normal TaskFlow args and merges returned runtime inputs.
• @task.hamilton rejects non-dict, non-None callable returns.
• @task.hamilton rejects coroutine callables in v1.
• No implementation path reads or writes Airflow metadata database tables directly.
• OpenLineage conversion returns empty lineage when no metadata exists.

Integration Tests

Unit tests with mocks cannot honestly validate behaviors that depend on the Airflow 3 task lifecycle (template rendering, mapped-task expansion, OpenLineage facet emission). These run against an in-process Airflow 3 task runner using temporary directories — hermetic, no external services required. Gated behind a [test] extras install so the base provider package stays lean.

• Template rendering with native objects. Execute a HamiltonOperator inside a DAG with render_template_as_native_obj=True; assert that dict-shaped Jinja templates in inputs, config, and overrides arrive as dicts (not stringified) and that a non-object string raises AirflowException with the expected guidance.
• End-to-end Hamilton execution via HamiltonOperator. Run a real Hamilton module end-to-end against a tmp dir using @load_from.csv / @save_to.parquet; assert final_vars are returned and that the file was written.
• TaskFlow chaining and mapped-task expansion via @task.hamilton. Run a small TaskFlow DAG where @task.hamilton consumes upstream output via XCom; additionally exercise mapped-task expansion (.expand(...)) to confirm the decorated operator passes Airflow's mapped-task validation path at scheduler time.
• OpenLineage facet emission. With the OpenLineage provider installed, execute a HamiltonOperator whose Hamilton module uses @load_from / @save_to; invoke get_openlineage_facets_on_complete(task_instance) and assert the returned OperatorLineage carries the expected input and output datasets. Also assert empty OperatorLineage() when the module performs no I/O.

Documentation

• Installation for Airflow 3 only.
• HamiltonOperator quickstart.
• @task.hamilton quickstart using individual TaskFlow arguments.
• Template rendering guidance, including render_template_as_native_obj=True.
• XCom backend guidance that distinguishes default DB XCom from object-storage/custom XCom backends.
• OpenLineage setup and an example using Hamilton @load_from / @save_to.

---

What Is NOT in v1

• Airflow 2 support: v1 is Airflow 3 only.
• Direct Airflow metadata DB access: use public Airflow APIs only.
• Custom XCom backend: deployment concern, not provider scope.
• Airflow Connection type / Hook: Hamilton runs local Python dataflows; there is no single external service connection.
• HamiltonTask / HamiltonDataClass decorator return type: not Airflow-native enough and weak for mapped tasks.
• Async/deferrable execution: requires a separate design using Hamilton async driver semantics and Airflow's deferrable model.
• Dynamic execution (Parallelizable[]): advanced Builder feature. Can be added later.
• IDE auto-completion for @task.hamilton in Airflow's core task object: applicable only if the provider moves into the Airflow monorepo or ships separate typing support.
• System tests against external services: Airflow's tests/system/ convention targets real external systems (S3, BigQuery, Snowflake, etc.). v1 has no external service surface — Hamilton runs local Python and the provider declares no Hook or Connection type — so there is nothing meaningful to system-test against. File I/O paths are exercised by the integration tests against tmp directories.

---

Dependencies

[project]
dependencies = [
    "apache-airflow>=3.0.0",
    "apache-airflow-providers-standard",
    "apache-airflow-providers-openlineage",
    "apache-hamilton",
]

[project.optional-dependencies]
test = [
    "pytest",
    "pandas",
    "pyarrow",
]

The provider may add narrower minimum provider versions once implementation verifies the exact Airflow OpenLineage and standard provider APIs required.

Prior Art

• Existing examples: examples/airflow/ demonstrates manual Hamilton+Airflow integration.
• Snowpark provider: @task.snowpark injects a Snowflake session into the callable and follows Airflow provider decorator patterns.
• PySpark provider: @task.pyspark passes Spark context and demonstrates DecoratedOperator plus a concrete operator.
• Airflow OpenLineage providers: existing providers expose get_openlineage_facets_on_complete() and return OperatorLineage.
• Astronomer XCom backend docs: document custom XCom backend strategies, including object storage behavior.
• Astronomer OpenLineage docs: practical Airflow OpenLineage setup reference.