temporal-ray-integration

name	temporal-ray-integration
description	Expert guidance for combining Temporal durable workflows with Ray distributed computing. Use when building resilient parallel processing pipelines, fault-tolerant ML workflows, or systems requiring both durability AND parallelism. Covers integration patterns, error handling across both systems, and production architecture.

Temporal + Ray Integration Skill

Overview

Combining Temporal and Ray provides the best of both worlds:

• Temporal: Durable execution, automatic retry, state persistence, failure recovery
• Ray: Parallel processing, distributed compute, actor-based stateful workers

Together, they enable resilient AND fast distributed applications.

When to Use This Skill

• Building document processing pipelines with parallel extraction
• ML workflows requiring distributed compute with fault tolerance
• Batch processing systems that must complete reliably
• Any workflow where individual steps need parallel execution
• Systems requiring both durability (survive crashes) and parallelism (scale compute)

Architecture Pattern

┌─────────────────────────────────────────────────────────────┐
│  Temporal Workflow (Durable, Resumable)                     │
│  ┌────────────────────────────────────────────────────────┐ │
│  │ Step 1: Validate inputs                                │ │
│  │ Step 2: Parallel processing ◄─── Ray Tasks (Parallel) │ │
│  │ Step 3: Aggregate results   ◄───────────────►          │ │
│  │ Step 4: Final processing                               │ │
│  └────────────────────────────────────────────────────────┘ │
└────────────────┬────────────────────────────────────────────┘
                 │
    ┌────────────┼────────────┐
    ▼            ▼            ▼
┌──────────┐ ┌────────────┐ ┌──────────┐
│ Temporal │ │ Ray Cluster │ │ External │
│ Server   │ │ (Workers)   │ │ APIs     │
└──────────┘ └────────────┘ └──────────┘

Key Principle: Temporal orchestrates the workflow; Ray parallelizes compute within activities.

Integration Pattern: Ray Inside Temporal Activities

# activities/parallel_processing.py
import ray
from temporalio import activity
from typing import List

@activity.defn(name="parallel_process_documents")
async def parallel_process_documents(doc_paths: List[str]) -> List[dict]:
    """
    Temporal Activity that uses Ray for parallel processing.
    
    - Temporal handles: retry, timeout, failure recovery
    - Ray handles: parallel execution across workers
    """
    # Initialize Ray (if not already running)
    if not ray.is_initialized():
        ray.init(ignore_reinit_error=True)
    
    # Define Ray remote task
    @ray.remote
    def process_single_document(path: str) -> dict:
        # Your processing logic here
        return {"path": path, "status": "processed"}
    
    # Launch parallel tasks
    activity.heartbeat("Starting parallel processing")
    futures = [process_single_document.remote(path) for path in doc_paths]
    
    # Wait for all results
    results = ray.get(futures)
    activity.heartbeat(f"Completed {len(results)} documents")
    
    return results

Complete Example: Document Processing Pipeline

1. Project Structure

temporal_ray_project/
├── src/
│   ├── workflows/
│   │   └── document_pipeline.py
│   ├── activities/
│   │   ├── ingestion.py
│   │   ├── ray_processing.py      # Ray-powered activities
│   │   └── llm_extraction.py
│   ├── ray_tasks/
│   │   ├── chunking.py
│   │   ├── embedding.py
│   │   └── extraction.py
│   └── config.py
├── worker.py
└── main.py

2. Ray Tasks Module

# src/ray_tasks/embedding.py
import ray
from sentence_transformers import SentenceTransformer

@ray.remote(num_gpus=0.5)  # Share GPU across 2 actors
class EmbeddingActor:
    """Stateful Ray actor for embedding generation."""
    
    def __init__(self, model_name: str = "all-MiniLM-L6-v2"):
        self.model = SentenceTransformer(model_name, device="cuda")
    
    def embed_batch(self, texts: list[str]) -> list[list[float]]:
        return self.model.encode(texts).tolist()


@ray.remote
def chunk_document(content: str, chunk_size: int = 1000) -> list[str]:
    """Stateless chunking task."""
    chunks = []
    for i in range(0, len(content), chunk_size):
        chunks.append(content[i:i + chunk_size])
    return chunks

3. Temporal Activity Using Ray

# src/activities/ray_processing.py
import ray
from temporalio import activity
from typing import List
from dataclasses import dataclass

@dataclass
class ProcessingResult:
    chunks: list[str]
    embeddings: list[list[float]]
    processing_time_ms: int

@activity.defn(name="ray_parallel_embed")
async def ray_parallel_embed(documents: List[str]) -> ProcessingResult:
    """
    Activity that orchestrates Ray parallel processing.
    
    Temporal provides:
    - Automatic retry on failure
    - Timeout protection
    - State persistence (if we crash, we retry this whole activity)
    
    Ray provides:
    - Parallel execution across workers
    - GPU utilization
    - Efficient data sharing
    """
    import time
    start = time.time()
    
    # Ensure Ray is initialized
    if not ray.is_initialized():
        ray.init(
            num_cpus=8,
            num_gpus=1,
            ignore_reinit_error=True
        )
    
    activity.heartbeat("Ray initialized")
    
    # Import Ray tasks
    from ray_tasks.embedding import EmbeddingActor, chunk_document
    
    # Step 1: Parallel chunking
    activity.heartbeat("Chunking documents in parallel")
    chunk_futures = [chunk_document.remote(doc) for doc in documents]
    all_chunks = ray.get(chunk_futures)
    flat_chunks = [chunk for doc_chunks in all_chunks for chunk in doc_chunks]
    
    # Step 2: Create embedding actor(s)
    activity.heartbeat("Generating embeddings")
    embedder = EmbeddingActor.remote()
    
    # Batch embeddings (to avoid GPU OOM)
    BATCH_SIZE = 32
    all_embeddings = []
    for i in range(0, len(flat_chunks), BATCH_SIZE):
        batch = flat_chunks[i:i + BATCH_SIZE]
        embeddings = ray.get(embedder.embed_batch.remote(batch))
        all_embeddings.extend(embeddings)
        activity.heartbeat(f"Embedded {min(i + BATCH_SIZE, len(flat_chunks))}/{len(flat_chunks)}")
    
    elapsed_ms = int((time.time() - start) * 1000)
    
    return ProcessingResult(
        chunks=flat_chunks,
        embeddings=all_embeddings,
        processing_time_ms=elapsed_ms
    )

4. Temporal Workflow

# src/workflows/document_pipeline.py
from datetime import timedelta
from dataclasses import dataclass
from temporalio import workflow
from temporalio.common import RetryPolicy

with workflow.unsafe.imports_passed_through():
    from activities.ingestion import load_documents
    from activities.ray_processing import ray_parallel_embed, ProcessingResult
    from activities.llm_extraction import extract_with_fallback

@dataclass
class PipelineInput:
    document_paths: list[str]
    extraction_model: str = "gpt-5.2"

@dataclass
class PipelineOutput:
    entities: list[dict]
    total_chunks: int
    processing_time_ms: int
    models_used: list[str]

@workflow.defn(name="DocumentProcessingPipeline")
class DocumentProcessingPipeline:
    """
    Durable document processing workflow.
    
    If process crashes at any step:
    - Temporal resumes from the last completed step
    - Ray tasks within activities are re-executed (they're idempotent)
    """
    
    @workflow.run
    async def run(self, input: PipelineInput) -> PipelineOutput:
        # Step 1: Load documents (simple activity)
        documents = await workflow.execute_activity(
            load_documents,
            input.document_paths,
            start_to_close_timeout=timedelta(minutes=5),
            retry_policy=RetryPolicy(maximum_attempts=3),
        )
        
        # Step 2: Parallel processing with Ray (long-running activity)
        processing_result: ProcessingResult = await workflow.execute_activity(
            ray_parallel_embed,
            documents,
            start_to_close_timeout=timedelta(minutes=30),
            heartbeat_timeout=timedelta(minutes=2),  # Activity must heartbeat
            retry_policy=RetryPolicy(
                initial_interval=timedelta(seconds=5),
                maximum_attempts=2,
            ),
        )
        
        # Step 3: LLM extraction with fallback
        models_used = []
        entities = []
        
        for i, chunk in enumerate(processing_result.chunks):
            result = await workflow.execute_activity(
                extract_with_fallback,
                {
                    "chunk": chunk,
                    "embedding": processing_result.embeddings[i],
                    "primary_model": input.extraction_model,
                },
                start_to_close_timeout=timedelta(minutes=5),
                retry_policy=RetryPolicy(maximum_attempts=3),
            )
            entities.append(result["entities"])
            if result["model_used"] not in models_used:
                models_used.append(result["model_used"])
        
        return PipelineOutput(
            entities=entities,
            total_chunks=len(processing_result.chunks),
            processing_time_ms=processing_result.processing_time_ms,
            models_used=models_used,
        )

5. Worker Configuration

# worker.py
import asyncio
import ray
from temporalio.client import Client
from temporalio.worker import Worker

from workflows.document_pipeline import DocumentProcessingPipeline
from activities.ingestion import load_documents
from activities.ray_processing import ray_parallel_embed
from activities.llm_extraction import extract_with_fallback

async def main():
    # Initialize Ray cluster BEFORE starting worker
    ray.init(
        num_cpus=16,
        num_gpus=1,
        object_store_memory=8 * 1024**3,
    )
    print(f"Ray initialized: {ray.cluster_resources()}")
    
    # Connect to Temporal
    client = await Client.connect("localhost:7233")
    
    # Create worker with workflows and activities
    worker = Worker(
        client,
        task_queue="document-processing",
        workflows=[DocumentProcessingPipeline],
        activities=[
            load_documents,
            ray_parallel_embed,
            extract_with_fallback,
        ],
    )
    
    print("Starting Temporal worker...")
    await worker.run()

if __name__ == "__main__":
    asyncio.run(main())

Error Handling Across Systems

Ray Failures Within Temporal Activities

@activity.defn(name="ray_task_with_recovery")
async def ray_task_with_recovery(items: list) -> list:
    """Handle Ray failures gracefully."""
    import ray
    
    if not ray.is_initialized():
        ray.init(ignore_reinit_error=True)
    
    @ray.remote
    def process_item(item):
        # May fail
        return do_processing(item)
    
    try:
        futures = [process_item.remote(item) for item in items]
        results = ray.get(futures)
        return results
        
    except ray.exceptions.RayTaskError as e:
        # Ray task failed - Temporal will retry the whole activity
        activity.heartbeat(f"Ray task failed: {e}")
        raise  # Let Temporal handle retry
        
    except ray.exceptions.RayActorError as e:
        # Ray actor died - Temporal will retry
        activity.heartbeat(f"Ray actor failed: {e}")
        raise

Graceful Shutdown

# worker.py
import signal
import ray

async def main():
    ray.init()
    client = await Client.connect("localhost:7233")
    worker = Worker(...)
    
    # Handle graceful shutdown
    def shutdown_handler(signum, frame):
        print("Shutting down...")
        ray.shutdown()
    
    signal.signal(signal.SIGINT, shutdown_handler)
    signal.signal(signal.SIGTERM, shutdown_handler)
    
    try:
        await worker.run()
    finally:
        ray.shutdown()

Performance Optimization

1. Keep Ray Cluster Running

# Initialize Ray once, reuse across activities
# In worker.py startup, NOT inside each activity

ray.init(
    address="auto",  # Connect to existing cluster
    ignore_reinit_error=True,
)

2. Use Ray Actors for Stateful Operations

# Actors persist between activity invocations
# Good for: loaded ML models, database connections

@ray.remote(num_gpus=1)
class ModelActor:
    def __init__(self):
        self.model = load_heavy_model()  # Load once
    
    def predict(self, data):
        return self.model(data)

# Create actor once, reuse across activities
model_actor = ModelActor.remote()

3. Batch Work Appropriately

# Too many small Ray tasks = overhead
# Too few large tasks = underutilization

# Good: batch items into reasonable chunks
BATCH_SIZE = 100
batches = [items[i:i+BATCH_SIZE] for i in range(0, len(items), BATCH_SIZE)]
futures = [process_batch.remote(batch) for batch in batches]

Observability

Temporal Web UI

• http://localhost:8080
• View workflow execution history
• See activity retries and failures
• Query workflow state

Ray Dashboard

• http://localhost:8265
• Monitor parallel task execution
• View actor utilization
• Track memory usage

Logging Best Practices

import logging
from temporalio import activity
import ray

logger = logging.getLogger(__name__)

@activity.defn
async def my_activity(data):
    # Log at activity level (Temporal)
    logger.info(f"Activity started with {len(data)} items")
    activity.heartbeat("Starting Ray processing")
    
    # Ray tasks can log too
    @ray.remote
    def ray_task(item):
        ray_logger = logging.getLogger("ray.task")
        ray_logger.info(f"Processing item: {item}")
        return process(item)
    
    results = ray.get([ray_task.remote(item) for item in data])
    logger.info(f"Activity completed: {len(results)} results")
    return results

When NOT to Use This Pattern

• Simple sequential workflows: Use Temporal alone
• Stateless parallel jobs: Use Ray alone
• Very short tasks (<100ms): Overhead may exceed benefit
• Memory-constrained environments: Both systems need RAM

Installation

pip install temporalio>=1.9.0 "ray[default]>=2.46.0"

References

• Temporal Documentation
• Ray Documentation
• Temporal Python SDK
• Ray Design Patterns