--- title: Airflow DAG-и description: Шість DAG-ів у dags/, ланцюг import → scorecard → test-status на assets, параметри запуску, секрети й типові пастки KubernetesPodOperator. --- Уся фонова автоматизація Kumquat живе в `dags/` — шість DAG-ів на Airflow 3.2.0. Жоден із них не робить роботи сам: кожен таск запускає под у неймспейсі `airflow` через `KubernetesPodOperator`, а логіка лежить в образі CLI-пакета (`backend/packages/kumquat-*`) або в SQL-скрипті. DAG-файл — це лише розклад, залежності й параметри. Джерело правди — самі файли `dags/*.py` і `dags/assets.py`; конвенції написання нових DAG-ів — у скілі `.claude/skills/airflow-k8s-dags/`. Хто потім читає наповнену таблицю — [Аналітика креативів](/backend/creative-analytics). ## Що є в `dags/` | DAG | Розклад | Ентрипойнт пода | Що робить | |---|---|---|---| | `dag_kumquat_import_creative_performance.py` | `20 4,12,16 * * *` | `clickhouse-client` (образ `clickhouse/clickhouse-server:24.8`) + `mysql` (`mysql:8.4`) | Дописує в `top_creo_for_tableau` дизайнера, маркетолога й дату створення з kumquat, потім дзеркалить останні 30 днів у ClickHouse `creative_performance`; окремою таскою дзеркалить `tt_creatives_data_country` у `tiktok_creative_performance` — чотири таски на продукт | | `dag_kumquat_creative_metrics.py` | asset `META_ADS_DAILY` | `kumquat-creative-metrics run` | Рахує скоркард креативів і `$set`-ить його в `creatives.metrics` | | `dag_kumquat_ch_sync_creatives.py` | `@hourly` | `kumquat-ch-sync --mode batch` | Інкрементальний догон MongoDB `creatives` → ClickHouse `creatives` | | `dag_kumquat_retention_purge.py` | `@daily` | `kumquat-retention run` | Видаляє soft-deleted паки/креативи і ресурси, якими вони володіють (Drive, S3, Qdrant, ClickHouse) | | `dag_kumquat_archive_done_tasks.py` | `0 6 * * 1` | `kumquat-cli archive-done-tasks` | Щопонеділка переносить у Archived задачі з Done, які `in_test` або `manual_launch` | | `dag_kumquat_sync_test_status.py` | asset `CREATIVE_METRICS_READY` | `kumquat-cli sync-test-status` | Тримає `test_status` креатива в згоді з відкруткою: `in_test` → `paused` і назад | | `dag_kumquat_import_solid_funnels.py` | `0 5 * * *` | `web-funnel-analytics import-solid-funnels` | Збирає ClickHouse `solid_funnels_kumquat` за останні 5 днів (рядки покупок — за `trial_type + 28` днів): спенд з `solid_funnels_kumquat` чотирьох продуктових баз + кроки воронки (`funnel_opens`, `second_screens`, `paywalls`, `checkouts`) з `funnel_prod`, `ad_id` → `creo_name` через `solid_funnels_kumquat_ad_info`, сума по ключу, `funnel_id` кожному рядку за назвою воронки з `funnel_prod.sessions_data` | Образи задаються константою `IMAGE` у кожному файлі. `creative_metrics` іде на спільному backend-образі (`ml-aps/kumquat/backend:backend_latest` в ECR) — CLI скоркарда лежить у ньому як консольний скрипт; import-DAG бере публічний образ ClickHouse і не має власного коду взагалі. Перші три зв'язані в один ланцюг, решта дві незалежні: ``` creo-marketologist (зовнішній) MySQL {app}.top_creo_for_tableau 03:20 / 11:20 / 15:20 UTC {app}.tt_creatives_data_country │ │ +1 година ▼ import_creative_performance (4 паралельні гілки) push_dim → apply_dim → import import_tiktok (поруч, не в ланцюжку) │ ▼ publish ──► META_ADS_DAILY extra["max_date"] │ ▼ creative_metrics (resolve_watermark → scorecard) │ ▼ CREATIVE_METRICS_READY │ ▼ sync_test_status (in_test → paused) ``` ### Чому TikTok — у цьому ж DAG-у, а не в окремому `tt_creatives_data_country` — власний звіт TikTok, і він єдиний міряє саме відео: старти, 6-секундні перегляди й реакції. Атрибуційний фід цього не має взагалі, тож без нього TikTok-креатив не має ні hook, ні hold. Таска не потребує збагачення (`kumquat_creative_dim` до неї не стосується), тож вона стоїть **поруч** із ланцюжком `push_dim → apply_dim → import`, а не після нього. Але вона так само **гейтить `publish`**, і саме тому живе тут, а не в окремому DAG-у: скоркард читає обидві таблиці **одним запитом**, тож «вікно залите» мусить означати обидві — інакше він порахує hook по половині даних і запише це в Mongo як факт. Асет лишився один і з тією самою назвою (`…/creative_performance`). Перейменувати його — означало б тихо відключити скоркард: Airflow не піднімає помилку на неіснуючий асет, консюмер просто перестає тригеритись. Ще дві речі, які ця таска робить і які варто знати: - **`creo_name` виводиться на імпорті, а не на читанні.** TikTok називає оголошення файлом (`1-4023_CNI_VARAI_VID_9x16_TIK_US_EN.mp4_tops`), тож ключ — це `ad_name` до першої крапки. Робиться це `substringIndex(…, '.', 1)` навмисно замість розбору сегментів: у базі одночасно живуть дві схеми неймінгу, і регулярка під будь-яку з них тихо загубила б половину карток. - **Після імпорту таска друкує діагностику**, не гейт: розподіл валют (нічого не конвертується) і відсоток `creo_name`, що звівся на картку. Зміна неймінгу в кабінеті видно тут як просідання матчу — задовго до того, як хтось помітить картку без hook rate. ## Чому import-DAG ще й пише в MySQL Колонки `designer`, `creo_marketologist` і `date_creative` в `top_creo_for_tableau` історично заповнював апстрім-пайплайн крео-маркетолога, **парсячи їх із назви креатива**. Неймінг, який kumquat генерує зараз, таких сегментів не має, тож для кожного нового креатива ці колонки порожні — і старі дашборди Tableau, що ріжуть по дизайнеру чи маркетологу, поступово сліпнуть. Заповнює їх тепер kumquat, двома тасками на продукт перед імпортом: - **`push_dim_{product}`** (образ ClickHouse) — `INSERT INTO FUNCTION mysql(...)` з `creatives_current` у MySQL-таблицю `kumquat_creative_dim` (її DDL — `db/mysql-init.sql`, по копії на кожну з чотирьох БД). Повне оновлення вимірної таблиці: `ON DUPLICATE KEY UPDATE` по `creo_name`. - **`apply_dim_{product}`** (образ `mysql:8.4`) — `UPDATE ... JOIN` вимірної таблиці на `top_creo_for_tableau` у тому ж 30-денному вікні, яке наступний таск переімпортовує. Порядок `push_dim → apply_dim → import` навмисний: ClickHouse отримує вже збагачені значення в тому самому рані, а не через цикл. Три речі, які тут легко зламати: - **`UPDATE` треба повторювати щоразу.** Апстрім видаляє й переливає те саме 30-денне вікно тричі на добу, тож збагачення не «залишається накладеним». - **`COALESCE(NULLIF(...))` — не косметика.** Легасі-картки теж є в kumquat і часто без `assignee`; без цього `UPDATE` затер би значення, яке апстрім розпарсив із назви коректно. `creo_name`, якого kumquat не знає, джойн не чіпає взагалі. - **Пишемо короткі коди, не імена.** Колонки `varchar(8)`, і легасі-рядки вже містять ініціали, тож у MySQL їде `users.initials` (через `creatives_current.assignee_code` / `task_owner_code`). Проставляє їх `kumquat-cli backfill-user-initials`; людина без коду лишає свій рядок без змін. `predesigner` не чіпаємо — відповідника в kumquat немає. Записи в MySQL вимагають прав: `CREATE`/`INSERT`/`UPDATE` на `kumquat_creative_dim`, `UPDATE` на `top_creo_for_tableau` і `GRANT WRITE ON MYSQL` для користувача ClickHouse (`db/init.sql`). Історію, старшу за 30 днів, вікно DAG-у не дістає — для неї є `db/backfill-creative-dim.sh`. ## Чому scorecard на asset-і, а не на крон-і Import DAG **видаляє й перезаливає** 30-денне вікно тричі на день. Скоркард на власному кроні рано чи пізно прочитав би джерело посеред перезаливки й записав би дірку в Mongo як факт. Asset-тригер робить порядок структурним, а не питанням розкладу. Ключове тут — таск `publish`, і те, що він **сам є сигналом повноти**: окремого валідаційного кроку немає. Він стоїть після всіх чотирьох імпортів, тож якщо хоч один упав — `publish` не виконується, asset не публікується, і скоркард просто не стартує. `extra["max_date"]` їде разом із подією, щоб консюмер не вгадував, по який день джерело повне. Це не зручність: **asset-тригерні рани в Airflow 3 не мають `logical_date`, `data_interval_start` і `data_interval_end`** — звернення до них кидає `KeyError`/`UndefinedError`. Ручний тригер їх *має*, тому код, що їх читає, працює всюди, крім продакшену. Через це `resolve_watermark` має явний ланцюг фолбеків (params → `max_date` з подій → `source_dag_run` → `dag_run.run_after`), а под отримує вже готову дату, а не шаблон. Події накопичуються між ранами (три імпорти, один скоркард) і приходять одним раном із багатьма подіями — тому береться `max` із `max_date`, а не перша подія. Обидва URI лежать у `dags/assets.py`, і обидві сторони мусять імпортувати звідти. URI, написаний двічі й потім відредагований з одного боку, нічого не ламає видимо: продюсер публікує далі, консюмер просто більше ніколи не тригериться, і єдиний симптом — скоркард, що тихо перестав оновлюватися. ## Параметри запуску Параметри мають два DAG-и (Trigger DAG w/ config) — `creative_metrics`: | Параметр | Тип | Дефолт | Що робить | |---|---|---|---| | `watermark` | `null`/`string`, `format: date` | `null` | `YYYY-MM-DD` в обхід asset-події; порожнє = резолвиться з події | | `media_sources` | `array` of `string` | `[]` | Скоуп рану, напр. `["fb", "tiktok"]`; порожнє = скоуп деплою | | `dry_run` | `boolean` | `false` | Тільки extract + transform, нічого не пишеться | `watermark` і `dry_run` йдуть у под флагами в `cmds`. `media_sources` — ні: под резолвить скоуп виключно з оточення (`CreativeMetricsSettings.media_sources`, тобто `KUMQUAT_CREATIVE_METRICS__MEDIA_SOURCES`), і в CLI опції для нього немає. Тому параметр інжектиться як `V1EnvVar` — саме та змінна, яку читає поле. Дві деталі, від яких залежить, чи под узагалі стартує: - **`| tojson` обов'язковий.** Дефолтний рендер списку в Jinja — Python-repr в одинарних лапках, а поле парситься тільки як JSON. `['fb', 'tiktok']` падає з `SettingsError` на старті пода, ще до першого запиту. - **Порожній параметр рендериться в порожній рядок, а не в `[]`.** pydantic-settings відкидає порожні значення (`env_ignore_empty=True`), тож залишається значення деплою. Жорсткий `[]` тут навпаки тихо розширював би скоуп назад на всі джерела при кожному ручному запуску. У Kubernetes `env` має пріоритет над `env_from` для того самого імені, тому явно заданий параметр перекриває секрет. Скоуп рану записується в `metrics.media_sources` і в `creatives__metrics_runs` (див. [Mongo-схема](/backend/db-schema)), тож із документів видно, з яким набором джерел він рахувався. І `sync_test_status`: | Параметр | Тип | Дефолт | Що робить | |---|---|---|---| | `idle_days` | `integer`, `minimum: 0` | `3` | Скільки днів без spend означає, що тест закінчився | | `resume_days` | `integer`, `minimum: 0` | `1` | Наскільки свіжим має бути spend, щоб картка повернулася в `in_test` | | `dry_run` | `boolean` | `false` | Друкує, кого перевів би, і нічого не пише | Усі три йдуть у под флагами в `arguments`. `dry_run` рендериться в `--dry-run`/`--no-dry-run`, а не в порожній рядок: порожній елемент argv доїхав би до typer як зайвий позиційний аргумент і поклав би под. `resume_days` більший за `idle_days` под **відхиляє** — див. мертву зону нижче. ## `sync_test_status`: два ребра, мертва зона і хто має право відкотити `test_status` креатива ставиться один раз — запуском (`mark_launched` стемпить `in_test`), — і далі його не рухає **ніщо**: `CreativeUpdate` має єдине поле `notes`, повторний звіт про запуск — no-op (`already_launched`), а bulk-запуск пака чіпає лише `not_tested` картки. Тож `in_test` поступово означає «колись запускали», і разом із ним пливуть усі фільтри галереї й колонки експорту, побудовані на ньому. Кабінет уже каже і коли тест закінчився, і коли почався знову — креатив перестає або знову починає коштувати грошей. Звідси рівно два ребра: | Було | Умова | Стало | |---|---|---| | `in_test` | немає spend понад `idle_days` (3) | `paused` | | `paused` | є spend у межах останніх `resume_days` (1) | `in_test` | `not_tested` і `tested` не чіпаються в жодному напрямку: перше — картка, яку не запускали, друге — вердикт людини. ### Дні рахуються не від `now()` Джерело — 30-денне вікно, яке апстрім перезаливає, а import-DAG дзеркалить тричі на день; `date` у ClickHouse відстає від сьогодні на день-два, і на скільки саме — плаває. Від `now()` кожен креатив набирав би день простою щоразу, коли імпорт застряг, а зламаний імпорт зрештою поставив би на паузу всю бібліотеку. Тому годинник — `max(date)` **того продукту, під яким креатив крутився**: у продукта, чий імпорт став, стає й годинник, і жодна його картка не рухається, поки дані не підуть знову. Тому й asset-тригер саме на `CREATIVE_METRICS_READY`, а не на `META_ADS_DAILY` і не на власному кроні: sweep стартує лише після скоркарда, який уже пройшов власний freshness-preflight, і ніколи — посеред перезаливки вікна, де дірка в даних читалася б як «перестав витрачати». Картку, яка не витратила **жодного** долара, sweep не чіпає: відсутність рядків у ClickHouse — це розрив синку або запуск, який ще не доїхав, а не відповідь кабінету. ### Мертва зона між ребрами `resume_days` навмисно менший за `idle_days`: проміжок між ними (2–3 дні на дефолтах) — зона, де не спрацьовує жодне ребро. Щоб повернутися, картка мусить крутитися **зараз**, а не просто мати колись spend усередині вікна паузи. Без цього креатив із нерівним польотом фліпав би статус щопрогону — тричі на добу, — а кожен фліп це audit-рядок, зміна кольору картки в галереї і зсув `updated_at`, тобто ще й перезалив рядка в ClickHouse через kumquat-ch-sync. Мертва зона заодно знешкоджує корекції даних: дорисований рядок за старе число нічого не воскрешає, бо resume міряє свіжість, а не факт появи рядка. `resume_days > idle_days` под відхиляє з `ValueError`: вище `idle_days` правила перетинаються, і той самий креатив був би одночасно кандидатом на паузу і на повернення. ### Відкотити можна лише власний запис Картка повертається в `in_test` тільки якщо **найсвіжіший audit-рядок, який змінював її `test_status`, — це рядок sweep-а** (екшен `creative.test_status_swept`, константа `SWEEP_TEST_STATUS_ACTION`). Запит — `AuditLogsRepository.last_action_touching`, один агрегат на весь шорт-ліст. Сьогодні іншого джерела паузи не існує взагалі, тож перевірка ні на що не впливає. Вона є заради дня, коли `test_status` додадуть у `CreativeUpdate`: ручний запис залишить рядок з іншим екшеном, і повернення для таких карток вимкнеться саме собою — ніхто не мусить пам'ятати про це правило. Маркер на документі (як `__search_text`) дав би зворотне: кожен новий writer був би зобов'язаний його чистити. ## Ідемпотентність `retries: 1` у `default_args` у всіх шести DAG-ах — це працює тільки тому, що кожен таск ідемпотентний, і в кожного це зроблено по-своєму: - **import** — delete + re-insert того самого 30-денного вікна, що й перезаливає апстрім. `SETTINGS insert_deduplicate = 0` тут обов'язковий: без нього повторний `INSERT` ідентичних даних після `DELETE` (retry або незмінений source) мовчки відкидається як дублікат блока, і вікно лишається порожнім. - **ch-sync** — цільова таблиця `ReplicatedReplacingMergeTree`, тож дублікати вікна перекриття й ретраї схлопуються самі; пропущені години не потребують нічого окремого. - **creative-metrics** — `$set` на `creatives.metrics` плюс sweep застарілих документів за `run_id`. - **retention-purge** — зовнішні ресурси знищуються **до** Mongo-документа, тож упалий ран лишає запис soft-deleted, і наступний переробляє все з нуля. Саме цей порядок робить `retries` безпечним на деструктивній джобі. `max_active_runs=1` теж не косметика: два одночасні пурджі порахували б набір «вижилих» кожен без знання про видалення іншого. - **archive-done-tasks** — заархівована задача більше не підпадає під селектор, тож повторний ран не має що робити. Кожна задача рухається тим самим `TasksService.transition_status`, що й з борду, а `expected=done` означає, що задачу, яку хтось встиг перевести самотужки, ран **пропускає**, а не переписує. - **sync-test-status** — картка, яку ран уже перевів, більше не підпадає під те саме ребро (мертва зона не дає їй одразу підпасти під протилежне). Обидва записи йдуть через `CreativesRepository.sweep_test_status`, guard якого вимагає, щоб картка все ще була в очікуваному статусі: статус, який людина встигла поставити сама, ран **звітує**, а не переписує. Окремий audit-екшен `creative.test_status_swept` лишає в історії слід, що картку перевів sweep, — і він же є тим, що дозволяє ранові відкотити власну паузу. ## ClickHouse: що DAG-и мусять враховувати ClickHouse — мультиреплічний кластер `ml` (один шард), і Service балансує **кожне з'єднання** на випадкову репліку. Для DAG-ів із цього випливають дві речі, які видно в коді import-DAG-а: - `ALTER TABLE ... DELETE` іде з `SETTINGS mutations_sync = 2` — інакше мутація чекає лише на репліку, що обслуговує це з'єднання. - `DELETE` і наступний `INSERT` виконуються **одним викликом `clickhouse-client --multiquery`**. Два окремі виклики — два з'єднання, і порядок операцій між репліками більше нічим не гарантований. Повний перелік обмежень кластера (DDL `ON CLUSTER 'ml'`, `ReplicatedMergeTree`, `GRANT READ ON MYSQL` для `mysql()`) — у `CLAUDE.md` в корені репозиторію. ## Секрети Усі поди беруть оточення з одного k8s-секрета через `env_from`: ```python APP_SECRETS = [ k8s.V1EnvFromSource(secret_ref=k8s.V1SecretEnvSource(name="kumquat-analytics-secrets")), ] ``` Сам секрет синкається External Secrets з AWS Secrets Manager (`ml/prod/kumquat/airflow/analytics-secrets`) — манифест `k8s/external-secrets.yaml`, застосовується як `kubectl apply -n airflow -f external-secrets.yaml`. У секреті **два написання тих самих значень**, і це навмисно. Історично там плоскі ключі (`CH_HOST`, `MONGO_URI`) — їх читають ch-sync і import-DAG. `kumquat-creative-metrics` читає загальнодеплойний контракт із подвійним підкреслювачем (`CLICKHOUSE__HOST`, `MONGO__URI`), бо `env_nested_delimiter` мапить його на вкладену секцію налаштувань. Обидві групи вказують на ті самі властивості в AWS. Одна пастка, яку легко «виправити» й зламати под: **`CLICKHOUSE__PORT` у секреті свідомо відсутній.** Ключ `CH_PORT` — це нативний порт (9000) для `clickhouse-client`, а `kumquat-creative-metrics` говорить HTTP і має працювати на дефолті 8123. Замапити `CH_PORT` на `CLICKHOUSE__PORT` — найкоротший шлях покласти скоркард. `web-funnel-analytics import-solid-funnels` (пакет `web-funnel-analytics`, свої налаштування, не `AppSettings`) читає ClickHouse зі спільних `CLICKHOUSE__*`, а MySQL — з `WEB_FUNNEL_ANALYTICS__ANALYTICS_MYSQL__*` (продуктові бази, ті самі AWS-властивості, що й плоскі `MYSQL_*`) і `WEB_FUNNEL_ANALYTICS__FUNNEL_MYSQL__*` (сервер `funnel_prod`, свої хост і юзер; властивості `FUNNEL_MYSQL_HOST/PORT/USER/PASSWORD` в AWS). Властивість, якої немає в AWS, валить синк **усього** секрета, тож спершу AWS, потім `kubectl apply`. ## Конвенції подів Кожен `KubernetesPodOperator` у цих DAG-ах ставить однаковий набір: `in_cluster=True`, `namespace="airflow"`, `startup_timeout_seconds=600`, `get_logs=True`, `on_finish_action="delete_succeeded_pod"`. - `delete_succeeded_pod`, а не `delete_pod` чи `keep_pod`: успішні поди прибираються, упалі лишаються для дебагу. Прибрати їх — `make clean-failed-pods`. - `image_pull_policy` залежить від тегу. `creative_metrics` іде з `Always`, бо тег `backend_latest` мутабельний (CI перепубліковує його) і закешований шар тихо виконував би вчорашні визначення метрик. Там, де образ зовнішній і версійований (`clickhouse-server:24.8`), достатньо `IfNotPresent`. - `start_date` — фіксований `pendulum.datetime(..., tz="UTC")`, `catchup=False`, `max_active_runs=1` усюди. - DAG-файл не робить нічого важкого на рівні модуля: планувальник перечитує його постійно. ## Дебаг - `python dags/dag_*.py` виконує `dag.test()` — таски виконуються **в процесі, а не в подах**. Це перевіряє структуру DAG-а й Python-логіку тасків, і не перевіряє ні образи, ні секрети, ні неймспейс. - Под у `Pending` → `startup_timeout_seconds`, місткість нод, або pull з ECR. - Таск упав без логів → скоріш за все OOM; `kubectl describe pod -n airflow` (упалі поди лишаються). - Змінної немає в контейнері → перевір, чи вона взагалі є в `kumquat-analytics-secrets`, і чи не переплутане одне підкреслювання з двома. - Скоркард тихо не оновлюється → asset-подія. Порівняй URI в `dags/assets.py` з тим, що видно в UI на вкладці Assets. ## Відомі межі - **Тестів на DAG-и немає.** Airflow не входить у локальне середовище (`backend/.venv` його не має), тож ні імпорт файлів, ні рендер шаблонів у CI не перевіряються. Помилка в Jinja-виразі вилазить у продакшені як упалий под. - **Як файли DAG-ів попадають у кластер — поза цим репозиторієм**, як і збірка образів: воркфлоу живе в спільних `plantin-org/ml-gha-workflows`. Через це константи `IMAGE` розписані непослідовно — `creative_metrics` тримає повний ECR-шлях, а `ch-sync` і `retention` досі мають заглушку `:latest # set by CI`, яку підставляє деплой. - **`container_resources` не задані ніде.** Поки поди легкі, це прийнятно; важкий таск без реквестів планувальник не зможе нормально розкласти й може бути OOM-killed без слідів. - **`media_sources` є лише в скоркарді.** Для import-DAG-а скоуп зашитий у константу `APPS`; додати продукт — це зміна коду.