Airflow DAG-и

Шість 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/. Хто потім читає наповнену таблицю — Аналітика креативів.

Що є в dags/

Образи задаються константою 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 і 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-схема), тож із документів видно, з яким набором джерел він рахувався.

І sync_test_status:

Усі три йдуть у под флагами в 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 поступово означає «колись запускали», і разом із ним пливуть усі фільтри галереї й колонки експорту, побудовані на ньому.

Кабінет уже каже і коли тест закінчився, і коли почався знову — креатив перестає або знову починає коштувати грошей. Звідси рівно два ребра:

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:

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; додати продукт — це зміна коду.