data-pipeline-backpressure-checkpointing

name	data-pipeline-backpressure-checkpointing
description	Patrón de pipeline de datos con backpressure y checkpointing: procesar flujos de datos sin saturar memoria ni API. Incluye rate limiting, batch processing, tolerancia a fallos y recuperación desde último checkpoint. Basado en Node.js Streams API y patrones de streaming systems.
version	1.0.0
author	Mastermind
license	MIT
metadata	{"hermes":{"tags":["data","pipeline","backpressure","checkpointing","streaming","rate-limiting","batch"],"related_skills":["fetch-paralelo-fallos-parciales","cache-multicapa-memoria-disco","conversion-unidades-api-externa"]}}

Pipeline de Datos con Backpressure y Checkpointing

Patrón para procesar grandes volúmenes de datos de forma fiable: sin saturar la memoria, respetando límites de API, y recuperándose de fallos sin perder datos.

¿Qué es y por qué importa?

Cuando procesas datos de APIs externas (como ESIOS, GTFS, OSM), el flujo normal es:

API → Datos en memoria → Transformar → Guardar

Pero esto falla cuando:

• La API devuelve 10.000+ registros (OOM)
• La API tiene rate limiting (429 Too Many Requests)
• El proceso se cae a mitad (pierdes todo lo procesado)
• La red falla (no hay recuperación)

Backpressure es el mecanismo que dice "espera, estoy procesando todavía" cuando el consumidor va más lento que el productor.

Checkpointing es guardar el progreso para poder reanudar desde donde se quedó si hay un fallo.

Datos reales: Node.js Streams API (documentación oficial) usa backpressure nativamente. Apache Kafka usa exactamente-once semantics con checkpointing. Estos patrones son fundamentales en sistemas de streaming.

Arquitectura del pipeline

┌──────────┐   ┌──────────────┐   ┌─────────────┐   ┌──────────┐
│  Source   │──▶│  Transformer │──▶│  Sink/Store │──▶│ Log/CKPT │
│  (API)   │   │  (validate,  │   │  (DB/file)  │   │ (progress)│
│  stream  │   │   transform) │   │             │   │          │
└──────────┘   └──────────────┘   └─────────────┘   └──────────┘
     │                │                   │              │
     ▼                ▼                   ▼              ▼
  Rate limiter   Schema check       Batch commit     Atomic write
  + jitter       + sanitization     + dedup          + index update

Implementación: Pipeline Class

// src/pipelines/DataPipeline.js
class DataPipeline {
  constructor(options = {}) {
    this.maxConcurrency = options.maxConcurrency || 5;
    this.batchSize = options.batchSize || 100;
    this.rateLimitMs = options.rateLimitMs || 1000; // mínimo entre requests
    this.rateLimitJitterMs = options.rateLimitJitterMs || 500; // jitter para evitar sincronización
    this.checkpointPath = options.checkpointPath || './.pipeline-checkpoint.json';
    this.maxRetries = options.maxRetries || 3;
    this.retryBaseDelayMs = options.retryBaseDelayMs || 1000;
    
    this.processedIds = new Set();
    this.pending = [];
    this.running = 0;
    this.checkpoint = this._loadCheckpoint();
  }

  // ===== RATE LIMITING CON JITTER =====
  _sleep(ms) {
    return new Promise(resolve => setTimeout(resolve, ms));
  }

  _rateLimit() {
    const jitter = Math.random() * this.rateLimitJitterMs;
    return this._sleep(this.rateLimitMs + jitter);
  }

  // ===== CHECKPOINTING =====
  _loadCheckpoint() {
    try {
      const fs = require('fs');
      if (fs.existsSync(this.checkpointPath)) {
        const data = JSON.parse(fs.readFileSync(this.checkpointPath, 'utf8'));
        console.log(`Checkpoint cargado: ${data.processedCount} registros`);
        return data;
      }
    } catch (e) {
      console.warn('No se pudo cargar checkpoint, empezando desde cero');
    }
    return { processedIds: [], processedCount: 0, lastRun: null };
  }

  _saveCheckpoint(processedIds) {
    this.checkpoint = {
      processedIds: Array.from(processedIds),
      processedCount: processedIds.size,
      lastRun: new Date().toISOString()
    };
    try {
      const fs = require('fs');
      // Atomic write: write to temp then rename
      const tmpPath = this.checkpointPath + '.tmp';
      fs.writeFileSync(tmpPath, JSON.stringify(this.checkpoint, null, 2));
      fs.renameSync(tmpPath, this.checkpointPath);
    } catch (e) {
      console.error('Error guardando checkpoint:', e.message);
    }
  }

  // ===== RETRY CON BACKOFF EXPONENCIAL =====
  async _withRetry(fn, context) {
    let lastError;
    for (let attempt = 1; attempt <= this.maxRetries; attempt++) {
      try {
        return await fn.call(context);
      } catch (error) {
        lastError = error;
        
        // Si es 4xx no retryable (400, 401, 403, 404), propagar
        if (error.status && error.status >= 400 && error.status < 500) {
          console.error(`Error no retryable (${error.status}):`, error.message);
          throw error;
        }
        
        const delay = this.retryBaseDelayMs * Math.pow(2, attempt - 1);
        console.warn(`Intento ${attempt}/${this.maxRetries} falló, reintentando en ${delay}ms`);
        await this._sleep(delay);
      }
    }
    throw lastError;
  }

  // ===== BACKPRESSURE: CONCURRENCIA LIMITADA =====
  async _processItem(item) {
    this.running++;
    try {
      // Transformar el dato
      const transformed = await this.transform(item);
      
      // Guardar (sink)
      await this.save(transformed);
      
      this.processedIds.add(item.id);
      return transformed;
    } finally {
      this.running--;
      this._tryDrain();
    }
  }

  // ===== ORQUESTADOR PRINCIPAL =====
  async run(source, transform, save) {
    this.transform = transform;
    this.save = save;
    
    let allItems = [];
    let offset = 0;
    const limit = this.batchSize;
    
    // Fetch con paginación
    do {
      const chunk = await this._withRetry(
        async () => source(offset, limit)
      );
      
      if (!chunk || chunk.length === 0) break;
      
      // Filtrar ya procesados
      const newItems = chunk.filter(item => 
        !this.checkpoint.processedIds.includes(item.id)
      );
      
      console.log(`Nuevos datos: ${newItems.length} (ya procesados: ${chunk.length - newItems.length})`);
      
      // Procesar con backpressure
      for (const item of newItems) {
        // Backpressure: esperar si hay demasiados corriendo
        while (this.running >= this.maxConcurrency) {
          await this._sleep(50);
        }
        
        // Rate limiting entre items
        await this._rateLimit();
        
        // Procesar en paralelo (dentro del límite)
        this._processItem(item);
      }
      
      // Esperar a que terminen todos los pendientes
      await this._drain();
      
      // Guardar checkpoint
      this._saveCheckpoint(this.processedIds);
      
      offset += chunk.length;
    } while (allItems.length > 0);
    
    console.log(`Pipeline completado: ${this.processedIds.size} registros procesados`);
    return this.processedIds.size;
  }

  // ===== DRAIN: esperar a que terminen los pendientes =====
  _drain() {
    return new Promise((resolve) => {
      const check = () => {
        if (this.running === 0) resolve();
        else setTimeout(check, 50);
      };
      check();
    });
  }

  _tryDrain() {
    // Se llama desde _processItem cuando termina un item
    // Si ya no hay running y hay pending, resolver el drain
  }
}

module.exports = DataPipeline;

Uso: Pipeline para datos ESIOS

// src/pipelines/esios-pipeline.js
const DataPipeline = require('./DataPipeline');

const pipeline = new DataPipeline({
  maxConcurrency: 3,       // máx 3 requests en paralelo
  batchSize: 50,           // batch size para paginación
  rateLimitMs: 1000,       // 1s entre requests
  rateLimitJitterMs: 500,  // jitter ±500ms
  checkpointPath: './.checkpoint-esios.json',
  maxRetries: 3,
  retryBaseDelayMs: 1000,
});

// Source: fetcher paginado de ESIOS
async function esiosSource(offset, limit) {
  const response = await fetch(
    `https://api.esios.ree.es/indicators/XXX/values?offset=${offset}&limit=${limit}`,
    { headers: { 'Authorization': `Bearer ${process.env.ESIOS_TOKEN}` } }
  );
  
  if (!response.ok) {
    const error = new Error(`HTTP ${response.status}`);
    error.status = response.status;
    throw error;
  }
  
  return response.json();
}

// Transform: validación + conversión de unidades
function esiosTransform(item) {
  // Schema validation
  if (!item.value || !item.timestamp) {
    throw new Error(`Dato inválido: ${JSON.stringify(item)}`);
  }
  
  // Sanitización
  const value = parseFloat(item.value);
  if (isNaN(value)) {
    console.warn(`Valor no numérico ignorado: ${item.value}`);
    return null;
  }
  
  // Conversión de unidades
  const converted = convertEsiosValue(item.indicator_id, value);
  
  return {
    id: item.id,
    timestamp: item.timestamp,
    value: converted,
    unit: item.unit,
    processedAt: new Date().toISOString(),
  };
}

// Save: guardar en SQLite o archivo
async function esiosSave(record) {
  // Ejemplo: insertar en SQLite
  await db.run(
    'INSERT OR IGNORE INTO esios_data (id, timestamp, value, unit, processed_at) VALUES (?, ?, ?, ?, ?)',
    [record.id, record.timestamp, record.value, record.unit, record.processedAt]
  );
}

// Ejecutar
pipeline.run(esiosSource, esiosTransform, esiosSave)
  .then(count => console.log(`Procesados ${count} registros`))
  .catch(err => console.error('Pipeline falló:', err));

Patrones avanzados

1. Pipeline con Web Workers (frontend)

// En el navegador, usar Web Workers para no bloquear UI
const worker = new Worker('./pipeline-worker.js');

worker.postMessage({
  type: 'start',
  sourceUrl: '/api/esios/data',
  batchSize: 100,
});

worker.onmessage = (e) => {
  switch (e.data.type) {
    case 'progress':
      updateProgressBar(e.data.progress); // 0-100
      break;
    case 'checkpoint':
      console.log(`Checkpoint: ${e.data.processed} registros`);
      break;
    case 'error':
      showError(e.data.message);
      break;
    case 'complete':
      showComplete(e.data.total);
      break;
  }
};

2. Pipeline con Node.js Streams (nativo)

const { Transform, PassThrough } = require('stream');
const { pipeline: pipelineStream } = require('stream/promises');

// Crear streams encadenados
const sourceStream = createApiStream('https://api.esios.ree.es/...');
const transformStream = new Transform({
  objectMode: true,
  transform(chunk, encoding, callback) {
    try {
      const validated = validateSchema(chunk);
      const converted = convertUnits(validated);
      callback(null, converted);
    } catch (err) {
      callback(err); // error propaga al pipeline
    }
  }
});
const sinkStream = createDbStream();

// pipeline() maneja backpressure automáticamente
await pipelineStream(sourceStream, transformStream, sinkStream);

3. Checkpoint con atomicidad

// Patrón write-then-rename para atomicidad
function atomicCheckpoint(data, path) {
  const tmp = path + '.tmp';
  const fd = fs.openSync(tmp, 'w');
  fs.writeSync(fd, JSON.stringify(data));
  fs.closeSync(fd);
  fs.renameSync(tmp, path); // rename es atómico en POSIX
}

Pitfalls

Pitfall	Solución
Checkpoint corrupto tras crash	Usar write-then-rename (atomic) en lugar de write-in-place
Rate limit en ráfaga	Siempre añadir jitter al delay para evitar sincronización entre múltiples consumers
Memory leak con Set gigante	Si procesas millones de IDs, usar un Bloom filter o un SQLite indexado en lugar de un Set
Duplicados tras recovery	Usar INSERT OR IGNORE en DB o verificar antes de insertar
Backpressure no funciona con async/await	Controlar manualmente la concurrencia con un semaphore (como en el ejemplo)
Checkpoint no se guarda	Guardar checkpoint después de cada batch, no solo al final

Cuándo usar / cuándo NO usar

Usar ✅	No usar ❌
APIs con rate limiting (429s frecuentes)	Procesamiento de < 100 registros
Datos que pueden tardar horas en procesarse	Procesamiento en tiempo real (< 100ms latency)
Procesos que pueden fallar a mitad	Entornos sin filesystem (serverless puro)
Múltiples consumidores del mismo dato	Un solo consumo puntual

Referencias

• Node.js Streams API — Documentación oficial de streams con backpressure nativo
• Apache Kafka Streams — Patrones de streaming con exactly-once semantics
• Checkpointing patterns — Confluent docs sobre checkpointing
• Backpressure in Node.js — Rising Stack explanation
• Atomic file operations — Atomicidad en sistemas de archivos POSIX