implement-batch-processing

name	implement-batch-processing
description	Implement high-performance batch API operations with job queues, progress...
shortcut	btch
category	api
difficulty	intermediate
estimated_time	2-4 hours
version	2.0.0

Implement Batch Processing

Creates high-performance batch API processing infrastructure for handling bulk operations efficiently. Implements job queues with Bull/BullMQ, real-time progress tracking, transaction management, and intelligent error recovery. Supports millions of records with optimal resource utilization.

When to Use

Use this command when:

• Processing thousands or millions of records in bulk operations
• Import/export functionality requires progress feedback
• Long-running operations exceed HTTP timeout limits
• Partial failures need graceful handling and retry logic
• Resource-intensive operations require rate limiting
• Background processing needs monitoring and management
• Data migration or synchronization between systems

Do NOT use this command for:

• Simple CRUD operations on single records
• Real-time operations requiring immediate responses
• Operations that must be synchronous by nature
• Small datasets that fit in memory (<1000 records)

Prerequisites

Before running this command, ensure:

• Redis is available for job queue management
• Database supports transactions or bulk operations
• API rate limits and quotas are understood
• Error handling strategy is defined
• Monitoring infrastructure is in place

Process

Step 1: Analyze Batch Requirements

The command examines your data processing needs:

• Identifies optimal batch sizes based on memory and performance
• Determines transaction boundaries for consistency
• Maps data validation requirements
• Calculates processing time estimates
• Defines retry and failure strategies

Step 2: Implement Job Queue System

Sets up Bull/BullMQ for reliable job processing:

• Queue configuration with concurrency limits
• Worker processes for parallel execution
• Dead letter queues for failed jobs
• Priority queues for urgent operations
• Rate limiting to prevent overload

Step 3: Create Batch API Endpoints

Implements RESTful endpoints for batch operations:

• Job submission with validation
• Status checking and progress monitoring
• Result retrieval with pagination
• Job cancellation and cleanup
• Error log access

Step 4: Implement Processing Logic

Creates efficient batch processing workflows:

• Chunked processing for memory efficiency
• Transaction management for data integrity
• Progress reporting at configurable intervals
• Error aggregation and reporting
• Result caching for retrieval

Step 5: Add Monitoring & Observability

Integrates comprehensive monitoring:

• Job metrics and performance tracking
• Error rate monitoring and alerting
• Queue depth and processing rate
• Resource utilization metrics
• Business-level success metrics

Output Format

The command generates a complete batch processing system:

batch-processing/
├── src/
│   ├── queues/
│   │   ├── batch-queue.js
│   │   ├── workers/
│   │   │   ├── batch-processor.js
│   │   │   └── chunk-worker.js
│   │   └── jobs/
│   │       ├── import-job.js
│   │       └── export-job.js
│   ├── api/
│   │   ├── batch-controller.js
│   │   └── batch-routes.js
│   ├── services/
│   │   ├── batch-service.js
│   │   ├── validation-service.js
│   │   └── transaction-manager.js
│   └── utils/
│       ├── chunking.js
│       └── progress-tracker.js
├── config/
│   └── queue-config.js
├── tests/
│   └── batch-processing.test.js
└── docs/
    └── batch-api.md

Examples

Example 1: User Import with Validation and Progress

Scenario: Import 100,000 users from CSV with validation and deduplication

Generated Implementation:

// queues/batch-queue.js
import Queue from 'bull';
import Redis from 'ioredis';

const batchQueue = new Queue('batch-processing', {
  redis: {
    host: process.env.REDIS_HOST,
    port: process.env.REDIS_PORT
  },
  defaultJobOptions: {
    removeOnComplete: 100,
    removeOnFail: 500,
    attempts: 3,
    backoff: {
      type: 'exponential',
      delay: 2000
    }
  }
});

// api/batch-controller.js
class BatchController {
  async createBatchJob(req, res) {
    const { type, data, options = {} } = req.body;

    // Validate batch request
    if (!this.validateBatchRequest(type, data)) {
      return res.status(400).json({
        error: 'Invalid batch request'
      });
    }

    // Create job with unique ID
     jobId = ;

     job =  batchQueue.(type, {
      data,
      : req..,
      : {
        : options. || ,
        : options. !== ,
        : options. || ,
        ...options
      }
    }, {
      jobId,
      : options. || 
    });

    
     res.().({
      : job.,
      : ,
      : .(data.),
      : ,
      : 
    });
  }

   () {
     { jobId } = req.;
     job =  batchQueue.(jobId);

     (!job) {
       res.().({ :  });
    }

     state =  job.();
     progress = job.();

     res.({
      : job.,
      : state,
      : {
        : progress. || ,
        : progress. || ,
        : progress. || ,
        : progress. || ,
        : progress. || ,
        : progress. || ,
        : progress. || 
      },
      : job.,
      : job.,
      : job.,
      : state ===  ? job. : 
    });
  }
}


  {
  () {
    .();
  }

  () {
    batchQueue.(,  (job) => {
       { data, options } = job.;
       chunks = .(data, options.);

       results = {
        : [],
        : [],
        : []
      };

      
       job.({
        : ,
        : data.,
        : chunks.,
        : ,
        : ,
        : 
      });

      
       ( i = ; i < chunks.; i++) {
         chunk = chunks[i];

         {
          
           chunkResults =  .(
            chunk,
            options,
            job
          );

          results..(...chunkResults.);
          results..(...chunkResults.);
          results..(...chunkResults.);

          
           processed = (i + ) * options.;
           job.({
            : .(, (processed / data.) * ),
            : .(processed, data.),
            : data.,
            : results..,
            : results..,
            : i + ,
            : chunks.
          });

          
           (options. && results.. > ) {
            ;
          }
        }  (error) {
          .(, error);

           (options.) {
             error;
          }

          
          chunk.( {
            results..({
              : item,
              : error.
            });
          });
        }
      }

      
       .(job., results);

       {
        : {
          : data.,
          : results..,
          : results..,
          : results..
        },
        : job.
      };
    });
  }

   () {
     results = {
      : [],
      : [],
      : []
    };

    
     trx =  db.();

     {
       ( item  chunk) {
         {
          
           (options.) {
             validation =  .(item);
             (!validation.) {
              results..({
                : item,
                : validation.
              });
              ;
            }
          }

          
           existing =  ()
            .(, item.)
            .();

           (existing) {
             (options.) {
              results..({
                : item,
                : 
              });
              ;
            }   (options.) {
               ()
                .(, item.)
                .(item);
              results..({
                : ,
                : existing.,
                : item
              });
              ;
            }
          }

          
           [userId] =  ().({
            ...item,
            :  (),
            : job.
          });

          results..({
            : ,
            : userId,
            : item
          });

        }  (error) {
          results..({
            : item,
            : error.
          });
        }
      }

      
       trx.();
    }  (error) {
       trx.();
       error;
    }

     results;
  }

  () {
     chunks = [];
     ( i = ; i < array.; i += size) {
      chunks.(array.(i, i + size));
    }
     chunks;
  }
}

---

Example 2: Export with Streaming and Compression

Scenario: Export millions of records with streaming and compression

Generated Streaming Export:

// services/export-service.js
import { Transform } from 'stream';
import zlib from 'zlib';

class ExportService {
  async createExportJob(query, format, options) {
    const job = await batchQueue.add('data-export', {
      query,
      format,
      options
    });

    return job;
  }

  async processExportJob(job) {
    const { query, format, options } = job.data;

    // Create export stream
    const exportStream = this.createExportStream(query, format);
    const outputPath = `/tmp/exports/${job.id}.${format}.gz`;

    // Create compression stream
    const gzip = zlib.createGzip();
    const writeStream = fs.createWriteStream(outputPath);

    let recordCount = 0;
    let errorCount = 0;

    return new Promise((resolve, reject) => {
      exportStream
        .pipe(new Transform({
          () {
            recordCount++;

            
             (recordCount %  === ) {
              job.({
                : recordCount,
                : .(, (recordCount / options.) * )
              });
            }

            (, chunk);
          }
        }))
        .(gzip)
        .(writeStream)
        .(,  () => {
          
           url =  .(outputPath, job.);

          ({
            recordCount,
            errorCount,
            : url,
            :  (.() +  *  *  * )
          });
        })
        .(, reject);
    });
  }

  () {
     stream = db.(query).();

     (format) {
       :
         stream.(.());
       :
         stream.(.());
       :
         stream.(.());
      :
          ();
    }
  }
}

---

Example 3: Parallel Processing with Rate Limiting

Scenario: Process API calls with rate limiting and retry logic

Generated Rate-Limited Processor:

// workers/rate-limited-processor.js
import Bottleneck from 'bottleneck';

class RateLimitedProcessor {
  constructor() {
    // Configure rate limiter: 10 requests per second
    this.limiter = new Bottleneck({
      maxConcurrent: 5,
      minTime: 100 // 100ms between requests
    });
  }

  async processBatch(job) {
    const { items, apiEndpoint, options } = job.data;
    const results = [];

    // Process items with rate limiting
    const promises = items.map((item, index) =>
      this.limiter.schedule(async () => {
        try {
          const result = await this.callAPI(apiEndpoint, item);

          // Update progress
          await job.progress({
            processed: index + 1,
            total: items.length,
            percentage: ((index + 1) / items.) * 
          });

           { : , : result };
        }  (error) {
           {
            : ,
            : error.,
            item
          };
        }
      })
    );

     results =  .(promises);

     {
      : results.( r.).,
      : results.( !r.),
      : items.
    };
  }
}

Error Handling

Error: Job Queue Connection Failed

Symptoms: Jobs not processing, Redis connection errors Cause: Redis server unavailable or misconfigured Solution:

batchQueue.on('error', (error) => {
  console.error('Queue error:', error);
  // Implement fallback or alerting
});

Prevention: Implement Redis Sentinel or cluster for high availability

Error: Memory Exhaustion

Symptoms: Process crashes with heap out of memory Cause: Processing chunks too large for available memory Solution: Reduce chunk size and implement streaming

Error: Transaction Deadlock

Symptoms: Batch processing hangs or fails with deadlock errors Cause: Concurrent transactions competing for same resources Solution: Implement retry logic with exponential backoff

Configuration Options

Option: --chunk-size

• Purpose: Set number of records per processing chunk
• Values: 100-10000 (integer)
• Default: 1000
• Example: /batch --chunk-size 500

Option: --concurrency

• Purpose: Number of parallel workers
• Values: 1-20 (integer)
• Default: 5
• Example: /batch --concurrency 10

Option: --retry-attempts

• Purpose: Number of retry attempts for failed items
• Values: 0-10 (integer)
• Default: 3
• Example: /batch --retry-attempts 5

Best Practices

✅ DO:

• Use transactions for data consistency
• Implement idempotent operations for retry safety
• Monitor queue depth and processing rates
• Store detailed error information for debugging
• Implement circuit breakers for external API calls

❌ DON'T:

• Process entire datasets in memory
• Ignore partial failures in batch operations
• Use synchronous processing for large batches
• Forget to implement job cleanup policies

💡 TIPS:

• Use priority queues for time-sensitive batches
• Implement progressive chunk sizing based on success rate
• Cache validation results to avoid redundant checks
• Use database bulk operations when possible

Related Commands

• /api-rate-limiter - Implement API rate limiting
• /api-event-emitter - Event-driven processing
• /api-monitoring-dashboard - Monitor batch jobs
• /database-bulk-operations - Database-level batch operations

Performance Considerations

• Optimal chunk size: 500-2000 records depending on complexity
• Memory per worker: ~512MB for typical operations
• Processing rate: 1000-10000 records/second depending on validation
• Redis memory: ~1KB per job + result storage

Security Notes

⚠️ Security Considerations:

• Validate all batch input data to prevent injection attacks
• Implement authentication for job status endpoints
• Sanitize error messages to avoid information leakage
• Use separate queues for different security contexts
• Implement job ownership validation

Troubleshooting

Issue: Jobs stuck in queue

Solution: Check worker processes and Redis connectivity

Issue: Slow processing speed

Solution: Increase chunk size and worker concurrency

Issue: High error rates

Solution: Review validation logic and add retry mechanisms

Getting Help

• Bull documentation: https://github.com/OptimalBits/bull
• BullMQ guide: https://docs.bullmq.io
• Redis Streams: https://redis.io/topics/streams

Version History

• v2.0.0 - Complete rewrite with streaming, rate limiting, and advanced error handling
• v1.0.0 - Initial batch processing implementation

---

Last updated: 2025-10-11 Quality score: 9.5/10 Tested with: Bull 4.x, BullMQ 3.x, Redis 7.0