Пайплайн данных приносит пользу, когда его этапы сбалансированы

Аналитик просит добавить в отчёт разрез по городам, и выясняется, что нужного поля нет ни в одной таблице. Разработчик идёт в источник, находит данные, пишет запрос и добавляет колонку. Через неделю выясняется, что поле заполнено мусором: часть городов записана латиницей, часть с сокращениями, а часть отсутствует вовсе. Отчёт готов, но доверять ему нельзя. Знакомая история, и она объясняет, почему тема пайплайнов так быстро переходит из технической плоскости в организационную.

Пайплайн строится по принципу сборочной линии и живёт за счёт параллельности

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

Полезно посчитать на небольшом примере. Пусть сборка состоит из трёх операций длительностью двадцать, десять и пятнадцать минут. Если всё делает один мастер, машина выходит раз в сорок пять минут. Если операции разнесены по трём постам, первая машина всё равно выйдет через сорок пять минут, зато каждая следующая будет появляться через двадцать. Общая задержка не сократилась, а пропускная способность выросла более чем вдвое.

Отсюда главный вывод для проектирования: пропускная способность системы не может превышать пропускную способность самого медленного этапа. Ускорять остальные бесполезно, пока узкое место не расширено.

В обработке данных это правило работает буквально. Если выгрузка занимает десять минут, а преобразование час, то сокращение выгрузки вдвое не даст почти ничего. Наращивать нужно преобразование.

Извлечение требует проверки данных на входе

Первый этап это извлечение. Данные забираются из одного или нескольких источников, и от качества этого шага зависит всё остальное.

Источники бывают разными. Реляционные базы, файлы, документы в формате обмена, ответы внешних служб, журналы веб-сервера. Каждый имеет свою организацию данных, свои типы и свои особенности.

Первое, что стоит сделать на входе, это проверка значений. Извлечённые данные сверяются с ожидаемым набором: соответствуют ли типы, попадают ли значения в допустимый диапазон, нет ли пустых обязательных полей.

Второе это обработка отклонений. Часть записей неизбежно окажется некорректной. Их нужно либо отклонить целиком, либо отбросить испорченные поля и продолжить. Молчаливое проглатывание ошибок опаснее отказа: оно приводит к тихому искажению отчётов.

Третье это возможность повторить извлечение. Сбой сети или падение источника не должен требовать ручного вмешательства. Процесс запускается заново с той же точки и даёт тот же результат.

import csv
from datetime import datetime

REQUIRED = ["order_id", "customer_id", "amount", "created_at"]

def read_orders(path):
    good, bad = [], []
    with open(path, encoding="utf-8", newline="") as handle:
        for row in csv.DictReader(handle):
            missing = [field for field in REQUIRED if not row.get(field)]
            if missing:
                bad.append({"row": row, "reason": "нет полей: " + ", ".join(missing)})
                continue
            try:
                row["amount"] = float(row["amount"])
                row["created_at"] = datetime.fromisoformat(row["created_at"])
            except ValueError as error:
                bad.append({"row": row, "reason": str(error)})
                continue
            good.append(row)
    return good, bad

Здесь важно, что отклонённые записи не выбрасываются. Их складывают отдельно, потому что разбор причин отказа часто вскрывает проблему на стороне источника.

Преобразование очищает, приводит к единому виду и обогащает

Второй этап самый объёмный. На входе сырые данные, на выходе подготовленные к анализу.

Первая задача это очистка. Удаляются дубликаты, отбрасываются записи с пустыми ключевыми полями, приводятся к единому виду справочные значения. Если в источнике пол обозначен цифрой один и два, а в хранилище ожидаются буквы, преобразование выполняет перевод.

Вторая задача это приведение типов и форматов. Даты в разных системах записываются по-разному, суммы хранятся то в копейках, то в рублях, то в виде строки с разделителями. Всё это приводится к одному виду.

Третья задача это вывод новых значений. Сумма продажи считается как произведение количества на цену, длительность доставки как разница между датой заказа и датой получения. Производные поля избавляют аналитика от повторяющихся вычислений.

Четвёртая задача это обогащение. К заказу подтягивается регион клиента, к товару категория, к сотруднику подразделение. Данные собираются из нескольких источников в одну широкую таблицу.

CITY_ALIASES = {
    "мск": "Москва",
    "moscow": "Москва",
    "спб": "Санкт-Петербург",
    "saint petersburg": "Санкт-Петербург",
    "екб": "Екатеринбург",
}

def normalize_city(raw):
    if not raw:
        return None
    key = raw.strip().lower().replace(".", "")
    if key in CITY_ALIASES:
        return CITY_ALIASES[key]
    return raw.strip().title()

def enrich(order, customers):
    customer = customers.get(order["customer_id"])
    if customer is None:
        order["region"] = None
        order["city"] = None
        return order
    order["region"] = customer["region"]
    order["city"] = normalize_city(customer.get("city"))
    return order

Словарь синонимов решает конкретную боль. Без него отчёт по городам покажет десяток вариантов написания одного и того же города, и группировка потеряет смысл.

Пятая задача это проверка бизнес-правил. Сумма не может быть отрицательной, дата получения не может быть раньше даты заказа, количество не может быть нулевым. Записи, нарушающие правила, отправляются на разбор.

Загрузка и её порядок определяют, когда появятся данные

Третий этап это запись результата в целевое хранилище. Способов два, и разница между ними принципиальна.

Первый способ это полная замена. Таблица очищается и заполняется заново. Просто и надёжно, но при больших объёмах занимает часы и требует места.

Второй способ это добавочная загрузка. Новые записи дописываются к существующим. Быстро, но требует механизма определения того, что уже загружено, иначе появятся дубликаты.

Третий способ это обновление и вставка. Изменённые записи перезаписываются, новые добавляются. Работает по ключу и требует, чтобы ключ был стабильным.

Отдельный вариант порядка операций называется обратным. Данные сначала загружаются в хранилище в исходном виде, а преобразование выполняется уже там средствами самого хранилища. Такой подход стал популярен с появлением облачных аналитических баз, где вычислительные ресурсы дешевле и масштабируются автоматически.

Выбор между прямым и обратным порядком определяется несколькими факторами. Если преобразования сложные и требуют внешних библиотек, их удобнее делать до загрузки. Если данных много, а хранилище мощное, выгоднее загрузить как есть и считать на месте.

Надёжность обеспечивается повторяемостью и изоляцией запусков

Пайплайн, который нельзя перезапустить без последствий, рано или поздно создаст проблему.

Первое требование это идемпотентность. Повторный запуск того же шага не должен менять результат. Достигается это либо полной заменой раздела, либо проверкой ключа перед вставкой.

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

Третье требование это изоляция от читателей. Аналитик не должен видеть наполовину загруженную таблицу. Решается либо отложенным переключением указателя на готовый раздел, либо использованием промежуточной таблицы и последующего переименования.

def load_partition(rows, period, staging, target):
    staging.replace_partition(period, rows)
    if staging.validate(period, expected_min_rows=1):
        target.swap_partition(period, staging.name)
        return {"period": period, "rows": len(rows), "status": "готово"}
    raise RuntimeError("Проверка раздела не пройдена: " + period)

Четвёртое требование это журналирование. Каждый запуск оставляет след: когда начался, сколько записей обработал, сколько отклонил, чем закончился. Без журнала разбор сбоя превращается в археологию.

Пятое требование это оповещение. Если шаг упал или обработал подозрительно мало записей, кто-то должен узнать об этом до того, как руководитель увидит пустой отчёт.

Расписание и зависимости избавляют от гонок

Пайплайн редко состоит из одного шага. Обычно это граф, где один шаг зависит от другого.

Первая ошибка в организации это запуск по времени без проверки готовности источников. Задача стартует в три часа ночи, а выгрузка из источника ещё идёт. Результат содержит половину данных, и никто этого не замечает.

Правильное решение состоит в отслеживании зависимостей. Шаг запускается не по часам, а после того, как предыдущий успешно завершился.

def run_dag(tasks, completed=None):
    completed = set(completed or [])
    while len(completed) < len(tasks):
        ready = [
            name for name, task in tasks.items()
            if name not in completed and set(task["needs"]) <= completed
        ]
        if not ready:
            raise RuntimeError("Обнаружен цикл в зависимостях")
        for name in ready:
            tasks[name]["run"]()
            completed.add(name)
    return completed

Проверка на цикл здесь не формальность. Ошибка в описании зависимостей приводит к тому, что планировщик зависает навсегда, а разбирательство занимает часы.

Вторая ошибка это слишком частый запуск. Задача, выполняющаяся двадцать минут и запускаемая каждые пять, создаёт очередь и конкуренцию за ресурсы. Интервал должен быть больше времени выполнения с запасом.

Третья ошибка это отсутствие приоритетов. Срочный пересчёт отчёта для руководства ждёт в общей очереди за плановой задачей. Разделение на очереди решает проблему.

Схема данных требует версионирования и осторожных изменений

Данные живут дольше кода, и изменения в структуре больно бьют по потребителям.

Первое правило касается добавления полей. Новое поле безопасно, если оно необязательное и заполнено значением по умолчанию. Потребители, которые о нём не знают, продолжат работать.

Второе правило касается удаления. Удалять поля сразу нельзя. Сначала поле помечается устаревшим, потом проверяется, кто его использует, и только через время удаляется.

Третье правило связано с изменением типа. Расширение типа без потери данных проходит спокойно, сужение ломает записи.

Четвёртое правило касается переименования. Переименование эквивалентно удалению и добавлению, поэтому выполняется в два этапа с периодом сосуществования.

Пятое правило это документирование. У каждой таблицы и колонки должно быть описание, назначение и владелец. Без этого через год никто не вспомнит, что означает поле с сокращённым названием.

Качество данных проверяется регулярно, а не по жалобам

Отчёты портятся не только сбоями, но и постепенным накоплением ошибок. Поэтому нужны постоянные проверки.

Первая проверка это полнота. Сравнивается число записей в источнике и в хранилище за период. Расхождение больше доли процента требует разбора.

Вторая проверка это уникальность. Ключ не должен повторяться. Появление дубликатов означает ошибку в логике загрузки.

Третья проверка это допустимость значений. Суммы в разумных пределах, даты не в будущем, обязательные поля заполнены.

Четвёртая проверка это согласованность связей. Каждый заказ ссылается на существующего клиента, каждая позиция на существующий товар.

Пятая проверка это своевременность. Данные за вчера должны появиться к оговорённому времени. Задержка сама по себе является инцидентом.

CHECKS = [
    ("полнота", lambda src, dst: abs(src - dst) / max(src, 1) < 0.001),
    ("уникальность", lambda _, duplicates: duplicates == 0),
    ("допустимость", lambda _, invalid: invalid == 0),
]

def run_checks(metrics, source_rows, target_rows):
    results = []
    for name, rule in CHECKS:
        results.append((name, bool(rule(source_rows, metrics.get(name, 0)))))
    return results

Такой набор проверок встраивается в конец пайплайна и не пропускает плохие данные дальше. Отчёт, построенный на непроверенных данных, опаснее отсутствия отчёта: он создаёт ложную уверенность.

Пайплайн данных получился конструкцией, в которой техническая часть предсказуема, а вся сложность сосредоточена в деталях. Сбалансированные этапы, повторяемые запуски, версионируемая схема и постоянные проверки качества. Ни один из этих пунктов не выглядит прорывом, зато вместе они дают то, ради чего всё затевалось: цифру, на которую можно опереться в решении.