Prefect или Dagster: что выбрать для конвейера данных

Prefect или Dagster: что выбрать для конвейера данных — вопрос про единицу работы. В Prefect оркеструется поток функций: flow и его tasks. В Dagster оркеструется набор данных: assets, jobs и ops. Выбор идёт от того, что команде важнее сопровождать — вызовы кода или таблицы и файлы с их происхождением.

Поток функций и набор данных как разные единицы оркестрации

В модели набора данных единицей становится материализованный объект. Документация Dagster описывает asset как определение, которое включает AssetKey, набор ключей вышестоящих assets и Python-функцию, вычисляющую содержимое из зависимостей и сохраняющую результат. Читатель видит не «шаг прошёл», а «таблица за такую-то версию кода существует».

Flows/tasks Prefect и assets/jobs/ops Dagster

В Prefect рабочая единица — flow. Документация проекта перечисляет его свойства: метаданные о каждом состоянии отслеживаются автоматически, входные аргументы проходят проверку типов как параметры workflow, при сбое выполняются повторы с настраиваемой задержкой и лимитом, задаются таймауты против затянувшихся запусков, а flow можно развернуть и получить API для удалённого взаимодействия. Flow может вызывать другие flows и tasks, такие запуски называются дочерними.

Отдельно документация Prefect предупреждает о цене крупного flow: если весь код лежит в одной функции потока и падает любая строка, падает весь flow и его нужно повторять с начала. Отсюда рабочая привычка — дробить работу на tasks, чтобы повтор касался меньшего объёма.

В Dagster рабочая единица — asset. Документация проекта описывает самый частый способ создания: аннотирование Python-функции декоратором asset, где функция вычисляет содержимое, например таблицу базы или файл. Для случаев, когда один asset собирается несколькими операциями, применяется декоратор graph_asset: он инкапсулирует серию операций и показывает наружу один итоговый объект. При определении можно передать первым параметром context, и Dagster предоставит объект с доступом к системной информации вроде логгеров и идентификатора запуска.

Job задаёт запуск, ops описывают операции. Для приблизительного сопоставления flow соответствует job, task — op. Asset дополнительно объявляет ожидаемый результат данных.

Один учебный CSV-конвейер в двух представлениях

Возьмём условный вход: файл orders.csv со строками A,10,paid; B,7,cancelled; C,5,paid. Конвейер extract → clean → sum должен дать итог 15. Это учебная схема пилота, продукты в ней не запускались; числа служат проверкой логики.

В представлении Prefect конвейер описывается как flow с тремя задачами. Задача extract читает файл, задача clean отбрасывает отменённые строки, задача sum складывает суммы. Повтор настраивается на уровне задач, поэтому падение clean не требует повторять extract.

from prefect import flow, task

@task(retries=2, retry_delay_seconds=30)
def extract(path: str) -> list[tuple[str, int, str]]:
    rows = []
    with open(path, encoding="utf-8") as fh:
        for line in fh:
            name, amount, status = line.strip().split(",")
            rows.append((name, int(amount), status))
    return rows

@task(retries=2, retry_delay_seconds=30)
def clean(rows: list[tuple[str, int, str]]) -> list[tuple[str, int]]:
    return [(name, amount) for name, amount, status in rows if status == "paid"]

@task
def total(rows: list[tuple[str, int]]) -> int:
    return sum(amount for _, amount in rows)

@flow
def orders_pipeline(path: str = "orders.csv") -> int:
    return total(clean(extract(path)))

if __name__ == "__main__":
    print(orders_pipeline())

В представлении Dagster тот же конвейер описывается как цепочка данных raw_orders → paid_orders → daily_total. Каждый asset объявляет ключ и зависимости, а функция отвечает за вычисление и сохранение содержимого.

import dagster as dg

@dg.asset
def raw_orders() -> list[tuple[str, int, str]]:
    rows = []
    with open("orders.csv", encoding="utf-8") as fh:
        for line in fh:
            name, amount, status = line.strip().split(",")
            rows.append((name, int(amount), status))
    return rows

@dg.asset
def paid_orders(raw_orders: list[tuple[str, int, str]]) -> list[tuple[str, int]]:
    return [(name, amount) for name, amount, status in raw_orders if status == "paid"]

@dg.asset
def daily_total(paid_orders: list[tuple[str, int]]) -> int:
    return sum(amount for _, amount in paid_orders)

Сбой clean: что повторить и какие результаты считать устаревшими

Учебный сбой: получен raw_orders версии v2, шаг clean упал, а daily_total остался от версии v1. Полная таблица состояний выглядит так: extract — success/v2, clean — failed, sum — not run, daily_total — stale/v1. Это условные статусы учебного примера. Реальный запуск здесь не выполнялся.

Первый шаг разбора — различить состояние отказа и состояние свежести. Шаг clean находится в состоянии failed: он не завершился. Результат daily_total находится в состоянии stale: он существует, но построен по предыдущей версии входа. Смешивать эти два состояния в одном статусе нельзя, иначе команда теряет понимание, что именно требует действия.

Второй шаг — определить объём повтора. Повторять нужно clean и следующий за ним sum. Успешную выгрузку extract версии v2 повторять без причины не следует: её результат уже получен и относится к текущему входу. Такой порядок экономит время и снижает нагрузку на источник.

Третий шаг — зафиксировать правило актуальности. Итог версии v1 нельзя помечать актуальным для версии v2. Пока clean и sum не пересчитаны, daily_total остаётся устаревшим, и любой потребитель, читающий эту таблицу, получает данные предыдущей версии. Явная пометка устаревания защищает от тихой ошибки в отчётах.

Четвёртый шаг — проверить, что механизм повторов настроен на нужном уровне. В Prefect задачи поддерживают повторы. В примере Dagster из документации RetryPolicy задана отдельной операции внутри graph_asset. В нашем сценарии после исправления clean пересчитывают зависимый итог; фактическую свежесть входа и результата проверяют отдельно. Метки success/v2 и stale/v1 в таблице служат условными обозначениями учебного разбора и не описывают автоматически полученные состояния интерфейса.

Lineage, partitions и materialization для сопровождения данных

Сопровождение данных опирается на три механизма, и они по-разному выражены в двух продуктах. Lineage отвечает на вопрос, откуда взялся результат. Partitions отвечают на вопрос, к какому срезу данных относится результат. Materialization отвечает на вопрос, когда и по какой версии кода результат построен.

Для проверки partitions на пилоте сначала определите, какую часть данных нужно пересчитывать независимо, какой вход ей соответствует и какие результаты от неё зависят. Затем проверьте, можно ли выбрать только эту часть, пересчитать её зависимости и отличить её результат от остальных. В варианте Prefect проверьте, как срез передаётся параметром flow и виден в истории запусков. В варианте Dagster проверьте, как выбранный срез связывается с материализациями assets. Это список проверок для вашего конвейера; приведённый CSV-пример не демонстрирует настройку API партиций.

В Dagster эти механизмы встроены в модель asset. Документация проекта указывает, что при материализации asset с версией кода сгенерированная запись материализации помечается этой версией, а интерфейс показывает, когда у asset версия кода отличается от версии, использованной при последней материализации. Это прямой ответ на вопрос о свежести: расхождение версий видно без ручного сравнения.

В Prefect происхождение собирается вокруг запусков. Документация проекта сообщает, что мониторинг запусков, логирование и отслеживание состояний идут в реальном времени, а графы зависимостей и DAG отображаются автоматически после запуска flow. Происхождение здесь читается через историю запусков и их связи, включая дочерние запуски flows и tasks.

Для сопровождения выберите, что проверять первым: актуальность набора данных либо состояние запуска. Затем зафиксируйте правило актуальности результата.

Control plane, worker и место исполнения кода

Место исполнения кода определяет, где живут зависимости и данные. Документация Prefect описывает путь от локального запуска к развёртыванию: flows стартуют локально для разработки, затем разворачиваются где угодно — от одного процесса до контейнеров, Kubernetes или облачных сервисов, без привязки к поставщику, а инфраструктура задаётся кодом. Это означает, что код конвейера остаётся вашим артефактом, а среда исполнения описывается программно.

Control plane отвечает за расписания, состояния и наблюдаемость. Документация Prefect перечисляет триггеры: расписания, внешние события и вызовы через API; паузы для вмешательства человека или подтверждения; связывание flows по состояниям, условиям и произвольной логике. Worker выполняет запуски в выбранной среде и получает задания от управляющего слоя.

Для Dagster место исполнения связано с материализацией assets: код вычисляет содержимое и сохраняет результат, а система фиксирует факт материализации и её версию. Расписания и запуски задаются на уровне jobs, а зависимости данных определяют порядок построения.

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

Матрица выбора по потребностям команды и границы простого cron

Границы простого cron стоит обозначить отдельно. Cron справляется с запуском по расписанию, когда конвейер линеен, сбой виден по коду возврата, а повтор безопасен целиком. Как только появляются зависимости между шагами, частичный повтор, различение устаревших результатов и происхождение данных, cron перестаёт давать нужную картину, и команда переходит к оркестратору.

Про приобретение Dagster компанией Prefect: страница Prefect описывает продукты как построенные вокруг дополняющих сильных сторон — декларативная модель Dagster определяет, какие данные и результаты должны существовать, а гибкий движок исполнения Prefect удерживает сложные workflow в продакшене. Там же сказано, что конкретные продуктовые изменения не объявляются, а продукт Prefect продолжает получать ту же поддержку. Миграция клиентов Prefect в рамках этого объявления не запрашивается. Это сведения о статусе продуктов, и они не означают тождества инструментов.

Частые вопросы

Prefect или Dagster: что выбрать для конвейера данных, если команда небольшая?

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

Чем flows и tasks отличаются от assets, jobs и ops?

Flows и tasks описывают выполняемую работу: flow — декорированная Python-функция с отслеживанием состояний, tasks — её шаги. Assets описывают ожидаемые данные: ключ, зависимости и функцию вычисления содержимого. Jobs и ops задают уровни запуска и операций. Различие в том, что вы объявляете первичным — вызов кода или существование данных.

Что делать, если шаг clean упал, а итог остался от прошлой версии?

Различите два состояния: clean находится в состоянии failed, а итог — в состоянии stale. Повторите clean и выполните sum, успешную выгрузку текущей версии повторять без причины не нужно. Пока пересчёт не завершён, помечайте итог устаревшим, чтобы потребители не читали данные предыдущей версии как актуальные.

Как в этих продуктах устроены повторы и расписания?

В Prefect повторы настраиваются на уровне задач с задержкой и лимитом, а запуски идут по расписаниям, внешним событиям или через API. В Dagster повтор связан с материализацией asset, а расписания задаются на уровне jobs. Проверьте на пилоте, какой объём работы перезапускается при сбое в вашем конвейере.

Когда хватает обычного cron?

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