Как работать с большими данными, которые не помещаются в память?

Работа с большими данными, которые не помещаются в память

Это классическая задача, с которой встречается каждый аналитик при работе с миллионами или миллиардами строк. Основной принцип: не загружай все в RAM, обрабатывай данные там, где они лежат.

Стратегия 1: Обработка в базе данных (SQL)

Это самый эффективный способ. База данных оптимизирована для работы с большими объемами.

Вместо этого (загрузить все в Python):

# НЕПРАВИЛЬНО! Может потребовать 100 GB памяти
df = pd.read_csv('huge_file.csv')  # Все в память
result = df.groupby('user_id')['revenue'].sum()

Делай так (обработка в БД):

-- В базе данных
SELECT 
  user_id,
  SUM(revenue) as total_revenue,
  COUNT(*) as transactions
FROM orders
GROUP BY user_id
ORDER BY total_revenue DESC;

Почему это работает:

• База использует индексы для быстрого доступа
• Группировка происходит поэтапно, не все данные в памяти
• Запрос оптимизируется query planner'ом

Аналитика на больших объемах в SQL

-- Когортный анализ миллиардов событий
WITH cohorts AS (
  SELECT 
    DATE_TRUNC('month', created_at)::DATE as cohort_month,
    user_id
  FROM users
),
retention AS (
  SELECT 
    c.cohort_month,
    c.user_id,
    EXTRACT(MONTH FROM e.event_date) - EXTRACT(MONTH FROM c.cohort_month) as months_since_signup
  FROM cohorts c
  LEFT JOIN events e ON c.user_id = e.user_id
)
SELECT 
  cohort_month,
  months_since_signup,
  COUNT(DISTINCT user_id) as users
FROM retention
GROUP BY cohort_month, months_since_signup
ORDER BY cohort_month, months_since_signup;

Стратегия 2: Chunking (обработка частями)

Когда нужно обработать CSV или другой файл, читай его по частям:

import pandas as pd

# Читать по 100k строк за раз
chunk_size = 100000
processed_data = []

for chunk in pd.read_csv('huge_file.csv', chunksize=chunk_size):
    # Обработать chunk
    result = chunk[chunk['revenue'] > 1000].groupby('user_id').agg({
        'revenue': 'sum',
        'date': 'count'
    })
    processed_data.append(result)

# Объединить результаты
final_result = pd.concat(processed_data).groupby('user_id').agg('sum')
print(final_result)

Преимущества:

• Используется только 100k * размер_строки памяти
• Можно распараллелить обработку
• Просто реализуется

Стратегия 3: Использование специализированных инструментов

Polars (быстрее Pandas)

import polars as pl

# Polars ленив по умолчанию, не загружает все в память
df = pl.scan_csv('huge_file.csv')

result = (
    df.filter(pl.col('revenue') > 1000)
    .groupby('user_id')
    .agg(pl.col('revenue').sum())
    .collect()  # Только теперь вычислить
)

DuckDB (SQL на больших файлах)

import duckdb

# SQL запрос прямо на CSV без загрузки в память
result = duckdb.sql("""
  SELECT 
    user_id,
    SUM(revenue) as total
  FROM 'huge_file.csv'
  GROUP BY user_id
  HAVING SUM(revenue) > 10000
  ORDER BY total DESC
""").df()

Это очень быстро даже на 100 GB файлах!

PySpark (распределенная обработка)

Для данных которые не помещаются даже на одной машине:

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName('BigData').getOrCreate()

# Spark автоматически разбивает на партиции и обрабатывает параллельно
df = spark.read.csv('huge_file.csv', header=True)

result = (
    df.filter(df.revenue > 1000)
    .groupby('user_id')
    .agg({'revenue': 'sum', 'transaction_id': 'count'})
    .collect()
)

Стратегия 4: Потоковая обработка (Streaming)

Когда данные генерируются в реальном времени (kafka, logs, events):

# Пример с потоком событий
from collections import defaultdict
import json

user_stats = defaultdict(lambda: {'count': 0, 'revenue': 0})

with open('events_stream.jsonl') as f:
    for line in f:  # Читаем по одной строке
        event = json.loads(line)
        user_id = event['user_id']
        user_stats[user_id]['count'] += 1
        user_stats[user_id]['revenue'] += event.get('amount', 0)

# Результат в памяти только статистика, не все события
print(user_stats)

Стратегия 5: Aggregation push-down (агрегация на лету)

Делай агрегацию максимально рано:

# НЕПРАВИЛЬНО: загружаешь все деньги в память
df = pd.read_csv('billion_rows.csv')
result = df.groupby('product').apply(lambda x: x[x['price'] > 100].shape[0])

# ПРАВИЛЬНО: агрегация на уровне БД
SELECT 
  product,
  COUNT(*) as expensive_items
FROM products
WHERE price > 100
GROUP BY product;

Практический пример: Анализ логов в 50 GB

import duckdb
from datetime import datetime, timedelta

# Задача: найти топ 10 пользователей по количеству ошибок за последние 7 дней

query = """
WITH logs AS (
  SELECT 
    user_id,
    log_level,
    timestamp,
    message
  FROM 'logs/**/*.jsonl'  -- DuckDB читает все файлы по паттерну
  WHERE 
    log_level = 'ERROR'
    AND timestamp >= CURRENT_DATE - INTERVAL 7 DAY
)
SELECT 
  user_id,
  COUNT(*) as error_count,
  COUNT(DISTINCT EXTRACT(DATE FROM timestamp)) as days_with_errors
FROM logs
GROUP BY user_id
ORDER BY error_count DESC
LIMIT 10
"""

result = duckdb.sql(query).df()
print(result)

Когда использовать какой подход

Лучшие практики

✅ DO:

• Фильтруй данные как можно раньше
• Группируй в БД, не в Python
• Используй индексы для быстрого поиска
• Читай по частям если нужно Python
• Пушь обработку в хранилище

❌ DON'T:

• Не загружай все в памяти без нужды
• Не обрабатывай в Python то, что может сделать БД
• Не игнорируй выстраивание индексов
• Не забывай про I/O лимиты

Заключение

Главное правило: Обрабатывай данные там, где они хранятся. SQL базы, DuckDB, Spark — все созданы для работы с большими объемами. Python нужен только для финального анализа и визуализации.