Node.js скрипт для ETL: как собрать данные с 25 API за час

📅 Опубликовано: 9 мая 2026 🔄 Обновлено: 26 июля 2026

ETLNode.jsпарсинг данных

Коротко: Node.js позволяет собрать данные с 25 API за час благодаря асинхронности и конкурентному выполнению запросов. Правильно настроенный ETL-скрипт обрабатывает до 500 запросов в секунду, что в 3-4 раза быстрее синхронных решений на Python.

Node.js скрипт для ETL: как собрать данные с 25 API за час

Содержание

Почему Node.js идеален для ETL-задач
Архитектура скрипта для массового сбора данных
Как реализовать параллельный парсинг API
Оптимизация производительности и мониторинг
Обработка ошибок и восстановление данных
Масштабирование для работы с сотнями источников

Почему Node.js идеален для ETL-задач

Когда к нам в DS495 приходят с задачей собрать данные с десятков API, первый вопрос — какую технологию использовать. Python с его библиотеками кажется очевидным выбором, но на практике Node.js показывает себя гораздо лучше для ETL-операций.

Основное преимущество Node.js — встроенная асинхронность. Пока Python обрабатывает один HTTP-запрос, Node.js уже отправил следующие 50. Мы тестировали оба подхода на реальном проекте: сбор данных о товарах с 15 маркетплейсов.

Метрика	Python (requests)	Python (aiohttp)	Node.js (axios)
Время выполнения	4.5 часа	1.2 часа	0.8 часа
Память	250 МБ	180 МБ	120 МБ
Сложность кода	Простая	Средняя	Простая
Конкурентные запросы	1	100	200+

Event Loop в Node.js работает как диспетчер в аэропорту — не ждёт, когда один самолёт приземлится, а сразу направляет следующие на посадку. Для ETL это критично, потому что 80% времени тратится на ожидание ответов от внешних сервисов.

Node.js позволяет обрабатывать тысячи одновременных подключений, используя один поток. Python требует многопоточности или многопроцессорности, что усложняет код и увеличивает потребление ресурсов.

Ещё один плюс — экосистема. NPM содержит специализированные библиотеки для работы с любыми API: от Instagram до крупных B2B-систем. Часто находишь готовое решение, которое экономит неделю разработки.

Архитектура скрипта для массового сбора данных

При проектировании ETL-скрипта мы следуем модульному подходу. Каждый компонент отвечает за свою задачу и может работать независимо.

Основные модули нашей архитектуры:

API Manager — управляет подключениями и аутентификацией
Queue Manager — распределяет задачи и контролирует нагрузку
Data Transformer — нормализует данные из разных источников
Storage Manager — сохраняет результаты в БД или файлы
Monitor — отслеживает производительность и ошибки

Такая структура позволяет легко добавлять новые API или менять формат выходных данных без переписывания всего скрипта.

Пример базовой структуры проекта:

etl-project/
├── src/
│   ├── managers/
│   │   ├── api-manager.js
│   │   ├── queue-manager.js
│   │   └── storage-manager.js
│   ├── transformers/
│   │   ├── data-normalizer.js
│   │   └── validators.js
│   ├── config/
│   │   ├── apis.json
│   │   └── settings.js
│   └── utils/
│       ├── logger.js
│       └── monitor.js
├── data/
└── logs/

API Manager хранит конфигурацию каждого источника: URL, ключи доступа, лимиты запросов, формат данных. Это позволяет добавлять новые API простым добавлением записи в конфиг.

Queue Manager — сердце системы автоматизации. Он определяет, сколько запросов отправлять одновременно, как обрабатывать rate limits, и в каком порядке запрашивать данные. Например, если один API ограничивает нас 100 запросами в минуту, а другой — 1000, очередь это учитывает.

Как реализовать параллельный парсинг API

Переходим к практике. Покажу, как написать скрипт, который соберёт данные с 25 API за час.

Нужна помощь с этой задачей? Команда DS495 решит её под ключ. Обсудить проект →

Начнём с базового класса для работы с API:

class APIClient {
  constructor(config) {
    this.name = config.name;
    this.baseURL = config.baseURL;
    this.headers = config.headers || {};
    this.rateLimit = config.rateLimit || 100; // запросов в минуту
    this.retryAttempts = config.retryAttempts || 3;
    this.requestQueue = [];
    this.isProcessing = false;
  }

  async makeRequest(endpoint, params = {}) {
    return new Promise((resolve, reject) => {
      this.requestQueue.push({
        endpoint,
        params,
        resolve,
        reject,
        attempts: 0
      });
      
      if (!this.isProcessing) {
        this.processQueue();
      }
    });
  }

  async processQueue() {
    this.isProcessing = true;
    const interval = 60000 / this.rateLimit; // интервал между запросами
    
    while (this.requestQueue.length > 0) {
      const request = this.requestQueue.shift();
      
      try {
        const response = await this.executeRequest(request);
        request.resolve(response);
      } catch (error) {
        if (request.attempts < this.retryAttempts) {
          request.attempts++;
          this.requestQueue.unshift(request); // возвращаем в начало очереди
        } else {
          request.reject(error);
        }
      }
      
      await this.sleep(interval);
    }
    
    this.isProcessing = false;
  }
}

Теперь создадим менеджер для координации всех API:

class ETLManager {
  constructor() {
    this.clients = new Map();
    this.results = [];
    this.errors = [];
    this.startTime = null;
  }

  addAPI(config) {
    const client = new APIClient(config);
    this.clients.set(config.name, client);
  }

  async collectData(tasks) {
    this.startTime = Date.now();
    const promises = [];

    for (const task of tasks) {
      const client = this.clients.get(task.api);
      if (!client) {
        this.errors.push(`API ${task.api} не найден`);
        continue;
      }

      const promise = client.makeRequest(task.endpoint, task.params)
        .then(data => this.processData(task.api, data))
        .catch(error => this.handleError(task.api, error));
      
      promises.push(promise);
    }

    await Promise.allSettled(promises);
    return this.generateReport();
  }

  processData(apiName, rawData) {
    // Трансформация данных под единый формат
    const processed = this.normalizeData(apiName, rawData);
    this.results.push({
      source: apiName,
      data: processed,
      timestamp: new Date().toISOString()
    });
  }
}

Пошаговая инструкция по настройке скрипта:

Установите зависимости: npm install axios bottleneck winston
Создайте конфиг API в файле config/apis.json с ключами и лимитами
Настройте логирование для отслеживания прогресса и ошибок
Определите схему данных для каждого источника
Реализуйте трансформеры для приведения данных к единому формату
Добавьте мониторинг производительности и уведомления об ошибках
Протестируйте на малом объёме данных перед полным запуском

Для более точного контроля rate limits используем библиотеку Bottleneck:

const Bottleneck = require('bottleneck');

// Создаём лимитер для каждого API
const limiter = new Bottleneck({
  minTime: 100, // минимум 100мс между запросами
  maxConcurrent: 10, // максимум 10 одновременных запросов
  reservoir: 1000, // 1000 запросов в период
  reservoirRefreshAmount: 1000,
  reservoirRefreshInterval: 60 * 1000 // обновляем лимит каждую минуту
});

const fetchData = limiter.wrap(async (url, options) => {
  return axios.get(url, options);
});

Оптимизация производительности и мониторинг

Когда скрипт работает с множеством API одновременно, производительность становится критичной. Мы используем несколько техник для максимизации скорости сбора данных.

Техника	Ускорение	Сложность реализации	Рекомендация
Connection pooling	15-25%	Низкая	Обязательно
Batch запросы	40-60%	Средняя	Если API поддерживает
Кэширование	200-500%	Средняя	Для статичных данных
Сжатие gzip	30-50%	Низкая	Всегда включать

Connection pooling настраивается в axios довольно просто:

const https = require('https');
const axios = require('axios');

const httpsAgent = new https.Agent({
  keepAlive: true,
  maxSockets: 50, // максимум соединений на хост
  maxFreeSockets: 10,
  timeout: 60000,
  freeSocketTimeout: 30000
});

const client = axios.create({
  httpsAgent,
  timeout: 30000,
  headers: {
    'Accept-Encoding': 'gzip, deflate, br'
  }
});

Для мониторинга мы создаём дашборд в реальном времени:

class ETLMonitor {
  constructor() {
    this.stats = {
      totalRequests: 0,
      successfulRequests: 0,
      failedRequests: 0,
      avgResponseTime: 0,
      dataPoints: 0,
      startTime: Date.now(),
      lastUpdate: Date.now()
    };
  }

  recordRequest(success, responseTime, dataPoints) {
    this.stats.totalRequests++;
    if (success) {
      this.stats.successfulRequests++;
      this.stats.dataPoints += dataPoints;
    } else {
      this.stats.failedRequests++;
    }
    
    // Обновляем среднее время ответа
    const alpha = 0.1; // коэффициент сглаживания
    this.stats.avgResponseTime = 
      alpha * responseTime + (1 - alpha) * this.stats.avgResponseTime;
    
    this.stats.lastUpdate = Date.now();
  }

  getProgress() {
    const elapsed = (Date.now() - this.stats.startTime) / 1000;
    const requestsPerSecond = this.stats.totalRequests / elapsed;
    const successRate = this.stats.successfulRequests / this.stats.totalRequests * 100;
    
    return {
      elapsed: Math.round(elapsed),
      requestsPerSecond: Math.round(requestsPerSecond),
      successRate: Math.round(successRate * 100) / 100,
      totalDataPoints: this.stats.dataPoints,
      avgResponseTime: Math.round(this.stats.avgResponseTime)
    };
  }
}

Очень важно настроить правильное логирование. Мы используем Winston с несколькими транспортами:

const winston = require('winston');

const logger = winston.createLogger({
  level: 'info',
  format: winston.format.combine(
    winston.format.timestamp(),
    winston.format.errors({ stack: true }),
    winston.format.json()
  ),
  transports: [
    new winston.transports.File({ 
      filename: 'logs/error.log', 
      level: 'error' 
    }),
    new winston.transports.File({ 
      filename: 'logs/combined.log' 
    }),
    new winston.transports.Console({
      format: winston.format.simple()
    })
  ]
});

Обработка ошибок и восстановление данных

При работе с внешними API ошибки неизбежны. Сети падают, серверы перегружаются, API возвращают неожиданные форматы данных. Надёжная обработка ошибок — это то, что отличает production-ready скрипт от прототипа.

Мы классифицируем ошибки на несколько типов:

Временные (5xx, таймауты) — повторяем запрос с экспоненциальной задержкой
Аутентификация (401, 403) — обновляем токены или уведомляем администратора
Лимиты (429) — ждём указанное время и продолжаем
Структурные (неверный JSON) — логируем и пропускаем
Критические (сеть недоступна) — останавливаем процесс с уведомлением

Реализация механизма retry с экспоненциальным backoff:

class RetryManager {
  constructor(maxRetries = 5, baseDelay = 1000, maxDelay = 30000) {
    this.maxRetries = maxRetries;
    this.baseDelay = baseDelay;
    this.maxDelay = maxDelay;
  }

  async executeWithRetry(operation, context) {
    let lastError;
    
    for (let attempt = 0; attempt <= this.maxRetries; attempt++) {
      try {
        return await operation();
      } catch (error) {
        lastError = error;
        
        if (!this.shouldRetry(error, attempt)) {
          break;
        }
        
        const delay = this.calculateDelay(attempt, error);
        console.log(`Попытка ${attempt + 1}/${this.maxRetries} неуспешна. ` +
                   `Ожидание ${delay}мс перед повтором...`);
        
        await this.sleep(delay);
      }
    }
    
    throw lastError;
  }

  shouldRetry(error, attempt) {
    if (attempt >= this.maxRetries) return false;
    
    // Не повторяем для клиентских ошибок (кроме 429)
    if (error.response && error.response.status >= 400 && 
        error.response.status < 500 && error.response.status !== 429) {
      return false;
    }
    
    // Повторяем для сетевых ошибок и серверных ошибок
    return true;
  }

  calculateDelay(attempt, error) {
    // Если сервер указал Retry-After, используем его
    if (error.response && error.response.headers['retry-after']) {
      return parseInt(error.response.headers['retry-after']) * 1000;
    }
    
    // Иначе экспоненциальная задержка с jitter
    const exponentialDelay = this.baseDelay * Math.pow(2, attempt);
    const jitter = Math.random() * 0.3; // ±30% случайности
    const delay = exponentialDelay * (1 + jitter);
    
    return Math.min(delay, this.maxDelay);
  }

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

Для восстановления данных после сбоев мы используем checkpoint-систему:

class CheckpointManager {
  constructor(checkpointFile = 'data/checkpoint.json') {
    this.checkpointFile = checkpointFile;
    this.checkpoint = this.loadCheckpoint();
  }

  saveProgress(apiName, lastProcessedId, processedCount) {
    this.checkpoint[apiName] = {
      lastProcessedId,
      processedCount,
      timestamp: Date.now()
    };
    
    fs.writeFileSync(this.checkpointFile, JSON.stringify(this.checkpoint, null, 2));
  }

  getLastPosition(apiName) {
    return this.checkpoint[apiName] || {
      lastProcessedId: null,
      processedCount: 0
    };
  }

  clearCheckpoint(apiName) {
    delete this.checkpoint[apiName];
    fs.writeFileSync(this.checkpointFile, JSON.stringify(this.checkpoint, null, 2));
  }

  loadCheckpoint() {
    try {
      if (fs.existsSync(this.checkpointFile)) {
        return JSON.parse(fs.readFileSync(this.checkpointFile, 'utf8'));
      }
    } catch (error) {
      console.warn('Не удалось загрузить checkpoint:', error.message);
    }
    return {};
  }
}

При сбое скрипта checkpoint-система позволяет продолжить с места остановки, а не начинать сначала. Это экономит часы работы при обработке больших объёмов данных.

Масштабирование для работы с сотнями источников

Когда количество API источников растёт, архитектура должна масштабироваться. 25 API — это только начало. В некоторых проектах мы работаем с 200+ источниками данных одновременно.

Для таких масштабов используем паттерн Worker Pool:

const { Worker, isMainThread, parentPort, workerData } = require('worker_threads');

class ETLWorkerPool {
  constructor(workerScript, poolSize = 4) {
    this.workerScript = workerScript;
    this.poolSize = poolSize;
    this.workers = [];
    this.taskQueue = [];
    this.activeWorkers = new Set();
    
    this.initializeWorkers();
  }

  initializeWorkers() {
    for (let i = 0; i < this.poolSize; i++) {
      this.createWorker();
    }
  }

  createWorker() {
    const worker = new Worker(this.workerScript);
    
    worker.on('message', (result) => {
      this.handleWorkerMessage(worker, result);
    });
    
    worker.on('error', (error) => {
      console.error('Worker error:', error);
      this.replaceWorker(worker);
    });
    
    this.workers.push(worker);
  }

  async processAPIs(apiConfigs) {
    return new Promise((resolve, reject) => {
      const results = [];
      let completedTasks = 0;
      const totalTasks = apiConfigs.length;

      const taskPromises = apiConfigs.map(config => {
        return this.addTask(config).then(result => {
          results.push(result);
          completedTasks++;
          
          if (completedTasks === totalTasks) {
            resolve(results);
          }
        });
      });
    });
  }

  addTask(apiConfig) {
    return new Promise((resolve, reject) => {
      const task = {
        config: apiConfig,
        resolve,
        reject,
        id: Date.now() + Math.random()
      };
      
      const availableWorker = this.findAvailableWorker();
      if (availableWorker) {
        this.assignTask(availableWorker, task);
      } else {
        this.taskQueue.push(task);
      }
    });
  }

  findAvailableWorker() {
    return this.workers.find(worker => !this.activeWorkers.has(worker));
  }

  assignTask(worker, task) {
    this.activeWorkers.add(worker);
    worker.postMessage({
      type: 'PROCESS_API',
      data: task.config,
      taskId: task.id
    });
    
    // Сохраняем ссылку на промис для обработки результата
    worker.currentTask = task;
  }
}

Сам worker выглядит так:

// worker.js
const { parentPort } = require('worker_threads');
const ETLManager = require('./etl-manager');

parentPort.on('message', async (message) => {
  if (message.type === 'PROCESS_API') {
    try {
      const manager = new ETLManager();
      const result = await manager.processAPI(message.data);
      
      parentPort.postMessage({
        type: 'SUCCESS',
        taskId: message.taskId,
        data: result
      });
    } catch (error) {
      parentPort.postMessage({
        type: 'ERROR',
        taskId: message.taskId,
        error: error.message
      });
    }
  }
});

Для мониторинга распределённой системы добавляем метрики:

class ClusterMonitor {
  constructor() {
    this.metrics = {
      workersActive: 0,
      tasksCompleted: 0,
      tasksInQueue: 0,
      avgTaskDuration: 0,
      errorRate: 0,
      throughput: 0 // задач в секунду
    };
    
    this.startTime = Date.now();
    this.taskHistory = [];
  }

  recordTaskCompletion(duration, success) {
    this.metrics.tasksCompleted++;
    
    if (success) {
      this.taskHistory.push({ duration, timestamp: Date.now() });
      
      // Обновляем среднее время выполнения
      const alpha = 0.1;
      this.metrics.avgTaskDuration = 
        alpha * duration + (1 - alpha) * this.metrics.avgTaskDuration;
    }
    
    this.updateThroughput();
    this.updateErrorRate();
  }

  updateThroughput() {
    const now = Date.now();
    const recentTasks = this.taskHistory.filter(
      task => now - task.timestamp < 60000 // последняя минута
    );
    
    this.metrics.throughput = recentTasks.length / 60; // задач в секунду
  }

  getStatus() {
    const uptime = (Date.now() - this.startTime) / 1000;
    
    return {
      uptime: Math.round(uptime),
      ...this.metrics,
      estimatedCompletion: this.estimateCompletion()
    };
  }
}

Частые вопросы

В: Сколько API можно опрашивать одновременно без потери производительности?

О: Зависит от ресурсов сервера и лимитов API. Обычно 20-50 одновременных подключений оптимально. Мы тестируем на малых объёмах и постепенно увеличиваем нагрузку, отслеживая время отклика.

В: Как обрабатывать API с разными форматами аутентификации?

О: Создаём адаптеры для каждого типа: API ключи, OAuth 2.0, JWT токены, Basic Auth. Каждый адаптер инкапсулирует логику обновления токенов и обработки ошибок аутентификации.

В: Что делать если один API работает очень медленно?

О: Выносим медленные API в отдельные воркеры с увеличенными таймаутами. Можно также кэшировать данные этого API и обновлять их реже, чем быстрые источники.

В: Как избежать блокировки по IP при массовых запросах?

О: Используем proxy rotation, соблюдаем rate limits API, добавляем случайные задержки между запросами. Некоторые API предоставляют увеличенные лимиты для зарегистрированных приложений.

В: Как тестировать ETL скрипт перед продакшеном?

О: Создаём sandbox окружение с mock API серверами, тестируем на малых объёмах данных, проверяем обработку ошибок с помощью chaos engineering. Обязательно тестируем восстановление после сбоев.

В: Стоит ли переходить с Python на Node.js для ETL?

О: Если текущее решение на Python работает и покрывает задачи — не стоит. Но для новых проектов с высокими требованиями к производительности Node.js даёт преимущества в скорости и потреблении ресурсов.

В: Как обеспечить целостность данных при параллельной обработке?

О: Используем транзакции БД, идемпотентные операции, уникальные идентификаторы записей. Checkpoint система помогает восстановить состояние после сбоев без дублирования данных.

Node.js скрипт для ETL: как собрать данные с 25 API за час

Node.js скрипт для ETL: как собрать данные с 25 API за час

Содержание

Почему Node.js идеален для ETL-задач

Архитектура скрипта для массового сбора данных

Как реализовать параллельный парсинг API

Оптимизация производительности и мониторинг

Обработка ошибок и восстановление данных

Масштабирование для работы с сотнями источников

Частые вопросы

Читайте также