# Hamilton Scale

> Performance and parallelization patterns for Hamilton including async I/O, Spark, Ray, Dask, caching, and multithreading. Use for scaling Hamilton workflows.

- Skill: `apache/hamilton-scale` (Agent Skill)
- Install (CLI): `npx skillmds@latest add apache/hamilton-scale`
- Raw SKILL.md: https://api.skillmd.com/api/skills/apache/hamilton-scale/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-scale

---

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

# Hamilton Scaling & Performance

This skill covers parallelization strategies and performance optimization for Apache Hamilton workflows.

## When to Scale

Choose your scaling strategy based on workload:
- **Async**: I/O-bound operations (API calls, database queries)
- **MultiThreading**: Synchronous I/O (legacy APIs without async support)
- **Spark**: Large datasets (multi-GB to multi-TB) on clusters
- **Ray/Dask**: Distributed Python computation
- **Caching**: Avoid redundant expensive computations

## Async Execution for I/O-Bound Workflows

**When to Use:**
- API calls (LLM providers, REST APIs, GraphQL)
- Database queries (async ORM operations)
- Dependent chains of I/O-bound calls
- Multiple parallel external service calls

**Key Benefits:**
- Automatic parallelization of independent async operations
- Natural expression of dependent chains
- Mix async and sync functions in the same DAG
- Significantly faster than sequential execution

**Basic Async Pattern:**
```python
from hamilton import async_driver
import aiohttp
from typing import List

# Mix async and sync functions
async def api_data(user_id: str) -> dict:
    """Fetch from API (async I/O)."""
    async with aiohttp.ClientSession() as session:
        async with session.get(f"https://api.example.com/users/{user_id}") as resp:
            return await resp.json()

def transform_data(api_data: dict) -> dict:
    """Transform data (sync CPU)."""
    return {k: v.upper() if isinstance(v, str) else v for k, v in api_data.items()}

async def save_data(transform_data: dict) -> str:
    """Save to database (async I/O)."""
    async with db_pool.acquire() as conn:
        await conn.execute("INSERT INTO data VALUES ($1)", transform_data)
    return "success"

# Use AsyncDriver
import my_async_module
dr = await async_driver.Builder().with_modules(my_async_module).build()
result = await dr.execute(['save_data'], inputs={'user_id': '123'})
```

**Parallel I/O Operations:**
```python
# These three operations execute in parallel automatically!
async def user_data(user_id: str) -> dict:
    """Fetch user data."""
    async with aiohttp.ClientSession() as session:
        async with session.get(f"https://api.example.com/users/{user_id}") as resp:
            return await resp.json()

async def user_orders(user_id: str) -> List[dict]:
    """Fetch user orders (parallel with user_data)."""
    async with aiohttp.ClientSession() as session:
        async with session.get(f"https://api.example.com/orders?user={user_id}") as resp:
            return await resp.json()

async def user_preferences(user_id: str) -> dict:
    """Fetch preferences (parallel with both above)."""
    async with db_pool.acquire() as conn:
        return await conn.fetchrow("SELECT * FROM preferences WHERE user_id=$1", user_id)

def user_profile(user_data: dict, user_orders: List[dict], user_preferences: dict) -> dict:
    """Combine all data (waits for all three to complete)."""
    return {"data": user_data, "orders": user_orders, "preferences": user_preferences}
```

## MultiThreading for Sync I/O

For synchronous I/O-bound code (legacy APIs, blocking libraries):

```python
from hamilton import driver
from hamilton.execution import executors

# Use multithreading for sync I/O operations
dr = driver.Builder()\
    .with_modules(my_functions)\
    .with_local_executor(executors.MultiThreadingExecutor(max_tasks=10))\
    .build()

# Sync functions that do I/O will run concurrently
results = dr.execute(['final_output'], inputs={...})
```

**When to Use:**
- Synchronous I/O code (requests library, blocking DB drivers)
- Legacy APIs without async support
- Simple parallelization without code rewrite

## Scaling with Apache Spark

**When to Use Spark:**
- Dataset size exceeds single-machine memory (multi-GB to multi-TB)
- You have access to a Spark cluster
- Need distributed data processing

**Basic PySpark Pattern:**
```python
from pyspark.sql import DataFrame as SparkDataFrame, SparkSession
from hamilton.plugins import h_spark

def raw_data(spark_session: SparkSession) -> SparkDataFrame:
    """Load data into Spark."""
    return spark_session.read.csv("data.csv", header=True)

def filtered_data(raw_data: SparkDataFrame) -> SparkDataFrame:
    """Filter using Spark operations."""
    return raw_data.filter(raw_data.age > 18)

def aggregated_stats(filtered_data: SparkDataFrame) -> SparkDataFrame:
    """Aggregate using Spark."""
    return filtered_data.groupBy("country").count()

# Driver Setup
dr = driver.Builder()\
    .with_modules(my_spark_functions)\
    .with_adapters(h_spark.SPARK_INPUT_CHECK)\
    .build()

result = dr.execute(['aggregated_stats'], inputs={'spark_session': spark})
```

**Column-Level Transformations with @with_columns:**
```python
from hamilton.plugins.h_spark import with_columns
import pandas as pd

# File: map_transforms.py
def normalized_amount(amount: pd.Series) -> pd.Series:
    """Pandas UDF for normalization."""
    return (amount - amount.mean()) / amount.std()

def amount_category(normalized_amount: pd.Series) -> pd.Series:
    """Categorize based on normalized amount."""
    return pd.cut(normalized_amount, bins=[-float('inf'), -1, 1, float('inf')],
                  labels=['low', 'medium', 'high'])

# Main dataflow
@with_columns(
    map_transforms,
    columns_to_pass=["amount"]
)
def enriched_data(raw_data: SparkDataFrame) -> SparkDataFrame:
    """Apply pandas UDFs to Spark DataFrame."""
    return raw_data
```

**Spark Best Practices:**
1. Keep transformations lazy - Don't call `.collect()` until final nodes
2. Use `@with_columns` for map operations
3. Partition wisely between major operations
4. Cache DataFrames accessed multiple times
5. Test with small data first (`.limit(1000)`)

## Ray for Distributed Python

**When to Use Ray:**
- Distributed Python computation
- Machine learning workloads
- Custom parallelization logic

```python
from hamilton.plugins import h_ray
import ray

ray.init()

ray_executor = h_ray.RayGraphAdapter(result_builder={"base": dict})

dr = driver.Driver({}, my_functions, adapter=ray_executor)
results = dr.execute(['large_computation'], inputs={...})
```

## Dask for Parallel Computing

**When to Use Dask:**
- Pandas-like operations on larger-than-memory datasets
- Parallel computation on single machine or cluster
- Incremental migration from pandas

```python
from hamilton.plugins import h_dask
from dask import distributed

cluster = distributed.LocalCluster()
client = distributed.Client(cluster)

dask_executor = h_dask.DaskExecutor(client=client)

dr = driver.Builder()\
    .with_remote_executor(dask_executor)\
    .with_modules(my_functions)\
    .build()
```

## Caching for Performance

**When to Use Caching:**
- Expensive API calls (LLM inference, external services)
- Time-consuming data processing
- Iterative development in notebooks
- Shared preprocessing between pipelines

**Basic Caching Setup:**
```python
from hamilton import driver

# Enable caching
dr = driver.Builder()\
    .with_modules(my_functions)\
    .with_cache()\
    .build()

# First execution: computes and caches
result1 = dr.execute(['final_output'], inputs={'data_path': 'data.csv'})

# Second execution: retrieves from cache (instant!)
result2 = dr.execute(['final_output'], inputs={'data_path': 'data.csv'})
```

**Controlling Cache Behavior:**
```python
from hamilton.function_modifiers import cache

# Always recompute (for data loaders)
@cache(behavior="recompute")
def live_api_data(api_key: str) -> dict:
    """Always fetch fresh data."""
    import requests
    response = requests.get("https://api.example.com/data",
                          headers={"Authorization": api_key})
    return response.json()

# Never cache (for non-deterministic operations)
@cache(behavior="disable")
def random_sample(data: pd.DataFrame) -> pd.DataFrame:
    """Random sampling should not be cached."""
    return data.sample(frac=0.1)

# Custom format for efficiency
@cache(format="parquet")
def large_dataframe(processed_data: pd.DataFrame) -> pd.DataFrame:
    """Store as parquet for efficiency."""
    return processed_data
```

**Driver-Level Cache Control:**
```python
dr = driver.Builder()\
    .with_modules(my_functions)\
    .with_cache(
        recompute=['raw_data'],  # Always recompute these
        disable=['random_sample'],  # Never cache these
        path="./my_cache"  # Custom location
    )\
    .build()

# Force complete refresh
dr = driver.Builder()\
    .with_modules(my_functions)\
    .with_cache(recompute=True)\
    .build()
```

**Cache Inspection:**
```python
# Visualize what was cached vs executed
dr.cache.view_run()  # Green = cache hit, Orange = executed

# Access cached results
run_id = dr.cache.last_run_id
data_version = dr.cache.data_versions[run_id]['my_node']
cached_result = dr.cache.result_store.get(data_version)
```

## Choosing the Right Scaling Strategy

**Decision Matrix:**

| Workload Type | Strategy | Use When |
|--------------|----------|----------|
| I/O-bound (async-capable) | AsyncDriver | Multiple API calls, async libraries available |
| I/O-bound (sync only) | MultiThreading | Legacy APIs, blocking I/O |
| Large datasets | Spark | Multi-GB/TB data, cluster available |
| Python computation | Ray/Dask | Custom parallel logic, ML workloads |
| Expensive operations | Caching | Repeated computations, LLM calls |

**Combining Strategies:**
- Cache + Async = Cache expensive API calls, parallelize independent calls
- Spark + Caching = Cache intermediate Spark results
- Ray + Caching = Distribute computation, cache results

## Performance Tips

1. **Profile first** - Identify bottlenecks before optimizing
2. **Start simple** - Begin with caching, add parallelization if needed
3. **Measure impact** - Benchmark before and after changes
4. **Consider costs** - Serialization overhead can negate parallelization benefits
5. **Test at scale** - Small data may hide parallelization overhead

## Additional Resources

- For basic Hamilton patterns, use `/hamilton-core`
- For LLM-specific workflows, use `/hamilton-llm`
- Apache Hamilton docs: hamilton.apache.org/concepts/parallel-task