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

Hamilton Observability & UI

This skill covers the Hamilton UI, SDK, and observability patterns for tracking and monitoring your dataflows in development and production.

What is Hamilton UI?

Hamilton UI is a web-based dashboard for:

• Tracking DAG executions - See every run with inputs, outputs, and timing
• Visualizing dataflows - Interactive DAG visualization
• Debugging failures - Inspect errors and intermediate values
• Lineage tracking - Understand data provenance
• Performance monitoring - Identify bottlenecks
• Team collaboration - Share DAGs and results

Quick Start

1. Install Hamilton with UI Support

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

2. Start the Hamilton UI

# Start the UI server locally
hamilton ui

# UI will be available at http://localhost:8241

3. Add Tracking to Your Code

"""Add HamiltonTracker to your driver."""

from hamilton_sdk import adapters
from hamilton import driver

# Create tracker
tracker = adapters.HamiltonTracker(
    project_id=1,  # Your project ID from UI
    username="your.email@example.com",
    dag_name="my_pipeline",
    tags={"environment": "dev", "team": "data-science"},
)

# Build driver with tracker
dr = (
    driver.Builder()
    .with_config(your_config)
    .with_modules(*your_modules)
    .with_adapters(tracker)
    .build()
)

# Execute as normal - runs are automatically tracked!
results = dr.execute(["final_output"], inputs={"data_path": "data.csv"})

4. View in UI

Open http://localhost:8241 and see:

• Your DAG visualization
• Execution history
• Node-level timing
• Input/output values

HamiltonTracker Features

Basic Tracking

"""Minimal tracking setup."""

from hamilton_sdk import adapters

tracker = adapters.HamiltonTracker(
    project_id=1, username="user@example.com", dag_name="etl_pipeline"
)

# Attach to driver
dr = driver.Builder().with_adapters(tracker).build()

Advanced Tracking with Tags

"""Use tags for filtering and organization."""

tracker = adapters.HamiltonTracker(
    project_id=1,
    username="user@example.com",
    dag_name="ml_training",
    tags={
        "environment": "production",
        "model_version": "v2.1",
        "team": "ml-platform",
        "experiment_id": "exp_123",
    },
)

# Tags appear in UI for filtering and search

Async Tracking

"""Track async workflows."""

from hamilton import async_driver
from hamilton_sdk import adapters

tracker = adapters.AsyncHamiltonTracker(
    project_id=1, username="user@example.com", dag_name="async_rag_pipeline"
)

dr = await async_driver.Builder().with_modules(async_module).with_adapters(tracker).build()

result = await dr.execute(["llm_response"], inputs={"query": "test"})

Project Organization

Creating Projects

Projects group related DAGs together:

# Create project via UI
# 1. Open http://localhost:8241
# 2. Click "New Project"
# 3. Name it (e.g., "Customer Analytics")
# 4. Get the project_id

# Or via API
import requests
response = requests.post(
    "http://localhost:8241/api/v1/projects",
    json={"name": "Customer Analytics", "description": "Customer data pipelines"}
)
project_id = response.json()['id']

Organizing by Team

"""Organize DAGs by team and environment."""

# Team A - Development
tracker_team_a_dev = adapters.HamiltonTracker(
    project_id=1,  # "Team A Analytics" project
    username="user@example.com",
    dag_name="user_segmentation",
    tags={"team": "team-a", "env": "dev"},
)

# Team A - Production
tracker_team_a_prod = adapters.HamiltonTracker(
    project_id=1,
    username="user@example.com",
    dag_name="user_segmentation",
    tags={"team": "team-a", "env": "prod"},
)

# Team B - Different project
tracker_team_b = adapters.HamiltonTracker(
    project_id=2,  # "Team B ML" project
    username="user@example.com",
    dag_name="recommendation_model",
    tags={"team": "team-b", "env": "dev"},
)

Debugging with Hamilton UI

Inspecting Failed Runs

When a DAG fails, the UI shows:

1. Which node failed - Visual highlighting
2. Error message - Full stack trace
3. Inputs to failed node - Inspect what caused the failure
4. Successful nodes - What completed before failure
5. Timing - Where time was spent before failure

"""DAG fails at 'processed_data' node."""
# In UI:
# - Navigate to failed run
# - Click on red 'processed_data' node
# - See error: "ValueError: Cannot convert string to float"
# - Inspect inputs: raw_data contains 'N/A' strings
# - Fix data cleaning logic

Comparing Runs

Compare two DAG runs side-by-side:

• Input differences
• Timing changes
• Output value changes
• Code changes

"""Compare dev vs prod performance."""
# Run 1: Development (10 seconds)
# Run 2: Production (45 seconds)

# In UI:
# - Select both runs
# - Click "Compare"
# - See: 'feature_engineering' node is 8x slower in prod
# - Reason: Prod has 10x more data
# - Solution: Add caching or parallelize

Node-Level Inspection

Drill into any node to see:

• Execution time
• Input values
• Output values (if stored)
• Error details (if failed)
• Code version

Lineage Tracking

Understanding Data Provenance

Hamilton UI automatically tracks:

• Upstream dependencies - What data contributed to this result?
• Downstream impact - What depends on this node?
• Cross-DAG lineage - Track data between different pipelines

"""Track lineage across training and inference."""

# Training pipeline
training_tracker = adapters.HamiltonTracker(
    project_id=1,
    username="user@example.com",
    dag_name="model_training",
    tags={"stage": "training", "model_version": "v2.1"},
)

# Inference pipeline (same project)
inference_tracker = adapters.HamiltonTracker(
    project_id=1,
    username="user@example.com",
    dag_name="model_inference",
    tags={"stage": "inference", "model_version": "v2.1"},
)

# In UI: Filter by model_version="v2.1" to see both pipelines

Production Monitoring

Key Metrics to Track

"""Track production metrics."""

tracker = adapters.HamiltonTracker(
    project_id=1,
    username="service@example.com",
    dag_name="production_etl",
    tags={
        "environment": "production",
        "service": "data-pipeline",
        "version": os.getenv("SERVICE_VERSION", "unknown"),
        "host": os.getenv("HOSTNAME", "unknown"),
    },
)

# Monitor in UI:
# - Execution frequency (runs per hour)
# - Success rate (failures per day)
# - Execution time trends
# - Node-level performance

Alerting on Failures

"""Set up failure notifications."""
# Hamilton UI can send alerts on:
# - DAG failures
# - Slow executions (> threshold)
# - Specific node failures

# Configure in UI:
# 1. Go to Project Settings
# 2. Set up webhook or email alerts
# 3. Define alert conditions

Performance Monitoring

Track performance over time:

"""Monitor performance degradation."""
# Week 1: Average execution time = 5 minutes
# Week 2: Average execution time = 8 minutes
# Week 3: Average execution time = 12 minutes

# In UI:
# - View execution time chart
# - Identify 'data_processing' node is slowing down
# - Root cause: Data volume increased 3x
# - Solution: Add partitioning or switch to Spark

Integration with Other Tools

MLflow Integration

"""Track both Hamilton and MLflow."""

from hamilton_sdk import adapters
import mlflow

hamilton_tracker = adapters.HamiltonTracker(
    project_id=1, username="user@example.com", dag_name="ml_training"
)

# Use both adapters
dr = driver.Builder().with_adapters(hamilton_tracker, mlflow_tracker).build()

# Results tracked in both Hamilton UI and MLflow

Airflow Integration

"""Track Hamilton DAGs in Airflow tasks."""

from airflow import DAG
from airflow.operators.python import PythonOperator
from hamilton_sdk import adapters


def run_hamilton_pipeline(**context):
    """Execute Hamilton with tracking."""
    tracker = adapters.HamiltonTracker(
        project_id=1,
        username="airflow@example.com",
        dag_name="airflow_etl",
        tags={
            "airflow_dag": context["dag"].dag_id,
            "airflow_run": context["run_id"],
            "task": context["task_instance"].task_id,
        },
    )

    dr = driver.Builder().with_modules(my_module).with_adapters(tracker).build()

    return dr.execute(["output"], inputs=context["params"])


with DAG("my_dag", schedule_interval="@daily") as dag:
    task = PythonOperator(task_id="hamilton_pipeline", python_callable=run_hamilton_pipeline)

SDK Advanced Usage

Querying Runs Programmatically

"""Query Hamilton UI via SDK."""

from hamilton_sdk import client

# Connect to Hamilton UI
hc = client.HamiltonClient(base_url="http://localhost:8241", username="user@example.com")

# Get recent runs
runs = hc.get_runs(project_id=1, dag_name="my_pipeline", limit=10)

for run in runs:
    print(f"Run {run.id}: {run.status} in {run.duration}s")

# Get specific run details
run_detail = hc.get_run(run_id=runs[0].id)
print(f"Inputs: {run_detail.inputs}")
print(f"Outputs: {run_detail.outputs}")

Custom Metadata

"""Add custom metadata to runs."""

tracker = adapters.HamiltonTracker(
    project_id=1,
    username="user@example.com",
    dag_name="my_pipeline",
    tags={
        "git_commit": subprocess.check_output(["git", "rev-parse", "HEAD"]).decode().strip(),
        "git_branch": subprocess.check_output(["git", "rev-parse", "--abbrev-ref", "HEAD"])
        .decode()
        .strip(),
        "dataset_version": "v2024.01",
        "experiment_name": "baseline_v2",
    },
)

# All metadata searchable in UI

Best Practices

1. Use descriptive dag_names - Make them searchable (e.g., "user_segmentation_daily" not "pipeline_1")
2. Tag consistently - Use standard keys (environment, team, version)
3. Track production - Always enable tracking in production
4. Monitor trends - Set up dashboards for key metrics
5. Clean up old runs - Archive or delete runs after retention period
6. Use projects - Organize by team/domain, not by environment
7. Document tags - Create team standard for tag keys and values

Troubleshooting

UI Not Showing Runs

# Check UI is running
curl http://localhost:8241/api/v1/ping

# Check tracker configuration
tracker = adapters.HamiltonTracker(
    project_id=1,  # Does this project exist?
    username="user@example.com",  # Is this user registered?
    dag_name="my_pipeline",
    api_url="http://localhost:8241"  # Override if UI is on different host
)

Slow UI Performance

"""Optimize tracking for large DAGs."""

tracker = adapters.HamiltonTracker(
    project_id=1,
    username="user@example.com",
    dag_name="large_pipeline",
    # Don't capture large outputs
    capture_data_statistics=False,  # Skip stats collection
    # Or be selective about what to capture
)

Additional Resources

• For core Hamilton patterns, use /hamilton-core
• For scaling patterns, use /hamilton-scale
• Hamilton UI docs: hamilton.apache.org/concepts/ui
• Hamilton SDK docs: github.com/apache/hamilton/tree/main/ui/sdk