# Hamilton Core

> Core Hamilton patterns for creating DAGs, applying decorators, testing, and debugging dataflows. Use for basic Hamilton development tasks.

- Skill: `apache/hamilton-core` (Agent Skill)
- Install (CLI): `npx skillmds@latest add apache/hamilton-core`
- Raw SKILL.md: https://api.skillmd.com/api/skills/apache/hamilton-core/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Coding & Dev Tools
- Author: apache (https://skillmd.com/u/apache)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/apache/hamilton-core

---

<!-- 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:**
```python
"""
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:**
```python
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:**
```python
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:**
```python
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:**
```python
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:**
```python
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:**
```python
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):**
```python
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):**
```python
"""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:**
```python
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:**
```python
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:**
```python
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:**
```python
# ❌ 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:**
```python
# ❌ 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:**
```python
# ❌ 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