Batch Processing

Batch processing handles large datasets efficiently. Learn to implement batch operations, queuing systems, and asynchronous processing for resource-intensive tasks.

Challenge

Design and implement batch processing systems that handle large volumes of data efficiently without overwhelming system resources.

Explanation

Batch Processing Patterns

Sequential: Process items one at a time Parallel: Process multiple items concurrently Pipeline: Multi-stage processing pipeline Distributed: Process across multiple workers Queue-Based: Asynchronous job processing

Use Cases

Data imports/exports
Report generation
Image processing
Email campaigns
Log analysis

Code Examples

1. Simple Batch Processor

class BatchProcessor {
  constructor(options = {}) {
    this.batchSize = options.batchSize || 100;
    this.parallelism = options.parallelism || 1;
    this.delay = options.delay || 0;
  }

  async process(items, processor) {
    const results = [];
    const batches = this.createBatches(items, this.batchSize);

    for (const batch of batches) {
      if (this.delay > 0) {
        await this.wait(this.delay);
      }

      const processed = await this.processBatch(batch, processor);
      results.push(...processed);

      console.log(`Processed ${results.length} items`);
    }

    return results;
  }

  async processBatch(batch, processor) {
    const tasks = batch.map(item => processor(item));

    if (this.parallelism === 1) {
      return Promise.all(tasks);
    }

    // Process with limited concurrency
    return this.parallelLimit(tasks, this.parallelism);
  }

  createBatches(items, size) {
    const batches = [];
    for (let i = 0; i < items.length; i += size) {
      batches.push(items.slice(i, i + size));
    }
    return batches;
  }

  async parallelLimit(promises, limit) {
    const results = [];
    const executing = [];

    for (const promise of promises) {
      const task = Promise.resolve(promise).then(result => {
        executing.splice(executing.indexOf(task), 1);
        return result;
      });

      executing.push(task);
      results.push(task);

      if (executing.length >= limit) {
        await Promise.race(executing);
      }
    }

    return Promise.all(results);
  }

  wait(ms) {
    return new Promise(resolve => setTimeout(resolve, ms));
  }
}

// Usage
const processor = new BatchProcessor({
  batchSize: 50,
  parallelism: 5,
  delay: 1000, // 1 second between batches
});

const items = Array.from({ length: 1000 }, (_, i) => i);
const results = await processor.process(items, async (item) => {
  return item * 2;
});

2. Job Queue with Bull

const Queue = require('bull');
const redis = require('redis');

// Create queues
const importQueue = new Queue('import', {
  redis: {
    host: 'localhost',
    port: 6379,
  },
});

// Process jobs
importQueue.process(5, async (job) => {
  const { filePath, userId } = job.data;

  try {
    // Update progress
    job.progress(10);

    // Parse file
    const records = await parseCSV(filePath);
    job.progress(30);

    // Validate records
    const validRecords = records.filter(validateRecord);
    job.progress(50);

    // Import to database
    await database.records.insertBatch(validRecords);
    job.progress(100);

    return {
      imported: validRecords.length,
      skipped: records.length - validRecords.length,
    };
  } catch (error) {
    throw error; // Job will retry
  }
});

// Handle job completion
importQueue.on('completed', (job, result) => {
  console.log(`Job ${job.id} completed:`, result);

  // Notify user
  notificationService.send(job.data.userId, {
    type: 'import-complete',
    data: result,
  });
});

// Handle job failure
importQueue.on('failed', (job, error) => {
  console.log(`Job ${job.id} failed:`, error.message);

  notificationService.send(job.data.userId, {
    type: 'import-failed',
    error: error.message,
  });
});

// Add job
app.post('/import', async (req, res) => {
  const job = await importQueue.add(
    {
      filePath: req.file.path,
      userId: req.user.id,
    },
    {
      attempts: 3,
      backoff: {
        type: 'exponential',
        delay: 2000,
      },
      removeOnComplete: true,
    }
  );

  res.json({ jobId: job.id });
});

// Get job status
app.get('/import/:jobId', async (req, res) => {
  const job = await importQueue.getJob(req.params.jobId);

  if (!job) {
    return res.status(404).json({ error: 'Job not found' });
  }

  const progress = job._progress || 0;
  const state = await job.getState();

  res.json({
    jobId: job.id,
    state,
    progress,
    data: job.data,
  });
});

3. Stream-Based Batch Processing

const fs = require('fs');
const csv = require('csv-parse');
const Transform = require('stream').Transform;

class BatchTransform extends Transform {
  constructor(batchSize = 100, processor) {
    super({ objectMode: true });
    this.batchSize = batchSize;
    this.processor = processor;
    this.batch = [];
  }

  async _transform(record, encoding, callback) {
    this.batch.push(record);

    if (this.batch.length >= this.batchSize) {
      try {
        const results = await this.processor(this.batch);
        this.batch = [];

        results.forEach(result => this.push(result));
        callback();
      } catch (error) {
        callback(error);
      }
    } else {
      callback();
    }
  }

  async _flush(callback) {
    if (this.batch.length > 0) {
      try {
        const results = await this.processor(this.batch);
        results.forEach(result => this.push(result));
        callback();
      } catch (error) {
        callback(error);
      }
    } else {
      callback();
    }
  }
}

// Usage
const input = fs.createReadStream('large-file.csv');
const parser = csv();
const batchTransform = new BatchTransform(100, async (batch) => {
  // Process batch (e.g., database insert)
  return await database.records.insertBatch(batch);
});
const output = fs.createWriteStream('results.json');

input
  .pipe(parser)
  .pipe(batchTransform)
  .pipe(output)
  .on('error', (error) => console.error('Stream error:', error));

4. Distributed Batch Processing

const cluster = require('cluster');
const os = require('os');

const BATCH_SIZE = 100;
const NUM_WORKERS = os.cpus().length;

if (cluster.isMaster) {
  console.log(`Master process ${process.pid} spawning ${NUM_WORKERS} workers`);

  const items = Array.from({ length: 10000 }, (_, i) => i);
  const workers = [];

  // Create workers
  for (let i = 0; i < NUM_WORKERS; i++) {
    const worker = cluster.fork();
    workers.push(worker);
  }

  // Distribute work
  let itemIndex = 0;
  workers.forEach((worker, index) => {
    const batch = items.slice(itemIndex, itemIndex + BATCH_SIZE);
    worker.send({ type: 'process-batch', batch });
    itemIndex += BATCH_SIZE;
  });

  // Handle results
  let completedWorkers = 0;
  workers.forEach(worker => {
    worker.on('message', (msg) => {
      if (msg.type === 'batch-complete') {
        console.log(`Worker ${worker.process.pid} completed batch`);
        completedWorkers++;

        if (completedWorkers === NUM_WORKERS) {
          console.log('All batches processed');
          process.exit(0);
        }

        // Send next batch
        if (itemIndex < items.length) {
          const batch = items.slice(itemIndex, itemIndex + BATCH_SIZE);
          worker.send({ type: 'process-batch', batch });
          itemIndex += BATCH_SIZE;
        }
      }
    });
  });
} else {
  // Worker process
  process.on('message', async (msg) => {
    if (msg.type === 'process-batch') {
      try {
        // Process batch
        const results = await processBatch(msg.batch);
        console.log(`Worker ${process.pid} processed ${msg.batch.length} items`);

        process.send({ type: 'batch-complete', count: results.length });
      } catch (error) {
        console.error('Batch processing error:', error);
      }
    }
  });
}

async function processBatch(items) {
  return items.map(item => item * 2);
}

5. Retry and Error Handling

class BatchProcessorWithRetry {
  constructor(options = {}) {
    this.maxRetries = options.maxRetries || 3;
    this.retryDelay = options.retryDelay || 1000;
    this.backoffMultiplier = options.backoffMultiplier || 2;
  }

  async process(items, processor) {
    const results = [];
    const failures = [];

    for (const item of items) {
      const result = await this.processWithRetry(item, processor);

      if (result.success) {
        results.push(result.data);
      } else {
        failures.push({
          item,
          error: result.error,
          attempts: result.attempts,
        });
      }
    }

    return { results, failures };
  }

  async processWithRetry(item, processor) {
    let lastError;
    let delay = this.retryDelay;

    for (let attempt = 1; attempt <= this.maxRetries; attempt++) {
      try {
        const data = await processor(item);
        return { success: true, data, attempts: attempt };
      } catch (error) {
        lastError = error;

        if (attempt < this.maxRetries) {
          console.log(`Retry ${attempt}/${this.maxRetries - 1} for item ${item}, waiting ${delay}ms`);
          await this.wait(delay);
          delay *= this.backoffMultiplier;
        }
      }
    }

    return {
      success: false,
      error: lastError.message,
      attempts: this.maxRetries,
    };
  }

  wait(ms) {
    return new Promise(resolve => setTimeout(resolve, ms));
  }
}

// Usage
const processor = new BatchProcessorWithRetry({
  maxRetries: 3,
  retryDelay: 1000,
  backoffMultiplier: 2,
});

const { results, failures } = await processor.process(items, async (item) => {
  return await unreliableService.process(item);
});

console.log(`Succeeded: ${results.length}, Failed: ${failures.length}`);

Best Practices

Size Selection
- Balance memory usage and performance
- Test with different batch sizes
- Consider item complexity
Error Handling
- Implement retry logic
- Track failed items
- Log detailed errors
Monitoring
- Track progress
- Monitor resource usage
- Alert on failures
Scalability
- Use distributed processing
- Implement job queues
- Scale horizontally
Data Integrity
- Validate items before processing
- Handle partial failures
- Implement idempotency

Real-World Applications

Data Imports: Bulk CSV/JSON imports
Exports: Large report generation
Image Processing: Thumbnail generation
Email Campaigns: Bulk email sending
Data Analysis: Processing large datasets