Сценарий: персистить промежуточные данные в parquet — best practices

Условие

В аналитическом пайплайне «сырьё → очистка → агрегаты → витрина». На каждом шаге сохраняем промежуточное. Почему parquet, а не csv? Как правильно партиционировать?

Решение

Почему parquet

Базовое сохранение

df.to_parquet(
    'clean/events.parquet',
    compression='snappy',     # 'zstd' даёт ещё меньше при чуть большем CPU
    engine='pyarrow',
    index=False,
)

Партиционирование

Полезно, когда чтение всегда фильтрует по одному-двум столбцам:

df.to_parquet(
    'clean/events/',
    partition_cols=['dt', 'country'],
    engine='pyarrow',
)
# Создаст дерево: clean/events/dt=2024-01-01/country=RU/part-0.parquet

При чтении:

df = pd.read_parquet(
    'clean/events/',
    filters=[('dt', '>=', '2024-01-01'), ('country', '==', 'RU')],
)

Правила партиционирования

1. Кардинальность ≤ ~10k — иначе тысячи маленьких файлов убьют чтение.
2. Партиционируйте по тому, по чему всегда фильтруют (dt, country).
3. Целевой размер файла — 128–512 МБ. Меньше — overhead; больше — плохо для параллельного чтения.
4. Один файл — одна партиция. repartition перед записью.

Подводные камни

1. partition_cols=['user_id'] при 10M уникальных user_id создаст 10M файлов и убьёт файловую систему.
2. NaN в partition column превращаются в строку __HIVE_DEFAULT_PARTITION__ — потом фильтр не находит.
3. Smashing мелких parquet — раз в день запускать compaction (объединение мелких файлов в крупные).
4. pyarrow vs fastparquet — pyarrow надёжнее, fastparquet чуть быстрее на простых типах. Использовать pyarrow.
5. Schema evolution — добавил колонку в новый файл, читаешь старые и новые вместе → pyarrow умеет, но fastparquet иногда падает.

Эталонный ответ

Parquet (snappy/zstd) с партиционированием по тому, по чему фильтруют (дата, страна). Партиции с кардинальностью ≤10k, размер файла 128–512 МБ. Не партиционируйте по user_id.