apache 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:
- Which node failed - Visual highlighting
- Error message - Full stack trace
- Inputs to failed node - Inspect what caused the failure
- Successful nodes - What completed before failure
- 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
- Use descriptive dag_names - Make them searchable (e.g., "user_segmentation_daily" not "pipeline_1")
- Tag consistently - Use standard keys (environment, team, version)
- Track production - Always enable tracking in production
- Monitor trends - Set up dashboards for key metrics
- Clean up old runs - Archive or delete runs after retention period
- Use projects - Organize by team/domain, not by environment
- 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