> ## Documentation Index
> Fetch the complete documentation index at: https://hyrex.io/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# Batch Processing

> Process large datasets efficiently in manageable chunks

Build scalable batch processing systems that handle large datasets by breaking them into manageable chunks with progress tracking and result aggregation.

## Overview

This example demonstrates efficient batch processing patterns:

* **Dataset chunking** - Split large datasets into optimal batch sizes
* **Parallel processing** - Process multiple batches simultaneously
* **Progress tracking** - Monitor batch completion and overall progress
* **Result aggregation** - Combine results from all processed batches
* **Error recovery** - Handle individual batch failures without stopping entire job
* **Resource optimization** - Balance throughput with memory usage

Perfect for data migrations, bulk operations, report generation, or any large-scale data processing.

## Task Definitions

<CodeGroup>
  ```python Python theme={null}
  from hyrex import HyrexRegistry, HyrexKV
  from pydantic import BaseModel
  from typing import List, Dict, Any, Optional
  import json
  import time
  import math
  from datetime import datetime

  hy = HyrexRegistry()

  class BatchContext(BaseModel):
      batch_id: int
      total_batches: int
      items: List[Dict[str, Any]]
      job_id: str
      batch_size: int
      processing_type: str

  class BatchJobContext(BaseModel):
      job_id: str
      total_items: int
      batch_size: int = 100
      max_parallel_batches: int = 5
      processing_type: str = "default"

  @hy.task(queue="batch-processing", max_retries=3)
  def process_batch(context: BatchContext):
      """Process a single batch of items"""
      
      start_time = time.time()
      results = []
      errors = []
      
      try:
          for i, item in enumerate(context.items):
              try:
                  # Process individual item based on type
                  if context.processing_type == "data_validation":
                      result = validate_data_item(item)
                  elif context.processing_type == "data_enrichment":
                      result = enrich_data_item(item)
                  elif context.processing_type == "format_conversion":
                      result = convert_data_format(item)
                  else:
                      result = process_default_item(item)
                  
                  results.append({
                      "item_id": item.get("id", i),
                      "status": "success",
                      "result": result
                  })
                  
              except Exception as e:
                  errors.append({
                      "item_id": item.get("id", i),
                      "error": str(e),
                      "item": item
                  })
          
          # Store batch results in KV store
          batch_result = {
              "batch_id": context.batch_id,
              "job_id": context.job_id,
              "processed_items": len(results),
              "failed_items": len(errors),
              "results": results,
              "errors": errors,
              "processing_time": time.time() - start_time,
              "processed_at": datetime.now().isoformat()
          }
          
          HyrexKV.set(
              f"batch_result:{context.job_id}:{context.batch_id}",
              json.dumps(batch_result),
              expiry_seconds=86400  # 24 hours
          )
          
          # Update job progress
          update_job_progress(context.job_id, context.batch_id, context.total_batches)
          
          return {
              "batch_id": context.batch_id,
              "processed": len(results),
              "failed": len(errors),
              "success": len(errors) == 0,
              "processing_time": time.time() - start_time
          }
          
      except Exception as e:
          # Batch-level error - store for retry
          error_result = {
              "batch_id": context.batch_id,
              "job_id": context.job_id,
              "status": "failed",
              "error": str(e),
              "failed_at": datetime.now().isoformat()
          }
          
          HyrexKV.set(
              f"batch_error:{context.job_id}:{context.batch_id}",
              json.dumps(error_result),
              expiry_seconds=86400
          )
          
          raise e

  @hy.task(timeout_seconds=3600)  # 1 hour timeout
  def process_large_dataset(context: BatchJobContext):
      """Orchestrate processing of large dataset in batches"""
      
      try:
          # Initialize job tracking
          job_metadata = {
              "job_id": context.job_id,
              "total_items": context.total_items,
              "batch_size": context.batch_size,
              "total_batches": math.ceil(context.total_items / context.batch_size),
              "processing_type": context.processing_type,
              "started_at": datetime.now().isoformat(),
              "status": "processing",
              "completed_batches": 0,
              "failed_batches": 0
          }
          
          HyrexKV.set(
              f"job_metadata:{context.job_id}",
              json.dumps(job_metadata),
              expiry_seconds=86400
          )
          
          # Load dataset (this would typically come from database/file)
          items = load_dataset(context.total_items, context.processing_type)
          
          # Split into batches
          batches = []
          total_batches = math.ceil(len(items) / context.batch_size)
          
          # Submit batches with concurrency control
          active_tasks = []
          batch_tasks = []
          
          for i in range(0, len(items), context.batch_size):
              batch_items = items[i:i + context.batch_size]
              batch_id = i // context.batch_size
              
              batch_context = BatchContext(
                  batch_id=batch_id,
                  total_batches=total_batches,
                  items=batch_items,
                  job_id=context.job_id,
                  batch_size=context.batch_size,
                  processing_type=context.processing_type
              )
              
              # Submit batch for processing
              task = process_batch.send(batch_context)
              batch_tasks.append({
                  "batch_id": batch_id,
                  "task_id": task.task_id,
                  "submitted_at": datetime.now().isoformat()
              })
          
          # Update job metadata with batch info
          job_metadata["batch_tasks"] = batch_tasks
          job_metadata["batches_submitted"] = len(batch_tasks)
          
          HyrexKV.set(
              f"job_metadata:{context.job_id}",
              json.dumps(job_metadata),
              expiry_seconds=86400
          )
          
          return {
              "job_id": context.job_id,
              "status": "batches_submitted",
              "total_batches": total_batches,
              "batch_size": context.batch_size,
              "submitted_at": datetime.now().isoformat()
          }
          
      except Exception as e:
          # Update job status to failed
          job_metadata["status"] = "failed"
          job_metadata["error"] = str(e)
          job_metadata["failed_at"] = datetime.now().isoformat()
          
          HyrexKV.set(
              f"job_metadata:{context.job_id}",
              json.dumps(job_metadata),
              expiry_seconds=86400
          )
          
          raise e

  @hy.task
  def aggregate_batch_results(job_id: str):
      """Aggregate results from all completed batches"""
      
      # Get job metadata
      job_metadata_str = HyrexKV.get(f"job_metadata:{job_id}")
      if not job_metadata_str:
          raise Exception(f"Job metadata not found for {job_id}")
      
      job_metadata = json.loads(job_metadata_str)
      total_batches = job_metadata["total_batches"]
      
      # Collect all batch results
      all_results = []
      all_errors = []
      completed_batches = 0
      failed_batches = 0
      total_processing_time = 0
      
      for batch_id in range(total_batches):
          # Try to get batch result
          batch_result_str = HyrexKV.get(f"batch_result:{job_id}:{batch_id}")
          if batch_result_str:
              batch_result = json.loads(batch_result_str)
              all_results.extend(batch_result["results"])
              all_errors.extend(batch_result["errors"])
              total_processing_time += batch_result["processing_time"]
              completed_batches += 1
          else:
              # Check if batch failed
              batch_error_str = HyrexKV.get(f"batch_error:{job_id}:{batch_id}")
              if batch_error_str:
                  failed_batches += 1
                  batch_error = json.loads(batch_error_str)
                  all_errors.append({
                      "batch_id": batch_id,
                      "batch_error": batch_error["error"],
                      "type": "batch_failure"
                  })
      
      # Generate final report
      final_report = {
          "job_id": job_id,
          "total_items_processed": len(all_results),
          "total_items_failed": len(all_errors),
          "success_rate": (len(all_results) / (len(all_results) + len(all_errors))) * 100 if (len(all_results) + len(all_errors)) > 0 else 0,
          "completed_batches": completed_batches,
          "failed_batches": failed_batches,
          "total_processing_time": total_processing_time,
          "avg_batch_time": total_processing_time / completed_batches if completed_batches > 0 else 0,
          "aggregated_at": datetime.now().isoformat(),
          "detailed_results": all_results[:100],  # First 100 for preview
          "error_summary": summarize_errors(all_errors)
      }
      
      # Store final report
      HyrexKV.set(
          f"final_report:{job_id}",
          json.dumps(final_report),
          expiry_seconds=604800  # 7 days
      )
      
      # Update job metadata
      job_metadata["status"] = "completed" if failed_batches == 0 else "completed_with_errors"
      job_metadata["completed_at"] = datetime.now().isoformat()
      job_metadata["final_report"] = final_report
      
      HyrexKV.set(
          f"job_metadata:{job_id}",
          json.dumps(job_metadata),
          expiry_seconds=604800
      )
      
      return final_report

  def update_job_progress(job_id: str, completed_batch_id: int, total_batches: int):
      """Update job progress tracking"""
      
      progress_key = f"job_progress:{job_id}"
      
      try:
          progress_data = HyrexKV.get(progress_key)
          if progress_data:
              progress = json.loads(progress_data)
          else:
              progress = {"completed_batches": set(), "total_batches": total_batches}
          
          progress["completed_batches"].add(completed_batch_id)
          progress["completion_percentage"] = (len(progress["completed_batches"]) / total_batches) * 100
          progress["last_updated"] = datetime.now().isoformat()
          
          # Convert set to list for JSON serialization
          progress_to_store = {
              **progress,
              "completed_batches": list(progress["completed_batches"])
          }
          
          HyrexKV.set(progress_key, json.dumps(progress_to_store), expiry_seconds=86400)
          
          # Check if all batches are complete
          if len(progress["completed_batches"]) == total_batches:
              # Trigger aggregation
              aggregate_batch_results.send(job_id)
          
      except Exception as e:
          print(f"Failed to update progress for job {job_id}: {e}")

  def load_dataset(total_items: int, processing_type: str) -> List[Dict[str, Any]]:
      """Load dataset for processing (mock implementation)"""
      
      # In production, this would load from database, file, API, etc.
      items = []
      for i in range(total_items):
          if processing_type == "data_validation":
              items.append({
                  "id": i,
                  "email": f"user{i}@example.com",
                  "age": 20 + (i % 60),
                  "country": ["US", "CA", "UK", "DE", "FR"][i % 5]
              })
          elif processing_type == "data_enrichment":
              items.append({
                  "id": i,
                  "user_id": i,
                  "purchase_amount": round((i % 1000) + 10.50, 2)
              })
          else:
              items.append({
                  "id": i,
                  "data": f"item_{i}_data",
                  "value": i * 2
              })
      
      return items

  def process_default_item(item: Dict[str, Any]) -> Dict[str, Any]:
      """Default item processing"""
      return {
          "processed_id": item["id"],
          "processed_value": item.get("value", 0) * 2,
          "processed_at": datetime.now().isoformat()
      }

  def validate_data_item(item: Dict[str, Any]) -> Dict[str, Any]:
      """Validate data item"""
      validations = {
          "email_valid": "@" in item.get("email", ""),
          "age_valid": 0 <= item.get("age", 0) <= 150,
          "country_valid": item.get("country") in ["US", "CA", "UK", "DE", "FR"]
      }
      
      return {
          "item_id": item["id"],
          "validations": validations,
          "is_valid": all(validations.values())
      }

  def enrich_data_item(item: Dict[str, Any]) -> Dict[str, Any]:
      """Enrich data item with additional information"""
      return {
          "original_item": item,
          "enriched_data": {
              "purchase_category": "premium" if item.get("purchase_amount", 0) > 500 else "standard",
              "customer_tier": "gold" if item.get("user_id", 0) % 10 == 0 else "silver"
          },
          "enriched_at": datetime.now().isoformat()
      }

  def convert_data_format(item: Dict[str, Any]) -> Dict[str, Any]:
      """Convert data item to different format"""
      return {
          "legacy_id": item["id"],
          "new_format": {
              "identifier": f"NEW_{item['id']:06d}",
              "metadata": item.get("data", ""),
              "computed_value": item.get("value", 0) ** 2
          }
      }

  def summarize_errors(errors: List[Dict[str, Any]]) -> Dict[str, Any]:
      """Summarize error patterns"""
      error_types = {}
      for error in errors:
          error_type = error.get("type", "processing_error")
          if error_type not in error_types:
              error_types[error_type] = 0
          error_types[error_type] += 1
      
      return {
          "total_errors": len(errors),
          "error_types": error_types,
          "sample_errors": errors[:10]  # First 10 errors as samples
      }
  ```

  ```typescript TypeScript theme={null}
  import { HyrexRegistry, HyrexKV } from 'hyrex';
  import { z } from 'zod';

  const hy = new HyrexRegistry();

  const BatchContextSchema = z.object({
      batchId: z.number(),
      totalBatches: z.number(),
      items: z.array(z.record(z.any())),
      jobId: z.string(),
      batchSize: z.number(),
      processingType: z.string()
  });

  const BatchJobContextSchema = z.object({
      jobId: z.string(),
      totalItems: z.number(),
      batchSize: z.number().default(100),
      maxParallelBatches: z.number().default(5),
      processingType: z.string().default("default")
  });

  type BatchContext = z.infer<typeof BatchContextSchema>;
  type BatchJobContext = z.infer<typeof BatchJobContextSchema>;

  const processBatch = hy.task({
      name: 'processBatch',
      config: {
          queue: 'batch-processing',
          maxRetries: 3
      },
      argSchema: BatchContextSchema,
      func: async (context: BatchContext) => {
          const startTime = Date.now();
          const results: any[] = [];
          const errors: any[] = [];
          
          try {
              for (let i = 0; i < context.items.length; i++) {
                  const item = context.items[i];
                  
                  try {
                      let result;
                      
                      // Process individual item based on type
                      switch (context.processingType) {
                          case "data_validation":
                              result = validateDataItem(item);
                              break;
                          case "data_enrichment":
                              result = enrichDataItem(item);
                              break;
                          case "format_conversion":
                              result = convertDataFormat(item);
                              break;
                          default:
                              result = processDefaultItem(item);
                      }
                      
                      results.push({
                          itemId: item.id || i,
                          status: "success",
                          result
                      });
                      
                  } catch (error: any) {
                      errors.push({
                          itemId: item.id || i,
                          error: error.message,
                          item
                      });
                  }
              }
              
              // Store batch results in KV store
              const batchResult = {
                  batchId: context.batchId,
                  jobId: context.jobId,
                  processedItems: results.length,
                  failedItems: errors.length,
                  results,
                  errors,
                  processingTime: (Date.now() - startTime) / 1000,
                  processedAt: new Date().toISOString()
              };
              
              await HyrexKV.set(
                  `batch_result:${context.jobId}:${context.batchId}`,
                  JSON.stringify(batchResult),
                  86400 // 24 hours
              );
              
              // Update job progress
              await updateJobProgress(context.jobId, context.batchId, context.totalBatches);
              
              return {
                  batchId: context.batchId,
                  processed: results.length,
                  failed: errors.length,
                  success: errors.length === 0,
                  processingTime: (Date.now() - startTime) / 1000
              };
              
          } catch (error: any) {
              // Batch-level error - store for retry
              const errorResult = {
                  batchId: context.batchId,
                  jobId: context.jobId,
                  status: "failed",
                  error: error.message,
                  failedAt: new Date().toISOString()
              };
              
              await HyrexKV.set(
                  `batch_error:${context.jobId}:${context.batchId}`,
                  JSON.stringify(errorResult),
                  86400
              );
              
              throw error;
          }
      }
  });

  const processLargeDataset = hy.task({
      name: 'processLargeDataset',
      config: { timeoutSeconds: 3600 }, // 1 hour
      argSchema: BatchJobContextSchema,
      func: async (context: BatchJobContext) => {
          try {
              // Initialize job tracking
              const jobMetadata = {
                  jobId: context.jobId,
                  totalItems: context.totalItems,
                  batchSize: context.batchSize,
                  totalBatches: Math.ceil(context.totalItems / context.batchSize),
                  processingType: context.processingType,
                  startedAt: new Date().toISOString(),
                  status: "processing",
                  completedBatches: 0,
                  failedBatches: 0
              };
              
              await HyrexKV.set(
                  `job_metadata:${context.jobId}`,
                  JSON.stringify(jobMetadata),
                  86400
              );
              
              // Load dataset
              const items = await loadDataset(context.totalItems, context.processingType);
              
              // Split into batches and submit
              const batchTasks: any[] = [];
              const totalBatches = Math.ceil(items.length / context.batchSize);
              
              for (let i = 0; i < items.length; i += context.batchSize) {
                  const batchItems = items.slice(i, i + context.batchSize);
                  const batchId = Math.floor(i / context.batchSize);
                  
                  const batchContext: BatchContext = {
                      batchId,
                      totalBatches,
                      items: batchItems,
                      jobId: context.jobId,
                      batchSize: context.batchSize,
                      processingType: context.processingType
                  };
                  
                  // Submit batch for processing
                  const task = await processBatch.send(batchContext);
                  batchTasks.push({
                      batchId,
                      taskId: task.taskId,
                      submittedAt: new Date().toISOString()
                  });
              }
              
              // Update job metadata with batch info
              const updatedMetadata = {
                  ...jobMetadata,
                  batchTasks,
                  batchesSubmitted: batchTasks.length
              };
              
              await HyrexKV.set(
                  `job_metadata:${context.jobId}`,
                  JSON.stringify(updatedMetadata),
                  86400
              );
              
              return {
                  jobId: context.jobId,
                  status: "batches_submitted",
                  totalBatches,
                  batchSize: context.batchSize,
                  submittedAt: new Date().toISOString()
              };
              
          } catch (error: any) {
              // Update job status to failed
              const failedMetadata = {
                  jobId: context.jobId,
                  status: "failed",
                  error: error.message,
                  failedAt: new Date().toISOString()
              };
              
              await HyrexKV.set(
                  `job_metadata:${context.jobId}`,
                  JSON.stringify(failedMetadata),
                  86400
              );
              
              throw error;
          }
      }
  });

  const aggregateBatchResults = hy.task({
      name: 'aggregateBatchResults',
      func: async (jobId: string) => {
          // Get job metadata
          const jobMetadataStr = await HyrexKV.get(`job_metadata:${jobId}`);
          if (!jobMetadataStr) {
              throw new Error(`Job metadata not found for ${jobId}`);
          }
          
          const jobMetadata = JSON.parse(jobMetadataStr);
          const totalBatches = jobMetadata.totalBatches;
          
          // Collect all batch results
          const allResults: any[] = [];
          const allErrors: any[] = [];
          let completedBatches = 0;
          let failedBatches = 0;
          let totalProcessingTime = 0;
          
          for (let batchId = 0; batchId < totalBatches; batchId++) {
              // Try to get batch result
              const batchResultStr = await HyrexKV.get(`batch_result:${jobId}:${batchId}`);
              if (batchResultStr) {
                  const batchResult = JSON.parse(batchResultStr);
                  allResults.push(...batchResult.results);
                  allErrors.push(...batchResult.errors);
                  totalProcessingTime += batchResult.processingTime;
                  completedBatches++;
              } else {
                  // Check if batch failed
                  const batchErrorStr = await HyrexKV.get(`batch_error:${jobId}:${batchId}`);
                  if (batchErrorStr) {
                      failedBatches++;
                      const batchError = JSON.parse(batchErrorStr);
                      allErrors.push({
                          batchId,
                          batchError: batchError.error,
                          type: "batch_failure"
                      });
                  }
              }
          }
          
          // Generate final report
          const finalReport = {
              jobId,
              totalItemsProcessed: allResults.length,
              totalItemsFailed: allErrors.length,
              successRate: (allResults.length + allErrors.length) > 0 ? 
                  (allResults.length / (allResults.length + allErrors.length)) * 100 : 0,
              completedBatches,
              failedBatches,
              totalProcessingTime,
              avgBatchTime: completedBatches > 0 ? totalProcessingTime / completedBatches : 0,
              aggregatedAt: new Date().toISOString(),
              detailedResults: allResults.slice(0, 100), // First 100 for preview
              errorSummary: summarizeErrors(allErrors)
          };
          
          // Store final report
          await HyrexKV.set(
              `final_report:${jobId}`,
              JSON.stringify(finalReport),
              604800 // 7 days
          );
          
          // Update job metadata
          const updatedMetadata = {
              ...jobMetadata,
              status: failedBatches === 0 ? "completed" : "completed_with_errors",
              completedAt: new Date().toISOString(),
              finalReport
          };
          
          await HyrexKV.set(
              `job_metadata:${jobId}`,
              JSON.stringify(updatedMetadata),
              604800
          );
          
          return finalReport;
      }
  });

  async function updateJobProgress(jobId: string, completedBatchId: number, totalBatches: number) {
      const progressKey = `job_progress:${jobId}`;
      
      try {
          const progressData = await HyrexKV.get(progressKey);
          let progress: any;
          
          if (progressData) {
              progress = JSON.parse(progressData);
          } else {
              progress = { completedBatches: [], totalBatches };
          }
          
          if (!progress.completedBatches.includes(completedBatchId)) {
              progress.completedBatches.push(completedBatchId);
          }
          
          progress.completionPercentage = (progress.completedBatches.length / totalBatches) * 100;
          progress.lastUpdated = new Date().toISOString();
          
          await HyrexKV.set(progressKey, JSON.stringify(progress), 86400);
          
          // Check if all batches are complete
          if (progress.completedBatches.length === totalBatches) {
              // Trigger aggregation
              await aggregateBatchResults.send(jobId);
          }
          
      } catch (error) {
          console.error(`Failed to update progress for job ${jobId}:`, error);
      }
  }

  async function loadDataset(totalItems: number, processingType: string): Promise<any[]> {
      // In production, this would load from database, file, API, etc.
      const items = [];
      
      for (let i = 0; i < totalItems; i++) {
          switch (processingType) {
              case "data_validation":
                  items.push({
                      id: i,
                      email: `user${i}@example.com`,
                      age: 20 + (i % 60),
                      country: ["US", "CA", "UK", "DE", "FR"][i % 5]
                  });
                  break;
              case "data_enrichment":
                  items.push({
                      id: i,
                      userId: i,
                      purchaseAmount: Math.round(((i % 1000) + 10.5) * 100) / 100
                  });
                  break;
              default:
                  items.push({
                      id: i,
                      data: `item_${i}_data`,
                      value: i * 2
                  });
          }
      }
      
      return items;
  }

  function processDefaultItem(item: any): any {
      return {
          processedId: item.id,
          processedValue: (item.value || 0) * 2,
          processedAt: new Date().toISOString()
      };
  }

  function validateDataItem(item: any): any {
      const validations = {
          emailValid: item.email?.includes("@") || false,
          ageValid: item.age >= 0 && item.age <= 150,
          countryValid: ["US", "CA", "UK", "DE", "FR"].includes(item.country)
      };
      
      return {
          itemId: item.id,
          validations,
          isValid: Object.values(validations).every(Boolean)
      };
  }

  function enrichDataItem(item: any): any {
      return {
          originalItem: item,
          enrichedData: {
              purchaseCategory: (item.purchaseAmount || 0) > 500 ? "premium" : "standard",
              customerTier: (item.userId || 0) % 10 === 0 ? "gold" : "silver"
          },
          enrichedAt: new Date().toISOString()
      };
  }

  function convertDataFormat(item: any): any {
      return {
          legacyId: item.id,
          newFormat: {
              identifier: `NEW_${String(item.id).padStart(6, '0')}`,
              metadata: item.data || "",
              computedValue: Math.pow(item.value || 0, 2)
          }
      };
  }

  function summarizeErrors(errors: any[]): any {
      const errorTypes: { [key: string]: number } = {};
      
      errors.forEach(error => {
          const errorType = error.type || "processing_error";
          errorTypes[errorType] = (errorTypes[errorType] || 0) + 1;
      });
      
      return {
          totalErrors: errors.length,
          errorTypes,
          sampleErrors: errors.slice(0, 10) // First 10 errors as samples
      };
  }
  ```
</CodeGroup>

## Usage Examples

### Start Batch Processing Job

```bash theme={null}
# Process large dataset with validation
curl -X POST http://localhost:8000/batch/start \
  -H "Content-Type: application/json" \
  -d '{
    "job_id": "validation_job_20241201",
    "total_items": 10000,
    "batch_size": 200,
    "processing_type": "data_validation"
  }'

# Check job progress
curl http://localhost:8000/batch/progress/validation_job_20241201

# Get final results
curl http://localhost:8000/batch/results/validation_job_20241201
```

### Monitor Batch Processing

```python theme={null}
@hy.task
def monitor_batch_jobs():
    """Monitor active batch processing jobs"""
    
    active_jobs = get_active_batch_jobs()
    
    for job in active_jobs:
        progress = get_job_progress(job["job_id"])
        
        if progress["completion_percentage"] > 0:
            estimated_completion = calculate_eta(
                progress["completion_percentage"],
                job["started_at"]
            )
            
            print(f"Job {job['job_id']}: {progress['completion_percentage']:.1f}% "
                  f"complete, ETA: {estimated_completion}")
```

## REST API Integration

```python theme={null}
from fastapi import FastAPI, HTTPException
from .tasks import process_large_dataset, get_job_progress, get_final_report

app = FastAPI()

@app.post("/batch/start")
async def start_batch_job(request: BatchJobContext):
    """Start a new batch processing job"""
    
    task = process_large_dataset.send(request)
    
    return {
        "message": "Batch job started",
        "task_id": task.task_id,
        "job_id": request.job_id
    }

@app.get("/batch/progress/{job_id}")
async def get_batch_progress(job_id: str):
    """Get batch job progress"""
    
    progress_data = HyrexKV.get(f"job_progress:{job_id}")
    if not progress_data:
        raise HTTPException(status_code=404, detail="Job not found")
    
    return json.loads(progress_data)

@app.get("/batch/results/{job_id}")
async def get_batch_results(job_id: str):
    """Get final batch processing results"""
    
    results_data = HyrexKV.get(f"final_report:{job_id}")
    if not results_data:
        raise HTTPException(status_code=404, detail="Results not available")
    
    return json.loads(results_data)
```

## Production Considerations

* **Memory management**: Monitor memory usage with large batches
* **Error isolation**: Ensure single batch failures don't affect others
* **Progress persistence**: Store progress to handle worker restarts
* **Result storage**: Use appropriate storage for large result sets
* **Concurrency control**: Balance throughput with resource constraints
* **Data consistency**: Ensure atomic operations for critical data

## Next Steps

<CardGroup cols={2}>
  <Card title="ETL Pipeline" href="/docs/examples/data-processing/etl-pipeline" icon="diagram-project">
    Build end-to-end data pipelines
  </Card>

  <Card title="Error Handling" href="/docs/examples/monitoring/error-handling" icon="triangle-exclamation">
    Handle batch processing failures
  </Card>
</CardGroup>