diff --git a/dags/dds_transform.py b/dags/dds_transform.py index 8cccc73..2930069 100644 --- a/dags/dds_transform.py +++ b/dags/dds_transform.py @@ -19,6 +19,8 @@ from __future__ import annotations import datetime import logging +import os +import time from pathlib import Path from airflow.exceptions import AirflowException @@ -34,6 +36,15 @@ START_DATE = datetime.datetime(2026, 1, 1, tzinfo=datetime.UTC) SQL_ROOT = Path("/opt/airflow/sql") CLICKHOUSE_CONNECTION = "clickhouse_default" +CLICKHOUSE_CLUSTER = "clickstream_cluster" + +KAFKA_BOOTSTRAP_SERVERS = os.environ["KAFKA_BOOTSTRAP_SERVERS"] +HITS_TOPIC = os.environ["KAFKA_TOPIC"] +HITS_GROUP = "clickstream_hits" +HITS_PARTITIONS = (0, 1) + +READINESS_TIMEOUT_SECONDS = 300 +READINESS_POLL_SECONDS = 3 # Позиция на оси модельного времени: номер первого несыгранного дня. Её ставит # пульт мира, и только она отличает прожитый день от живого — зачем это @@ -41,6 +52,178 @@ CLICKHOUSE_CONNECTION = "clickhouse_default" WORLD_POSITION = "world_position" +def _kafka_readiness() -> tuple[bool, str]: + """Получить подтвержденные и верхние смещения обоих разделов событий.""" + from confluent_kafka import Consumer, ConsumerGroupTopicPartitions, TopicPartition + from confluent_kafka.admin import AdminClient + + admin = AdminClient({"bootstrap.servers": KAFKA_BOOTSTRAP_SERVERS}) + groups = admin.list_consumer_groups(request_timeout=10).result() + if groups.errors: + raise RuntimeError(f"Kafka не вернула список групп: {groups.errors}") + + group_exists = any(group.group_id == HITS_GROUP for group in groups.valid) + committed: dict[int, int] = {} + if group_exists: + request = ConsumerGroupTopicPartitions( + HITS_GROUP, + [TopicPartition(HITS_TOPIC, partition) for partition in HITS_PARTITIONS], + ) + result = admin.list_consumer_group_offsets([request], request_timeout=10)[ + HITS_GROUP + ].result() + for topic_partition in result.topic_partitions: + if topic_partition.error is None and topic_partition.offset >= 0: + committed[topic_partition.partition] = topic_partition.offset + + high_watermarks: dict[int, int] = {} + watermark_errors: dict[int, str] = {} + consumer = Consumer( + { + "bootstrap.servers": KAFKA_BOOTSTRAP_SERVERS, + # Клиент требует group.id, но без подписки и poll() в группу не + # входит: здесь он только запрашивает верхние смещения топика. + "group.id": "etl-readiness", + "enable.auto.commit": False, + } + ) + try: + for partition in HITS_PARTITIONS: + try: + bounds = consumer.get_watermark_offsets( + TopicPartition(HITS_TOPIC, partition), timeout=10 + ) + if bounds is not None and bounds[1] >= 0: + high_watermarks[partition] = bounds[1] + except Exception as error: + watermark_errors[partition] = str(error) + finally: + consumer.close() + + all_known = group_exists + total_lag = 0 + parts: list[str] = [] + for partition in HITS_PARTITIONS: + offset = committed.get(partition) + high = high_watermarks.get(partition) + lag = high - offset if offset is not None and high is not None else None + if lag is None or lag < 0: + all_known = False + else: + total_lag += lag + logging.info( + "Kafka %s[%s]: подтверждено=%s, верхнее смещение=%s, отставание=%s", + HITS_TOPIC, + partition, + offset, + high, + lag, + ) + parts.append( + f"{HITS_TOPIC}[{partition}]: подтверждено={offset}, " + f"верхнее смещение={high}, отставание={lag}" + ) + if partition in watermark_errors: + parts[-1] += ( + f", ошибка чтения верхнего смещения={watermark_errors[partition]}" + ) + + if not group_exists: + logging.info("группа потребителей %s отсутствует", HITS_GROUP) + lag_text = str(total_lag) if all_known else "неизвестно" + logging.info("суммарное отставание Kafka: %s", lag_text) + detail = "; ".join(parts) + if not group_exists: + detail = f"группа {HITS_GROUP} отсутствует; {detail}" + return all_known and total_lag == 0, detail + + +def _delivery_queue_readiness() -> tuple[bool, str]: + """Получить число недоставленных файлов событий на каждой ноде.""" + query = f""" + SELECT + node, + sum(queued_files) AS files, + arrayStringConcat( + groupArrayIf( + last_exception, + queued_files > 0 AND notEmpty(last_exception) + ), + '; ' + ) AS last_exception + FROM + ( + SELECT + hostName() AS node, + toUInt64(0) AS queued_files, + '' AS last_exception + FROM clusterAllReplicas('{CLICKHOUSE_CLUSTER}', system.one) + + UNION ALL + + SELECT + hostName(), + data_files + broken_data_files AS queued_files, + last_exception + FROM clusterAllReplicas( + '{CLICKHOUSE_CLUSTER}', system.distribution_queue + ) + WHERE database = 'ods' AND table = 'event_dist' + ) + GROUP BY node + ORDER BY node + """ + rows = ClickHouseHook(clickhouse_conn_id=CLICKHOUSE_CONNECTION).get_records(query) + ready = True + nodes: list[str] = [] + for node, files, last_exception in rows: + files = int(files) + ready = ready and files == 0 + logging.info( + "очередь ods.event_dist на %s: файлов=%s, последняя ошибка=%s", + node, + files, + last_exception or "нет", + ) + detail = f"{node}: файлов={files}" + if last_exception: + detail += f", последняя ошибка={last_exception}" + nodes.append(detail) + return ready, "; ".join(nodes) + + +def _wait_for_phase( + probe, + phase: str, + deadline: float, +) -> None: + """Дождаться прохождения одного барьера до общего крайнего срока.""" + last_observation = "первый опрос не выполнен" + last_error: str | None = None + while time.monotonic() < deadline: + try: + ready, last_observation = probe() + last_error = None + except Exception as error: + ready = False + last_error = str(error) + logging.warning("%s: ошибка опроса: %s", phase, last_error) + if ready: + return + + remaining = deadline - time.monotonic() + if remaining > 0: + time.sleep(min(READINESS_POLL_SECONDS, remaining)) + + detail = last_observation + if last_error: + detail += f"; ошибка последнего опроса: {last_error}" + raise AirflowException( + f"общий срок ожидания готовности в {READINESS_TIMEOUT_SECONDS} с " + f'истек на этапе "{phase}": {detail}' + ) + + def _world_position() -> int: """Позиция мира числом, с внятным отказом на непригодном значении. @@ -110,6 +293,17 @@ def replacements(scope: dict[str, object]) -> list[dict[str, object]]: def dds_transform(): """Пересобрать сущности DDS по текущему состоянию источников.""" + @task + def wait_for_source() -> None: + """Дождаться чтения событий из Kafka и их доставки в ODS.""" + deadline = time.monotonic() + READINESS_TIMEOUT_SECONDS + _wait_for_phase(_kafka_readiness, "Kafka", deadline) + _wait_for_phase( + _delivery_queue_readiness, + "очередь доставки ClickHouse", + deadline, + ) + @task_group(group_id="order") def order(): """Заказ: дышащее окно плюс дни, которых в слое ещё нет.""" @@ -229,8 +423,10 @@ def dds_transform(): rebuild >> replace + source_ready = wait_for_source() orders = order() - session() + sessions = session() + source_ready >> [orders, sessions] identity_map = SQLExecuteQueryOperator( task_id="identity_map", diff --git a/dags/etl_pipeline.py b/dags/etl_pipeline.py new file mode 100644 index 0000000..918dc2d --- /dev/null +++ b/dags/etl_pipeline.py @@ -0,0 +1,43 @@ +"""Дневной конвейер: сначала DDS, затем DM. + +Своих преобразований у дага нет. Ждущие триггеры связывают состояния: +если слой краснеет, конвейер не запускает следующий слой и краснеет сам. +""" + +from __future__ import annotations + +import datetime + +from airflow.providers.standard.operators.trigger_dagrun import TriggerDagRunOperator +from airflow.sdk import dag + +START_DATE = datetime.datetime(2026, 1, 1, tzinfo=datetime.UTC) + + +@dag( + dag_id="etl_pipeline", + schedule=None, + start_date=START_DATE, + is_paused_upon_creation=False, + max_active_runs=1, + tags=["etl"], +) +def etl_pipeline(): + """Собрать слои последовательно по текущему состоянию данных.""" + dds = TriggerDagRunOperator( + task_id="trigger_dds", + trigger_dag_id="dds_transform", + wait_for_completion=True, + poke_interval=10, + ) + dm = TriggerDagRunOperator( + task_id="trigger_dm", + trigger_dag_id="dm_transform", + wait_for_completion=True, + poke_interval=10, + ) + + dds >> dm + + +etl_pipeline() diff --git a/dags/world_control.py b/dags/world_control.py index 47d23a8..e991c08 100644 --- a/dags/world_control.py +++ b/dags/world_control.py @@ -12,6 +12,9 @@ идут отправка слепка тем же контейнером генератора и ждущий триггер дага приёма `orders_ingest`. +Живой день после установки позиции дожидается еще и `etl_pipeline`. Пачка +только разгоняет источники: слои после нее менти догоняет ручным запуском. + Разделение не косметическое. Расписание на самом работнике заставило бы кнопку паузы значить две вещи разом — «мир не едет сам» и «даг выключен», — а работник при этом выглядел бы в списке выключенным, хотя нажимают его каждый день. @@ -209,7 +212,14 @@ def world_live_day(): ["snapshot", "--day", PLAYED_DAY, "--topic", ORDERS_TOPIC], ) - first_day >> played >> sent >> _ingest_orders() >> remember_played(first_day) + position = remember_played(first_day) + etl = TriggerDagRunOperator( + task_id="trigger_etl_pipeline", + trigger_dag_id="etl_pipeline", + wait_for_completion=True, + poke_interval=10, + ) + first_day >> played >> sent >> _ingest_orders() >> position >> etl @dag( diff --git a/dags/world_lifecycle.py b/dags/world_lifecycle.py index abcd356..130a73a 100644 --- a/dags/world_lifecycle.py +++ b/dags/world_lifecycle.py @@ -1,9 +1,10 @@ """Создание и пересоздание прикладного мира. Compose поднимает службы, а этот граф создаёт всё прикладное поверх них: -топики, схему ClickHouse, данные стартовых дней и позицию модельного времени. -`world_initialize` безопасен для обычного `make up`; `world_recreate` — явная -операция обслуживания, которая сначала объявляет прежний мир недействительным. +топики, схему ClickHouse, данные стартовых дней, позицию модельного времени и +готовые слои. `world_initialize` безопасен для обычного `make up`; +`world_recreate` — явная операция обслуживания, которая сначала объявляет +прежний мир недействительным. Контракт состояний и доводы — ADR 0013 и спека жизненного цикла мира. """ @@ -262,9 +263,15 @@ def world_initialize(): poke_interval=10, ) position = remember_starting_position() + etl = TriggerDagRunOperator( + task_id="trigger_etl_pipeline", + trigger_dag_id="etl_pipeline", + wait_for_completion=True, + poke_interval=10, + ) route >> [ready, topics] - chain(topics, *ddl_tasks, image, events, arrived, snapshots, orders, position) + chain(topics, *ddl_tasks, image, events, arrived, snapshots, orders, position, etl) @dag( diff --git a/docs/adr/0013-world-lifecycle.md b/docs/adr/0013-world-lifecycle.md index 8b48a40..f930a97 100644 --- a/docs/adr/0013-world-lifecycle.md +++ b/docs/adr/0013-world-lifecycle.md @@ -25,9 +25,11 @@ Kafka, DDL прикладных баз ClickHouse, стартовыми данн `world_recreate` ставит `-1` до первого удаления и `0` только после успешной очистки. `world_initialize` принимает отсутствие позиции или `0`, а значение -`8` и больше считает готовым миром. Позиция становится `8` последней задачей, -после генерации стартовых дней и успешного приёма обоих источников. Обычные -даги движения мира работают только с готовой позицией. +`8` и больше считает готовым миром. После генерации стартовых дней и успешного +приема обоих источников позиция становится `8`; затем инициализация последним +шагом ждет дневной конвейер `etl_pipeline`. При отказе преобразований мир +остается готовым на позиции `8`, а конвейер после исправления запускают рукой. +Обычные даги движения мира работают только с готовой позицией. ## Почему @@ -39,9 +41,10 @@ Kafka, DDL прикладных баз ClickHouse, стартовыми данн Kafka сообщения и офсеты прежнего мира. Airflow уже ведёт последующие дни и приём заказов. Отдав ему начало той же оси, -получаем один наблюдаемый процесс и одну точку успеха: зелёный -`world_initialize` означает, что схема создана, стартовый мир принят и его -следующий день известен. +получаем один наблюдаемый процесс и одну точку успеха: зеленая ветка создания в +`world_initialize` подтверждает, что схема создана, стартовый мир принят, его +следующий день известен, а слои DDS и DM собраны. Короткая ветка готового мира +по-прежнему подтверждает позицию и ничего не меняет. ## Что это решение уточняет diff --git a/docs/architecture/dm.md b/docs/architecture/dm.md index 51c9827..0012328 100644 --- a/docs/architecture/dm.md +++ b/docs/architecture/dm.md @@ -125,9 +125,9 @@ Airflow. Партиция — день данных; прогон звена DQ `TRUNCATE+INSERT` выполняется той же механикой замены, что и у витрин, — ничего отдельного учить не надо. -## Звено DQ конвейера +## Будущее звено DQ конвейера -Встаёт в `etl_pipeline` после DM ([конвейер](etl.md)) и проверяет только +Встанет в `etl_pipeline` после DM ([конвейер](etl.md)) и проверит только объекты стенда: - счётная сверка сессий с эталоном `VisitID` — кука × день, эталон читается diff --git a/docs/architecture/etl.md b/docs/architecture/etl.md index cd1f54b..27b112c 100644 --- a/docs/architecture/etl.md +++ b/docs/architecture/etl.md @@ -13,8 +13,8 @@ Дневной конвейер на стенде один — даг `etl_pipeline`. Своей работы у него нет: он дёргает слойные даги последовательно и ждёт каждый — -`dds_transform`, затем `dm_transform`. Контур DQ этапа 4 встаёт его же -звеном после DM; состав звена — в [доке слоя DM](dm.md). +`dds_transform`, затем `dm_transform`. Будущее звено DQ встанет после DM. +Его состав описан в [доке слоя DM](dm.md). Слойные даги устроены по образцу `orders_ingest`: без расписания, один активный прогон, запускаются триггером конвейера или рукой при разборе @@ -22,6 +22,14 @@ ## Три пути запуска +```text +world_initialize ─┐ +world_live_day ───┼─> etl_pipeline ─> dds_transform ─> dm_transform +запуск рукой ─────┘ + +world_next_day ─────> источники без ETL +``` + - **Живой день.** `world_live_day` дёргает конвейер с ожиданием — последним шагом, после установки позиции. Порядок несёт смысл: день сыгран — факт, упавшая трансформация его не отменяет, поэтому позиция ставится раньше; но @@ -73,17 +81,26 @@ ODS и ещё не собранный в DDS, затем при необходи ## Готовность источника -Первым шагом `dds_transform` обязан дождаться, пока хвост топика `hits` -дожёван: отставание его потребителей нулевое (`system.kafka_consumers`). -Зелёный приём заказов о событиях не говорит ничего — топики разные и -потребители разные. В живом режиме зазор ничтожен, но в пакетном генератор -выстреливает дни в топик быстрее, чем чтец их прожёвывает, и сессии собираются -по недожёванному дню. DQ-сверка такой день подсветит, но починить его до -этапа 5 нечем: профилактика дешевле ремонта, которого ещё нет. +Первый шаг `dds_transform` проходит два барьера за общий срок в 300 секунд. +Сначала клиент `confluent-kafka` сверяет подтвержденные смещения группы +`clickstream_hits` с верхними смещениями обоих разделов `hits`. Группа, оба +раздела и все смещения должны существовать. Суммарное отставание должно стать +нулевым. Каждый опрос пишет в журнал оба смещения и отставание по разделу. -Шага этого в даге ещё нет — он приезжает отдельным тикетом (#156), и до тех -пор дыра открыта: сессии считают день прожитым по позиции мира, а хвост потока -не стережёт никто. На приёмке отставание смотрят руками. +Только после этого шаг проверяет обе ноды ClickHouse через +`clusterAllReplicas`: у `ods.event_dist` в `system.distribution_queue` не +должно остаться файлов. Каждый опрос пишет в журнал ноды, число файлов и +`last_exception` очереди. Оба барьера опрашиваются раз в три секунды и делят +один крайний срок. После Kafka срок не отсчитывается заново. + +Зелёный приём заказов о событиях не говорит ничего — топики разные и +потребители разные. Одного нулевого отставания Kafka тоже мало: чтец мог +подтвердить сообщение раньше, чем Distributed-таблица доставила файл на шард. +Два барьера закрывают обе части пути до того, как сессии выберут свой объем. + +Ждущие вызовы сделаны `TriggerDagRunOperator`. Проверка через Context7 по +ближайшей доступной документации Airflow 3.x (версия 3.1.6) подтвердила, что +оператор опрашивает точный созданный им прогон и краснеет вслед за потомком. ## Идемпотентность diff --git a/docs/architecture/testing.md b/docs/architecture/testing.md index 535678d..84d06b2 100644 --- a/docs/architecture/testing.md +++ b/docs/architecture/testing.md @@ -237,6 +237,33 @@ Kafka → STG → ODS, и живёт он в `make check-clickhouse`. ## Что проверено +**Приемка #156 27 августа 2026 года.** `make rebuild-storage` занял 7 минут. +На чистом прикладном мире `world_initialize` сам довел позицию до `8`, дождался +`etl_pipeline` и оставил заполненными DDS и обе витрины. После пересборки прошли +`make smoke` 21/21 и `make check-clickhouse` 10/10. + +Два прогона `world_next_day` не изменили слои; один ручной `etl_pipeline` без +параметров догнал оба дня. Полный прогон `world_live_day` занял 27 м 2 с и +завершился только после успешных DDS и DM. + +Временная синтаксическая поломка `sql/dm/daily_traffic_rebuild.sql` дала +красную цепочку: DDS завершился успешно, а `daily_traffic`, DM, `etl_pipeline` +и `world_live_day` покраснели. После возврата исправного SQL ручной +`etl_pipeline` без параметров завершился успешно на позиции `12` и восстановил +слои. + +Для проверки пакетного барьера `stg.hits_raw_mv` временно отсоединили командой +`DETACH` на обеих нодах. После `world_next_day` отставание Kafka составило +46 367 (19 128 + 27 239). Во время ожидания матвью вернули командой `ATTACH`. +Журнал готовности записал повторные опросы с отставанием 46 367, затем 27 239 +и 0; после этого очередь `ods.event_dist` содержала 0 файлов на обеих нодах, +и ETL завершился успешно. Проверка `EXISTS` снова дала `1` на обеих нодах, +позиция равна `13`. + +После пересборки `make check-services` дважды дал 5/1: только импорт Superset +не уложился в существующий тайм-аут 30 с. Диагностический пробник импорта с +ожиданием 90 с прошел 3/0. Финальный штатный повтор прошел 7/7 за 129 с. + **Перезамер 21 августа 2026 года при исполнении #124.** Образ Airflow собран с `apache-airflow-providers-clickhousedb==1.0.0`, `apache-airflow-providers-common-sql==2.0.1` и diff --git a/docs/specs/2026-08-21-world-lifecycle.md b/docs/specs/2026-08-21-world-lifecycle.md index 5890ae2..b73f698 100644 --- a/docs/specs/2026-08-21-world-lifecycle.md +++ b/docs/specs/2026-08-21-world-lifecycle.md @@ -20,8 +20,9 @@ DDL и двумя одноразовыми службами заливал во **Чему учит переезд.** Менти видит один наблюдаемый граф, которому принадлежат прикладная схема, данные мира и позиция модельного времени, — и Compose, -который держит только инфраструктуру. Зелёный `world_initialize` значит ровно -одно: схема создана, оба источника приняты, следующий день известен. +который держит только инфраструктуру. Зеленая ветка создания в +`world_initialize` подтверждает: схема создана, оба источника приняты, +следующий день известен, слои DDS и DM собраны. ## Решение @@ -106,11 +107,16 @@ Kafka, `SQLExecuteQueryOperator` для исполняемых SQL-файлов приехало» условием не считается. 7. Отправляет слепки заказов вторым прогоном образа — со сдвигом, дни 0…6 ([заказы в стартовом мире](../architecture/orders/start-world.md)). -8. Дёргает `orders_ingest` ждущим триггером и зеленеет только вслед за ним. -9. Последней успешной задачей ставит позицию в `WORLD_STARTING_DAYS`. +8. Дергает `orders_ingest` ждущим триггером и продолжает только после его + успеха. +9. Ставит позицию в `WORLD_STARTING_DAYS`: источники приняты, мир уже создан. +10. Последним шагом дергает `etl_pipeline` ждущим триггером. Конвейер + последовательно ждет `dds_transform` и `dm_transform`; отказ слоя красит + инициализацию. Порядок задач связывает проверки: каждый шаг подтверждён предыдущим, а красная -задача называет место остановки. +задача называет место остановки. Ветка "мир уже существует" конвейер не +запускает: обычный повтор `make up` по-прежнему ничего не меняет. ### `world_recreate` — снести мир и создать заново @@ -163,9 +169,15 @@ Kafka, `SQLExecuteQueryOperator` для исполняемых SQL-файлов недействителен и пересборку надо повторить. Стенд обязан назвать незавершённую очистку вслух и не продолжать работу молча. -**Очистка прошла, потомок не доработал.** Позиция остаётся `0`. Инициализация -детерминирована, поэтому следующий `make up` просто пробует её заново — с -того же места и с тем же результатом. +**Очистка прошла, создание остановилось до установки позиции.** Позиция +остаётся `0`. Инициализация детерминирована, поэтому следующий `make up` просто +пробует её заново — с того же места и с тем же результатом. + +**ETL упал после установки позиции.** Мир и оба источника уже готовы, поэтому +позиция остается в `WORLD_STARTING_DAYS`. После исправления запускают рукой +`etl_pipeline`: он сам выводит объем из состояния данных. Повторный +`world_initialize` здесь не поможет: при готовой позиции инициализация уйдет в +короткую ветку без изменений. **Позицию в `WORLD_STARTING_DAYS` ставит только успех обоих источников.** События доехали, слепки приняты, `orders_ingest` зелёный — тогда и только тогда @@ -181,10 +193,11 @@ ODS истинно сразу — этим оно и ограничено, об заливка» ни один счёт дать не может. Сторожа на этот случай нет: ошибку менти делает себе сам и тут же видит. -**Слоты Airflow.** Их четыре, и вложенные ожидания в них укладываются, пока -пересборка не наложилась на движение мира: `world_recreate` ждёт -`world_initialize`, тот ждёт `orders_ingest` — три слота из четырёх. Одновременно -работающий выключатель добавляет свои и загоняет стенд в тупик ожиданий. +**Слоты Airflow.** Их четыре. При пересборке `world_recreate` ждет +`world_initialize`, тот ждет `etl_pipeline`, а конвейер ждет слой. В живом +режиме такую же цепочку образуют `world_live`, `world_live_day`, `etl_pipeline` +и слой. В обоих случаях три ожидающие задачи оставляют слою последний слот. +Одновременная пересборка добавляет ожидания и загоняет стенд в тупик. Поэтому пересборка — явная операция обслуживания: оператор ставит `world_live` на паузу и даёт работникам доиграть. Предусловие остаётся в документации; механизм блокировок стоил бы дороже этой строки. @@ -214,8 +227,8 @@ ODS истинно сразу — этим оно и ограничено, об мир уже есть. - **`make rebuild-storage`** возвращает канонический мир: позиция проходит `-1` → `0` → `WORLD_STARTING_DAYS`, в ODS ровно даты описи, топики созданы - заново, офсеты групп идут с нуля. Цена прогона называется числом и сверяется - с прежними 2 м 25 с. + заново, офсеты групп идут с нуля, DDS и DM собраны. Время прогона измеряется + заново. - **Красные пути проверяются своими отказами.** Сторож работника при отсутствующей, нулевой и отрицательной позиции; `world_initialize` на `-1`; оборванная очистка, оставляющая `-1`. Проверка, не видевшая своей поломки, diff --git a/sql/dds/session_scope.sql b/sql/dds/session_scope.sql index fc86d8d..4079e32 100644 --- a/sql/dds/session_scope.sql +++ b/sql/dds/session_scope.sql @@ -20,9 +20,9 @@ -- ранние дни из ods.event не исчезают: срок жизни назначен сырью и таблицам -- ошибок, самому событию — нет (sql/ddl/20-ods-tables.sql). -- --- За «доехал» позиция не отвечает — только за «прожит». Дождаться хвоста --- потока — работа отдельного шага готовности источника, и в даге его пока нет --- (#156): до него недожёванный хвост даёт замёрзший недособранный день +-- За «доехал» позиция не отвечает — только за «прожит». Перед расчётом объёма +-- первый шаг дага дожидается хвоста потока и очереди доставки: иначе +-- недожеванный хвост заморозил бы недособранный день -- (docs/architecture/etl.md, «Готовность источника»). WITH (SELECT min(EventDate) FROM ods.event_dist) AS first_source_day,