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
Task Definitions
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
}
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
};
}
Usage Examples
Start Batch Processing Job
# 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
@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
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
ETL Pipeline
Build end-to-end data pipelines
Error Handling
Handle batch processing failures