dbt Cloud и Airflow: от загрузки PostgreSQL до обновления Power BI

Обновлено 2 сентября 2026 года.

Airflow нужен рядом с dbt Cloud, когда преобразования зависят от внешней загрузки, других систем или общего расписания. Если весь процесс ограничен одним dbt-проектом, сначала оцените собственный планировщик платформы: два оркестратора не обязательны. В сложной цепочке назначьте один компонент ответственным за запуск, повторы и итоговое состояние.

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

Практический маршрут: подтвердить завершение загрузки → запустить dbt → дождаться результата → проверить бизнес-показатели → обновить модель Power BI Import → проверить результат обновления. Ответ API «задание принято» не означает, что данные уже готовы.

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

Оператор dbt Cloud

В провайдере Airflow для dbt Cloud есть DbtCloudRunJobOperator. Он может запустить job и ждать завершения. При асинхронном запуске используется DbtCloudJobRunSensor с идентификатором полученного run. ExternalTaskSensor предназначен для задач Airflow, а не для произвольного задания dbt Cloud.

Пример фрагмента внутри уже настроенного DAG:

from airflow.providers.dbt.cloud.operators.dbt import DbtCloudRunJobOperator

run_models = DbtCloudRunJobOperator(
    task_id="run_models",
    dbt_cloud_conn_id="dbt_cloud_default",
    job_id=12345,  # замените ID реального задания
    wait_for_termination=True,
    check_interval=60,
    timeout=3600,
)

Предварительно установите совместимую версию провайдера, создайте Connection с реквизитами нужного аккаунта и настройте само задание. Значения таймаута и периода опроса в примере требуют настройки под ваш проект. Фрагмент не устанавливает Airflow, не загружает источники и не создаёт dbt-проект.

CI и рабочий запуск — разные проверки

В CI проверяйте изменения моделей на отдельной схеме и подходящих тестовых данных. Использование state:modified+ требует корректного состояния сравнения. Чтобы неудачный CI блокировал слияние, настройте обязательную проверку в репозитории: само существование job этого не обеспечивает.

В рабочем запуске выполняйте предусмотренные проектом тесты данных. dbt build объединяет построение и проверки выбранных ресурсов. Прошедший CI не доказывает, что свежая выгрузка CRM полна. Сбой проверки бизнес-данных обычно требует выяснить причину; три повтора одного и того же неправильного набора её не устранят.

Модель и публикация витрины

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

У инкрементальных моделей продумайте пустую таблицу, позднее изменение и удаление записи. Фильтр updated_at >= max(updated_at) при пустой цели с NULL может не загрузить ни одной строки. Стабильный ключ нужен для обновлений, но не заменяет правила удаления.

Права на будущие объекты PostgreSQL назначаются для роли, которая их создаёт. Проверьте USAGE на схему и SELECT для BI после реального запуска dbt. Если неудачный тест уже оставил изменённые таблицы, один запрет обновления Power BI не защищает других читателей: при необходимости стройте проверяемый набор отдельно и публикуйте его после контроля.

Последний шаг — Power BI

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

Приёмка включает не только успешный запуск, но и отказ dbt-теста, таймаут API, повтор DAG и сбой обновления Power BI. Пользователь должен видеть либо подтверждённо свежие показатели, либо понятную дату последнего успешного среза.