diff --git a/dags/dds_transform.py b/dags/dds_transform.py index 2930069..46ef624 100644 --- a/dags/dds_transform.py +++ b/dags/dds_transform.py @@ -53,9 +53,9 @@ WORLD_POSITION = "world_position" def _kafka_readiness() -> tuple[bool, str]: - """Получить подтвержденные и верхние смещения обоих разделов событий.""" - from confluent_kafka import Consumer, ConsumerGroupTopicPartitions, TopicPartition - from confluent_kafka.admin import AdminClient + """Проверить, дочитала ли группа оба раздела событий.""" + from confluent_kafka import ConsumerGroupTopicPartitions, TopicPartition + from confluent_kafka.admin import AdminClient, OffsetSpec admin = AdminClient({"bootstrap.servers": KAFKA_BOOTSTRAP_SERVERS}) groups = admin.list_consumer_groups(request_timeout=10).result() @@ -78,27 +78,22 @@ def _kafka_readiness() -> tuple[bool, str]: high_watermarks: dict[int, int] = {} watermark_errors: dict[int, str] = {} - consumer = Consumer( + # OffsetSpec.latest возвращает следующее смещение после хвоста; сверено + # через Context7 и на установленном confluent-kafka 2.15.0. + latest = admin.list_offsets( { - "bootstrap.servers": KAFKA_BOOTSTRAP_SERVERS, - # Клиент требует group.id, но без подписки и poll() в группу не - # входит: здесь он только запрашивает верхние смещения топика. - "group.id": "etl-readiness", - "enable.auto.commit": False, - } + TopicPartition(HITS_TOPIC, partition): OffsetSpec.latest() + for partition in HITS_PARTITIONS + }, + request_timeout=10, ) - 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() + for topic_partition, future in latest.items(): + try: + offset = future.result().offset + if offset >= 0: + high_watermarks[topic_partition.partition] = offset + except Exception as error: + watermark_errors[topic_partition.partition] = str(error) all_known = group_exists total_lag = 0 @@ -139,7 +134,7 @@ def _kafka_readiness() -> tuple[bool, str]: def _delivery_queue_readiness() -> tuple[bool, str]: - """Получить число недоставленных файлов событий на каждой ноде.""" + """Проверить, пуста ли очередь доставки событий на каждой ноде.""" query = f""" SELECT node, @@ -153,6 +148,8 @@ def _delivery_queue_readiness() -> tuple[bool, str]: ) AS last_exception FROM ( + -- У пустой очереди нет строк; system.one оставляет обе ноды + -- видимыми и в зеленом опросе. SELECT hostName() AS node, toUInt64(0) AS queued_files, diff --git a/docs/architecture/etl.md b/docs/architecture/etl.md index 27b112c..05ea681 100644 --- a/docs/architecture/etl.md +++ b/docs/architecture/etl.md @@ -16,6 +16,10 @@ `dds_transform`, затем `dm_transform`. Будущее звено DQ встанет после DM. Его состав описан в [доке слоя DM](dm.md). +Ждущие вызовы сделаны `TriggerDagRunOperator`. Проверка через Context7 по +ближайшей доступной документации Airflow 3.x (версия 3.1.6) подтвердила, что +оператор опрашивает точный созданный им прогон и краснеет вслед за потомком. + Слойные даги устроены по образцу `orders_ingest`: без расписания, один активный прогон, запускаются триггером конвейера или рукой при разборе происшествий. Витрины внутри `dm_transform` — отдельные задачи. @@ -98,10 +102,6 @@ ODS и ещё не собранный в DDS, затем при необходи подтвердить сообщение раньше, чем Distributed-таблица доставила файл на шард. Два барьера закрывают обе части пути до того, как сессии выберут свой объем. -Ждущие вызовы сделаны `TriggerDagRunOperator`. Проверка через Context7 по -ближайшей доступной документации Airflow 3.x (версия 3.1.6) подтвердила, что -оператор опрашивает точный созданный им прогон и краснеет вслед за потомком. - ## Идемпотентность Партиция не заменяется простой вставкой. День собирается в таблицу-двойник — diff --git a/docs/architecture/testing.md b/docs/architecture/testing.md index 84d06b2..a53666f 100644 --- a/docs/architecture/testing.md +++ b/docs/architecture/testing.md @@ -240,7 +240,8 @@ 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. +`make smoke` 21/21 и `make check-clickhouse` 10/10. Перед стендовыми сценариями +прошли `make config-test` и `make lint`. Два прогона `world_next_day` не изменили слои; один ручной `etl_pipeline` без параметров догнал оба дня. Полный прогон `world_live_day` занял 27 м 2 с и diff --git a/docs/specs/2026-08-21-world-lifecycle.md b/docs/specs/2026-08-21-world-lifecycle.md index b73f698..1837e58 100644 --- a/docs/specs/2026-08-21-world-lifecycle.md +++ b/docs/specs/2026-08-21-world-lifecycle.md @@ -227,8 +227,7 @@ ODS истинно сразу — этим оно и ограничено, об мир уже есть. - **`make rebuild-storage`** возвращает канонический мир: позиция проходит `-1` → `0` → `WORLD_STARTING_DAYS`, в ODS ровно даты описи, топики созданы - заново, офсеты групп идут с нуля, DDS и DM собраны. Время прогона измеряется - заново. + заново, офсеты групп идут с нуля, DDS и DM собраны. - **Красные пути проверяются своими отказами.** Сторож работника при отсутствующей, нулевой и отрицательной позиции; `world_initialize` на `-1`; оборванная очистка, оставляющая `-1`. Проверка, не видевшая своей поломки,