Обработка больших данных на TypeScript: как перестать бояться и начать анализировать

Мир больших данных и… TypeScript? Серьёзно?

Когда говорят «обработка больших данных», в голову обычно приходит Python с его Pandas и Spark, или Java с Hadoop. Но мир не стоит на месте. TypeScript — это уже не просто язык для «формочек» и кнопочек. Благодаря мощной системе типов, асинхронности и огромной экосистеме Node.js, он превращается в полноценного бойца на поле big data. Представьте: вы можете писать надёжные, типизированные пайплайны данных, которые работают быстро и не падают в продакшене с ошибкой undefined is not a function. Звучит заманчиво? Поехали разбираться.

Почему TypeScript, а не Python?

Не будем спорить, Python — король анализа данных. Но у TypeScript есть козыри в рукаве:

• Типизация на уровне компилятора: Вы отлавливаете ошибки на этапе написания кода, а не после 3-часовой обработки датасета.
• Асинхронность из коробки: async/await и стримы позволяют обрабатывать гигабайты данных, не блокируя поток.
• Единый стек: Если ваша команда пишет фронтенд и бэкенд на TS, не нужно учить новый синтаксис для работы с данными.
• Производительность: Современный Node.js (особенно с V8) очень быстр. Для ETL-процессов и реалтайм-аналитики это идеально.

Конечно, для обучения нейросетей Python всё ещё лучше. Но для построения надёжных data pipelines и обработки потоковых данных TypeScript — отличный выбор.

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

Давайте посмотрим на конкретные техники, которые превращают ваш код в машину для пережёвывания терабайт информации.

1. Стриминг: обрабатываем, не загружая всё в память

Самый частый убийца приложений на Node.js — попытка загрузить огромный JSON-файл целиком. Стримы решают эту проблему. Мы читаем данные кусками (chunks) и обрабатываем их на лету.

Вот пример пайплайна для обработки CSV-файла с миллионами строк:

import fs from 'fs';
import { Transform } from 'stream';
import csv from 'csv-parser';

interface UserData {
  id: string;
  name: string;
  email: string;
  signupDate: string;
}

// Трансформер для валидации и очистки данных
const validateUser = new Transform({
  objectMode: true,
  transform(chunk: UserData, encoding, callback) {
    // Простейшая валидация: отбрасываем строки без email
    if (chunk.email && chunk.email.includes('@')) {
      // Можно менять формат данных на лету
      this.push({
        userId: parseInt(chunk.id, 10),
        displayName: chunk.name.trim(),
        email: chunk.email.toLowerCase(),
        registeredAt: new Date(chunk.signupDate)
      });
    }
    callback();
  }
});

// Создаём pipeline (читаем -> валидируем -> выводим в новый файл)
fs.createReadStream('huge_users.csv')
  .pipe(csv())
  .pipe(validateUser)
  .pipe(fs.createWriteStream('clean_users.ndjson'))
  .on('finish', () => console.log('Обработка завершена! Память спасена.'));

Обратите внимание: мы используем TypeScript-интерфейсы для описания структуры данных. Это делает код самодокументируемым и безопасным.

2. Асинхронные итераторы и генераторы для ленивых вычислений

Если стримы кажутся слишком низкоуровневыми, используйте async generators. Это более современный и читаемый способ работы с потоками данных. Отлично подходит для обработки данных из API или баз данных.

import { createInterface } from 'readline';
import { createReadStream } from 'fs';

// Генератор, который лениво читает строки из файла
async function* readLines(filePath: string): AsyncGenerator<string> {
  const fileStream = createReadStream(filePath);
  const rl = createInterface({ input: fileStream, crlfDelay: Infinity });

  for await (const line of rl) {
    yield line; // Отдаём строку по одной, не загружая всё в массив
  }
}

// Используем генератор для анализа логов
async function analyzeLogs(logFile: string): Promise<void> {
  let errorCount = 0;
  let warningCount = 0;

  for await (const line of readLines(logFile)) {
    if (line.includes('[ERROR]')) errorCount++;
    if (line.includes('[WARN]')) warningCount++;
    // Можно добавить пагинацию или фильтрацию
  }

  console.log(`Найдено ошибок: ${errorCount}, предупреждений: ${warningCount}`);
}

// Запускаем
analyzeLogs('server_2024.log').catch(console.error);

Такой подход экономит память, так как в любой момент времени в оперативке находится только одна строка, а не весь файл.

3. Параллельная обработка с worker_threads

Node.js однопоточный, но это не значит, что мы не можем использовать все ядра процессора. Модуль worker_threads позволяет запускать настоящие потоки для тяжёлых вычислений.

Представьте, что вам нужно посчитать статистику по 10 миллионам записей. Разделим задачу на 4 воркера:

// main.ts
import { Worker } from 'worker_threads';
import path from 'path';

interface StatsResult {
  total: number;
  sum: number;
  min: number;
  max: number;
}

const CHUNK_SIZE = 2500000; // 2.5 млн записей на воркер
const TOTAL_RECORDS = 10_000_000;

async function runParallelProcessing(): Promise<void> {
  const workers: Promise<StatsResult>[] = [];

  for (let i = 0; i < TOTAL_RECORDS; i += CHUNK_SIZE) {
    const start = i;
    const end = Math.min(i + CHUNK_SIZE, TOTAL_RECORDS);

    workers.push(new Promise<StatsResult>((resolve, reject) => {
      const worker = new Worker(path.resolve('./worker.js'), {
        workerData: { start, end }
      });
      worker.on('message', resolve);
      worker.on('error', reject);
    }));
  }

  const results = await Promise.all(workers);
  // Агрегируем результаты от всех воркеров
  const finalStats: StatsResult = results.reduce((acc, curr) => ({
    total: acc.total + curr.total,
    sum: acc.sum + curr.sum,
    min: Math.min(acc.min, curr.min),
    max: Math.max(acc.max, curr.max)
  }));

  console.log('Итоговая статистика:', finalStats);
}

runParallelProcessing();

// worker.js
import { parentPort, workerData } from 'worker_threads';

function calculateChunk(start: number, end: number): StatsResult {
  let sum = 0;
  let min = Infinity;
  let max = -Infinity;
  let total = 0;

  for (let i = start; i < end; i++) {
    // Симуляция сложных вычислений
    const value = Math.sin(i) * 1000 + Math.random() * 500;
    sum += value;
    min = Math.min(min, value);
    max = Math.max(max, value);
    total++;
  }

  return { total, sum, min, max };
}

const result = calculateChunk(