<!-- SPDX-License-Identifier: Apache-2.0 -->

Hamilton Core Development Assistant

Apache Hamilton is a lightweight Python framework for building Directed Acyclic Graphs (DAGs) of data transformations using declarative, function-based definitions.

Core Principles

Function-Based DAG Definition

• Functions with type hints define nodes in the DAG
• Function parameters automatically create edges (dependencies)
• Function names become node names in the DAG
• Pure functions enable easy testing and reusability

Key Architecture Components

• Functions: Define transformations with parameters as dependencies
• Driver: Builds and manages DAG execution (.execute() runs the DAG)
• FunctionGraph: Internal DAG representation
• Function Modifiers: Decorators that modify DAG behavior
• Adapters: Result formatters and lifecycle hooks

Separation of Concerns

• Definition layer: Pure Python functions (testable, reusable)
• Execution layer: Driver configuration (where/how to run)
• Observation layer: Monitoring, lineage, caching

Common Tasks

1. Creating New Hamilton Modules

Basic Module Structure:

"""
Module docstring explaining the DAG's purpose.
"""

import pandas as pd
from hamilton.function_modifiers import extract_columns


def raw_data(data_path: str) -> pd.DataFrame:
    """Load raw data from source.

    :param data_path: Path to data file (passed as input)
    :return: Raw DataFrame
    """
    return pd.read_csv(data_path)


def cleaned_data(raw_data: pd.DataFrame) -> pd.DataFrame:
    """Remove null values and duplicates.

    :param raw_data: Raw data from previous node
    :return: Cleaned DataFrame
    """
    return raw_data.dropna().drop_duplicates()


def feature_a(cleaned_data: pd.DataFrame) -> pd.Series:
    """Calculate feature A.

    :param cleaned_data: Cleaned data
    :return: Feature A values
    """
    return cleaned_data["column_a"] * 2

Driver Setup:

from hamilton import driver
import my_functions

dr = driver.Driver({}, my_functions)
results = dr.execute(["feature_a", "cleaned_data"], inputs={"data_path": "data.csv"})

Best Practices:

1. ✅ Add type hints to ALL function signatures
2. ✅ Write clear docstrings with :param and :return
3. ✅ Keep functions pure (no side effects)
4. ✅ Name functions after the output they produce
5. ✅ Use function parameters for dependencies (not globals)
6. ✅ Create unit tests for each function
7. ❌ Don't use classes unless needed (functions are preferred)
8. ❌ Don't mutate inputs (return new objects)

2. Applying Function Modifiers (Decorators)

Configuration & Polymorphism:

from hamilton.function_modifiers import config


@config.when(model_type="linear")
def predictions(features: pd.DataFrame) -> pd.Series:
    """Linear model predictions."""
    from sklearn.linear_model import LinearRegression

    model = LinearRegression()
    return model.fit_predict(features)


@config.when(model_type="tree")
def predictions(features: pd.DataFrame) -> pd.Series:
    """Tree model predictions."""
    from sklearn.tree import DecisionTreeRegressor

    model = DecisionTreeRegressor()
    return model.fit_predict(features)


# Use: driver.Driver({'model_type': 'linear'}, module)

Parameterization - Creating Multiple Nodes:

from hamilton.function_modifiers import parameterize


@parameterize(
    rolling_7d={"window": 7},
    rolling_30d={"window": 30},
    rolling_90d={"window": 90},
)
def rolling_average(spend: pd.Series, window: int) -> pd.Series:
    """Calculate rolling average for different windows."""
    return spend.rolling(window).mean()


# Creates 3 nodes: rolling_7d, rolling_30d, rolling_90d

Column Extraction - DataFrames to Series:

from hamilton.function_modifiers import extract_columns


@extract_columns("feature_1", "feature_2", "feature_3")
def features(cleaned_data: pd.DataFrame) -> pd.DataFrame:
    """Generate multiple features."""
    return pd.DataFrame(
        {
            "feature_1": cleaned_data["a"] * 2,
            "feature_2": cleaned_data["b"] ** 2,
            "feature_3": cleaned_data["a"] + cleaned_data["b"],
        }
    )


# Creates 3 nodes: feature_1, feature_2, feature_3 (each a Series)

Data Quality Validation:

from hamilton.function_modifiers import check_output
import pandera as pa


@check_output(data_type=float, range=(0, 100), importance="fail")
def revenue_percentage(revenue: float, total: float) -> float:
    """Calculate revenue as percentage."""
    return (revenue / total) * 100


# With Pandera schemas
@check_output(schema=pa.SeriesSchema(float, pa.Check.greater_than(0)), importance="fail")
def positive_values(data: pd.Series) -> pd.Series:
    """Ensure all values are positive."""
    return data

I/O Materialization:

from hamilton.function_modifiers import save_to, load_from
from hamilton.io.materialization import to


@save_to(to.csv(path="output.csv"))
def final_results(aggregated_data: pd.DataFrame) -> pd.DataFrame:
    """Save final results to CSV."""
    return aggregated_data


@load_from(from_="data.parquet", reader="parquet")
def input_data() -> pd.DataFrame:
    """Load data from parquet."""
    pass  # Function body ignored when using @load_from

3. Converting Existing Code to Hamilton

Before (Script):

import pandas as pd

df = pd.read_csv("data.csv")
df = df.dropna()
df["feature"] = df["col_a"] * 2
result = df.groupby("category")["feature"].mean()
print(result)

After (Hamilton Module):

"""Data processing DAG."""

import pandas as pd


def raw_data(data_path: str) -> pd.DataFrame:
    """Load raw data."""
    return pd.read_csv(data_path)


def cleaned_data(raw_data: pd.DataFrame) -> pd.DataFrame:
    """Remove nulls."""
    return raw_data.dropna()


def feature(cleaned_data: pd.DataFrame) -> pd.Series:
    """Calculate feature."""
    return cleaned_data["col_a"] * 2


def data_with_feature(cleaned_data: pd.DataFrame, feature: pd.Series) -> pd.DataFrame:
    """Add feature to dataset."""
    df = cleaned_data.copy()
    df["feature"] = feature
    return df


def result(data_with_feature: pd.DataFrame) -> pd.Series:
    """Aggregate by category."""
    return data_with_feature.groupby("category")["feature"].mean()

Conversion Guidelines:

1. Identify distinct computation steps
2. Extract each step into a pure function
3. Use previous step's variable name as function parameter
4. Add type hints and docstrings
5. Remove imperative variable assignments
6. Test each function independently

4. Visualizing & Understanding DAGs

Generate Visualization:

from hamilton import driver
import my_functions

dr = driver.Driver({}, my_functions)

# Create visualization
dr.display_all_functions("dag.png")  # All nodes
dr.visualize_execution(
    ["final_output"], "execution.png", inputs={"input_data": ...}
)  # Execution path only

Understanding DAG Structure:

• Each function becomes a node
• Function parameters create directed edges
• No cycles allowed (DAG = Directed Acyclic Graph)
• Execution order determined by dependencies
• Multiple paths execute in parallel when possible

Debugging Tips:

1. Check for circular dependencies (A depends on B depends on A)
2. Verify all parameter names match existing function names
3. Look for typos in parameter names
4. Use dr.list_available_variables() to see all nodes
5. Check dr.what_is_downstream_of('node_name') for dependencies

5. Testing Hamilton Functions

Unit Testing Pattern:

import pytest
import pandas as pd
from my_functions import cleaned_data, feature


def test_cleaned_data():
    """Test data cleaning."""
    raw = pd.DataFrame({"col_a": [1, 2, None, 4], "col_b": ["a", "b", "c", "d"]})
    result = cleaned_data(raw)
    assert len(result) == 3
    assert result["col_a"].isna().sum() == 0


def test_feature():
    """Test feature calculation."""
    data = pd.DataFrame({"col_a": [1, 2, 3]})
    result = feature(data)
    pd.testing.assert_series_equal(result, pd.Series([2, 4, 6], name="col_a"))

Integration Testing with Driver:

def test_full_pipeline():
    """Test complete DAG execution."""
    from hamilton import driver
    import my_functions

    dr = driver.Driver({}, my_functions)
    result = dr.execute(["result"], inputs={"data_path": "test_data.csv"})
    assert "result" in result
    assert result["result"].sum() > 0

Common Pitfalls & Solutions

Circular Dependencies:

# ❌ Bad - circular dependency
def a(b: int) -> int:
    return b + 1


def b(a: int) -> int:
    return a + 1


# ✅ Good - break the cycle
def a(input_value: int) -> int:
    return input_value + 1


def b(a: int) -> int:
    return a + 1

Missing Type Hints:

# ❌ Bad - no type hints
def process(data):
    return data * 2


# ✅ Good - clear types
def process(data: pd.Series) -> pd.Series:
    return data * 2

Mutating Inputs:

# ❌ Bad - mutates input
def add_column(df: pd.DataFrame, col_name: str) -> pd.DataFrame:
    df[col_name] = 0  # Modifies original!
    return df


# ✅ Good - returns new object
def add_column(df: pd.DataFrame, col_name: str) -> pd.DataFrame:
    result = df.copy()
    result[col_name] = 0
    return result

Key Files & Locations

• Core library: hamilton/ - Main package code
• Driver: hamilton/driver.py - Main orchestration class
• Function modifiers: hamilton/function_modifiers/ - Decorators
• Examples: examples/ - Production examples
• Tests: tests/ - Unit and integration tests
• Docs: docs/ - Official documentation

Getting Help

• Documentation: docs/ directory in repo
• Examples: examples/ directory for patterns
• Community: Apache Hamilton Slack, GitHub issues
• Other Skills: Use /hamilton-scale for async/Spark, /hamilton-llm for AI workflows

Additional Resources

For detailed reference material, see:

• Apache Hamilton official docs at hamilton.apache.org
• Apache Hamilton GitHub repository examples folder