apache/hamilton/core

Trust tier 2
  • community
  • ext:[{"type":"hook","description":"Pre/post lifecycle hooks"}]
Source check pending

Checking that the SKILL.md file is still in the repository.

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

Install

npx vskill@latest install apache/hamilton/core

Semi-extensibleMentions customization, but needs skill-specific knowledge. Read the extensible skills standard

SKILL.md

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

Security

  • Security scanPassed, 100 of 100Pattern checks on the skill's files.
  • Model reviewPassed

See the full security report

Quality evaluation

  • Task evaluationNot runNo evaluation yet. Run one in Skill Studio.

See evaluation results

Source

Popularity

Repository stars
2,546
Repository forks
198
7-day trend score
39

Stars and forks are for the whole GitHub repository, not this skill alone. Not refreshed yet.

Extension points

  • hookPre/post lifecycle hooks

Learn the extensible skills standard

Works with

Any agent vskill installs to: 56 targets in all. These 39 have a page on this site.

  • Claude Code
  • Cursor
  • GitHub Copilot
  • Windsurf
  • Codex
  • Gemini CLI
  • Cline
  • Amp
  • Roo Code
  • Goose
  • Continue
  • Kilo Code
  • Junie
  • Trae
  • OpenHands
  • Qwen Code
  • AI Agent
  • Tabnine Chat
  • Sourcegraph Cody
  • Amazon Q
  • Firebase Genkit
  • Sweep
  • Devin
  • Phind
  • Mentat
  • Claude Desktop
  • Aider
  • GPT Pilot
  • Plandex
  • bolt.new
  • v0
  • Replit Agent
  • Anse
  • TypingMind
  • BetterChatGPT
  • LibreChat
  • Warp AI
  • Pieces
  • Fig AI
Embed badge
vskill VERIFIED badge
[![vskill VERIFIED](https://verified-skill.com/api/v1/skills/apache/hamilton/core/badge)](https://verified-skill.com/skills/apache/hamilton/core)

Something wrong with this skill or its page? Tell us.