ddmitry/clickstream-data-platform
1
0 Fork 0
Code Issues 5 Pull Requests Actions Packages Projects 1 Releases Wiki Activity

Конвейер etl_pipeline: сцепка дагов и пути запуска #156

New Issue
Closed
opened 2026-08-25 15:01:06 +03:00 by ddmitry · 1 comment
ddmitry commented 2026-08-25 15:01:06 +03:00
Owner
Copy link
Copy source

Part of #6.

Цель

Дневной конвейер: etl_pipeline дёргает dds_transform → dm_transform
с ожиданием, готовность источника — первым шагом, конвейер вписан в
жизненный цикл мира. Учебный результат (etl.md): один и тот же запуск
одинаково верен из триггера, с кнопки менти и после падения.

Что войдёт

  • Даг etl_pipeline: последовательный триггер слойных дагов с ожиданием
    каждого (ADR 0003 — зелёный конец графа не переживает отказ выше); без
    расписания и параметров.
  • Первый шаг dds_transform — ожидание дожёванного хвоста hits:
    нулевое отставание потребителей по system.kafka_consumers, опрос с
    таймаутом (etl.md, «Готовность источника»). Таблица эта локальна, а
    топик читают обе ноды — отставание проверять по обеим
    (clusterAllReplicas), не только по ноде подключения.
  • Пути запуска (etl.md): world_live_day дёргает конвейер с ожиданием
    последним шагом, после установки позиции; world_initialize —
    последним шагом, make up отдаёт собранные слои; world_next_day
    конвейер не дёргает. Отказ конвейера на инициализации чинится тем же
    путём, что и в живом дне, — ручным перезапуском etl_pipeline: позиция
    уже стоит, повторный world_initialize не нужен; назвать это в доке
    той же правкой.
  • Правка графов world_initialize и world_live_day — спеку жизненного
    цикла (docs/specs/2026-08-21-world-lifecycle.md) и задетые доки
    обновить тем же PR.
  • API триггера и ожидания в Airflow 3.x — проверить через MCP Context7
    (правило спеки §11), решение зафиксировать комментарием.

Границы

  • Партиционная переобработка и параметры дага — этап 5 (#7); стык даты
    мира и logical_date — #111.
  • Звено DQ — свой тикет: встанет в этот даг после DM.
  • Внутренности слойных дагов не менять — только сцепка, ожидание и пути
    запуска.

Кому что

Каркас дагов — Opus. Codex — прогоны трёх путей запуска и разбор логов по
брифу.

Сначала прочитать

  • docs/architecture/etl.md — целиком, это спека тикета;
  • docs/specs/2026-08-21-world-lifecycle.md — графы работников пульта;
  • ADR 0003, ADR 0009, ADR 0013;
  • dags/world_lifecycle.py и dags/world_control.py — куда встраивается
    триггер.

Проверка

  • make lint; на стенде make smoke, make check-clickhouse и
    make check-services (родился даг) зелёные.
  • make rebuild-storage + подъём: world_initialize зелёный, слои
    DDS и витрины revenue_daily / daily_traffic наполнены без единого
    ручного шага.
  • Живой день: world_live_day зелёный целиком; после него в
    сессиях — новый день, окно заказов пересобрано, витрины свежие.
  • Разгон пачкой: несколько world_next_day, затем один ручной запуск
    etl_pipeline догоняет всё (сессии — хвост, заказы — окно, карта и
    daily_traffic — целиком).
  • Падение трансформации (искусственное) валит etl_pipeline и
    world_live_day; повторный ручной запуск etl_pipeline чинит без
    параметров и памяти о прошлом прогоне.
  • Ожидание хвоста наблюдаемо в пакетном прогоне: отставание > 0 →
    опрос → 0 → трансформации.
Part of #6. ## Цель Дневной конвейер: `etl_pipeline` дёргает `dds_transform` → `dm_transform` с ожиданием, готовность источника — первым шагом, конвейер вписан в жизненный цикл мира. Учебный результат (etl.md): один и тот же запуск одинаково верен из триггера, с кнопки менти и после падения. ## Что войдёт - Даг `etl_pipeline`: последовательный триггер слойных дагов с ожиданием каждого (ADR 0003 — зелёный конец графа не переживает отказ выше); без расписания и параметров. - Первый шаг `dds_transform` — ожидание дожёванного хвоста `hits`: нулевое отставание потребителей по `system.kafka_consumers`, опрос с таймаутом (etl.md, «Готовность источника»). Таблица эта локальна, а топик читают обе ноды — отставание проверять по обеим (`clusterAllReplicas`), не только по ноде подключения. - Пути запуска (etl.md): `world_live_day` дёргает конвейер с ожиданием последним шагом, после установки позиции; `world_initialize` — последним шагом, `make up` отдаёт собранные слои; `world_next_day` конвейер не дёргает. Отказ конвейера на инициализации чинится тем же путём, что и в живом дне, — ручным перезапуском `etl_pipeline`: позиция уже стоит, повторный `world_initialize` не нужен; назвать это в доке той же правкой. - Правка графов `world_initialize` и `world_live_day` — спеку жизненного цикла (docs/specs/2026-08-21-world-lifecycle.md) и задетые доки обновить тем же PR. - API триггера и ожидания в Airflow 3.x — проверить через MCP Context7 (правило спеки §11), решение зафиксировать комментарием. ## Границы - Партиционная переобработка и параметры дага — этап 5 (#7); стык даты мира и `logical_date` — #111. - Звено DQ — свой тикет: встанет в этот даг после DM. - Внутренности слойных дагов не менять — только сцепка, ожидание и пути запуска. ## Кому что Каркас дагов — Opus. Codex — прогоны трёх путей запуска и разбор логов по брифу. ## Сначала прочитать - docs/architecture/etl.md — целиком, это спека тикета; - docs/specs/2026-08-21-world-lifecycle.md — графы работников пульта; - ADR 0003, ADR 0009, ADR 0013; - dags/world_lifecycle.py и dags/world_control.py — куда встраивается триггер. ## Проверка - [x] `make lint`; на стенде `make smoke`, `make check-clickhouse` и `make check-services` (родился даг) зелёные. - [x] `make rebuild-storage` + подъём: `world_initialize` зелёный, слои DDS и витрины `revenue_daily` / `daily_traffic` наполнены без единого ручного шага. - [x] Живой день: `world_live_day` зелёный целиком; после него в сессиях — новый день, окно заказов пересобрано, витрины свежие. - [x] Разгон пачкой: несколько `world_next_day`, затем один ручной запуск `etl_pipeline` догоняет всё (сессии — хвост, заказы — окно, карта и `daily_traffic` — целиком). - [x] Падение трансформации (искусственное) валит `etl_pipeline` и `world_live_day`; повторный ручной запуск `etl_pipeline` чинит без параметров и памяти о прошлом прогоне. - [x] Ожидание хвоста наблюдаемо в пакетном прогоне: отставание > 0 → опрос → 0 → трансформации.
ddmitry added a new dependency 2026-08-25 15:01:27 +03:00
#158 Звено DQ конвейера и dm.dq_summary
ddmitry commented 2026-08-26 22:52:10 +03:00
Author
Owner
Copy link
Copy source

Уточнение постановки после проверки живого стенда и развилки источника готовности.

Готовность hits

system.kafka_consumers в ClickHouse 26.3 показывает назначение разделов и текущий offset, но не конечный offset топика и не отставание. assignments.intent_size имеет другой смысл и на стенде равен NULL, rdkafka_stat пуст даже у активных потребителей. Поэтому нулевое отставание через эту таблицу доказать нельзя.

Готовность проверяется двумя последовательными барьерами внутри первого шага dds_transform.

  1. Kafka дочитана. Через уже установленный confluent-kafka шаг читает для группы clickstream_hits текущий и конечный offsets обоих разделов топика hits. Барьер пройден, когда группа и оба раздела присутствуют, оба offsets известны, суммарное отставание равно нулю. Отсутствие группы, раздела или значения означает «ещё не готово», а не ноль.
  2. ODS догружен. Коммит offsets означает запись блока в локальный спул Distributed, но не его доставку на шард: этот размен принят в docs/architecture/storage.md, раздел «Приём». Поэтому после нулевого отставания шаг через clusterAllReplicas ждёт, пока в system.distribution_queue не останется файлов для ods.event_dist ни на одной ноде. Остальные распределённые таблицы ждать не нужно: источником DDS служит ods.event_v, а сырьё и таблицу ошибок трансформации не читают.

Обе проверки опрашиваются раз в 3 с и делят один предел ожидания 300 с, а не получают по 300 с каждая. Каждый опрос пишет в журнал offsets и отставание по разделам, затем число файлов очереди по нодам. При отказе сообщение показывает последние offsets и незакрытые файлы с именами нод и last_exception.

Собственного состояния между опросами нет: offsets хранит Kafka, текущую очередь — ClickHouse, после перезапуска шаг вычисляет готовность заново. История > 0 → 0 остаётся только в журнале Airflow. Этого достаточно, потому что штатный генератор заканчивает запись до запуска ETL; параллельный посторонний производитель в контракт стенда не входит.

clusterAllReplicas(system.kafka_consumers) из условия готовности уходит: Kafka отвечает сразу за всю группу и все разделы. clusterAllReplicas применяется там, где у каждой ноды своя правда, — к очереди доставки ClickHouse.

Отклонено: брать готовую метрику у Kafka exporter — рабочий конвейер стал бы зависеть от мониторинга; выводить готовность по строкам ODS — потребовался бы ожидаемый объём каждого прогона и связь ETL с генератором; включать и разбирать статистику librdkafka через ClickHouse — косвенный источник и лишнее изменение DDL ради вопроса, которым владеет Kafka.

Проверка

К списку команд добавить make config-test: это дешёвая обязательная проверка правок DAG. Искусственный отказ не получает постоянного выключателя в коде: на приёмке временно ломается SQL daily_traffic, после красного world_live_day тестовая поломка убирается и ручной etl_pipeline должен починить слои без параметров.

Отдельный ADR не нужен. Исправление раздела «Готовность источника» в docs/architecture/etl.md входит в PR задачи.

Уточнение постановки после проверки живого стенда и развилки источника готовности. ## Готовность `hits` `system.kafka_consumers` в ClickHouse 26.3 показывает назначение разделов и текущий offset, но не конечный offset топика и не отставание. `assignments.intent_size` имеет другой смысл и на стенде равен `NULL`, `rdkafka_stat` пуст даже у активных потребителей. Поэтому нулевое отставание через эту таблицу доказать нельзя. Готовность проверяется двумя последовательными барьерами внутри первого шага `dds_transform`. 1. **Kafka дочитана.** Через уже установленный `confluent-kafka` шаг читает для группы `clickstream_hits` текущий и конечный offsets обоих разделов топика `hits`. Барьер пройден, когда группа и оба раздела присутствуют, оба offsets известны, суммарное отставание равно нулю. Отсутствие группы, раздела или значения означает «ещё не готово», а не ноль. 2. **ODS догружен.** Коммит offsets означает запись блока в локальный спул `Distributed`, но не его доставку на шард: этот размен принят в `docs/architecture/storage.md`, раздел «Приём». Поэтому после нулевого отставания шаг через `clusterAllReplicas` ждёт, пока в `system.distribution_queue` не останется файлов для `ods.event_dist` ни на одной ноде. Остальные распределённые таблицы ждать не нужно: источником DDS служит `ods.event_v`, а сырьё и таблицу ошибок трансформации не читают. Обе проверки опрашиваются раз в 3 с и делят один предел ожидания 300 с, а не получают по 300 с каждая. Каждый опрос пишет в журнал offsets и отставание по разделам, затем число файлов очереди по нодам. При отказе сообщение показывает последние offsets и незакрытые файлы с именами нод и `last_exception`. Собственного состояния между опросами нет: offsets хранит Kafka, текущую очередь — ClickHouse, после перезапуска шаг вычисляет готовность заново. История `> 0 → 0` остаётся только в журнале Airflow. Этого достаточно, потому что штатный генератор заканчивает запись до запуска ETL; параллельный посторонний производитель в контракт стенда не входит. `clusterAllReplicas(system.kafka_consumers)` из условия готовности уходит: Kafka отвечает сразу за всю группу и все разделы. `clusterAllReplicas` применяется там, где у каждой ноды своя правда, — к очереди доставки ClickHouse. Отклонено: брать готовую метрику у Kafka exporter — рабочий конвейер стал бы зависеть от мониторинга; выводить готовность по строкам ODS — потребовался бы ожидаемый объём каждого прогона и связь ETL с генератором; включать и разбирать статистику librdkafka через ClickHouse — косвенный источник и лишнее изменение DDL ради вопроса, которым владеет Kafka. ## Проверка К списку команд добавить `make config-test`: это дешёвая обязательная проверка правок DAG. Искусственный отказ не получает постоянного выключателя в коде: на приёмке временно ломается SQL `daily_traffic`, после красного `world_live_day` тестовая поломка убирается и ручной `etl_pipeline` должен починить слои без параметров. Отдельный ADR не нужен. Исправление раздела «Готовность источника» в `docs/architecture/etl.md` входит в PR задачи.
ddmitry self-assigned this 2026-08-26 23:09:38 +03:00
Sign in to join this conversation.
No Branch/Tag Specified
Labels
Clear labels
needs-info
needs-triage
ready-for-agent
ready-for-human
wayfinder:grilling
wayfinder:map
wayfinder:prototype
wayfinder:research
wayfinder:task
wontfix
No labels
Milestone
No items
No Milestone
Projects
Clear projects
No projects
Assignees
Clear assignees
ddmitry (Dmitry Dementiev)
No Assignees ddmitry
1 Participants
Notifications
Due Date
No due date set.
Blocks
#158 Звено DQ конвейера и dm.dq_summary
ddmitry/clickstream-data-platform
Depends on
#154 Витрины стенда: даг dm_transform, revenue_daily и daily_traffic
ddmitry/clickstream-data-platform
Reference: ddmitry/clickstream-data-platform#156