diff --git a/dags/dq_check.py b/dags/dq_check.py index 73b9cda..563feeb 100644 --- a/dags/dq_check.py +++ b/dags/dq_check.py @@ -74,7 +74,11 @@ def _publish_summary( ) -> list[str]: """Записать дневную сводку и вернуть малый срез готовой диагностики.""" if not summaries: - raise AirflowException(f"{check_name}: область проверки пуста") + raise AirflowException( + f"{check_name}: запрос не вернул ни одного дня для проверки; " + "проверьте наличие данных в источниках сверки. Состав проверок " + "описан в docs/architecture/dm.md" + ) diagnostics: list[str] = [] context = get_current_context() @@ -148,7 +152,10 @@ def build_classes_vs_inventory() -> list[str]: for mismatch_class, count in day["orders"].items() } if not expected: - raise AirflowException("classes_vs_inventory: в паспорте нет закрытых дней") + raise AirflowException( + "classes_vs_inventory: в data/world-inventory.json нет счетчиков " + "заказов для закрытых дней; сообщите наставнику" + ) query = (SQL_ROOT / "dq" / "classes_vs_inventory.sql").read_text(encoding="utf-8") # GROUP BY закономерно не возвращает группу из нуля строк. @@ -179,9 +186,9 @@ def build_classes_vs_inventory() -> list[str]: failed_rows += 1 actual_text = actual_count if actual_count is not None else "нет ключа" details.append( - f"{day} key={mismatch_class}: " - f"эталон={reference_count}, " - f"объект={actual_text}" + f"{day} mismatch_class={mismatch_class}: " + f"заказов в паспорте={reference_count}, " + f"в dm.purchase_vs_orders_v={actual_text}" ) summaries.append((data_date, reference_rows, actual_rows, failed_rows, details)) return _publish_summary("classes_vs_inventory", summaries) @@ -194,7 +201,10 @@ def publication_scope() -> list[dict[str, dict[str, str]]]: "SELECT DISTINCT data_date FROM dm.dq_summary_stage_dist ORDER BY data_date" ) if not rows: - raise AirflowException("донор dm.dq_summary пуст: публиковать нечего") + raise AirflowException( + "dm.dq_summary_stage_dist пуста: нет результатов для записи " + "в сводку DQ; проверьте журналы задач build_* этого запуска" + ) return [{"params": {"day": data_date.isoformat()}} for (data_date,) in rows] @@ -239,10 +249,15 @@ def assert_check(check_name: str, diagnostics: list[str]) -> None: logging.error("%s: %s", check_name, detail) summary = "; ".join( f"{data_date}: эталонных ключей {reference_rows}, " - f"фактических {actual_rows}, нарушено {failed_rows}" + f"ключей в проверяемом объекте {actual_rows}, " + f"отсутствуют или различаются {failed_rows}" for data_date, reference_rows, actual_rows, failed_rows in failed_days ) - raise AirflowException(f"{check_name}: {summary}") + raise AirflowException( + f"{check_name}: {summary}. Примеры расхождений — в журнале этой задачи; " + "итоги по дням — в dm.dq_summary_v. Источники каждой сверки описаны " + "в docs/architecture/dm.md" + ) @dag( diff --git a/dags/homework_check.py b/dags/homework_check.py index 9c1d786..a8bc1bc 100644 --- a/dags/homework_check.py +++ b/dags/homework_check.py @@ -45,7 +45,10 @@ def check_mart( """Сверить одну домашнюю витрину с независимым ответом из источника.""" hook = _clickhouse_hook() if not _mart_exists(hook, mart): - raise AirflowException(f"витрина {mart} не найдена; выполните ТЗ {lab_path}") + raise AirflowException( + f"витрина {mart} не найдена на узле подключения clickhouse_default; " + f"проверьте имя и место создания витрины по заданию {lab_path}" + ) try: query = (SQL_ROOT / sql_name).read_text(encoding="utf-8") @@ -72,7 +75,8 @@ def check_mart( except Exception as error: raise AirflowException( f"не удалось проверить витрину {mart}: {error}; " - f"сверьте договор колонок и проверку в ТЗ {lab_path}" + f"сверьте имена и типы колонок с заданием {lab_path}. " + "Подробности ошибки — в журнале этой задачи" ) from error failed_rows = sum(summary[3] for summary in summaries) @@ -93,7 +97,8 @@ def check_mart( compared_sides = diagnostics[0] raise AirflowException( f"витрина {mart} расходится с источником: {days}. " - f"Пример сравнения: {compared_sides}. Сверьтесь с ТЗ {lab_path}" + f"Пример сравнения: {compared_sides}. Другие примеры — в журнале этой " + f"задачи. Проверьте расчет для указанных ключей по заданию {lab_path}" ) diff --git a/dags/world_control.py b/dags/world_control.py index 1b6fb9d..469684c 100644 --- a/dags/world_control.py +++ b/dags/world_control.py @@ -146,25 +146,25 @@ def _ingest_orders() -> TriggerDagRunOperator: type="integer", minimum=1, maximum=7, - title="Сколько дней прожить", + title="Число модельных дней для генерации", # description_md рендерится подсказкой прямо в форме запуска # (params.rst, Airflow 3.3), чтобы правило было видно сразу. description_md=( - "Форма задает только число дней. Номер первого несыгранного " - "дня шаговый даг берёт из переменной `world_position` (Admin → " - "Variables). Поля Logical date и Run ID нужны самому Airflow " - "и на выбор дня не влияют." + "Создать данные за следующие 1–7 модельных дней; по умолчанию " + "за один день. События отправляются без пауз. Номер первого дня " + "берется из `world_position` (Admin → Variables). Поля Logical " + "date и Run ID не меняют выбор дня. После завершения запустите " + "`etl_pipeline`, чтобы обновить DDS и витрины DM." ), ) }, ) def world_next_day(): - """Прожить следующие дни пачкой, без пауз. + """Создать данные за следующие модельные дни без пауз между событиями. - Запускается руками. День по умолчанию один, но разгон вперёд идёт одним - нажимом, а не десятью: сколько дней играть — параметр запуска. Первый - день шаговый даг берёт из `world_position`. Логическая дата прогона на выбор - дня не влияет. + Параметр `days` задает число дней: от 1 до 7, по умолчанию 1. Первый день + берется из `world_position`. После загрузки событий и слепков заказов + запустите `etl_pipeline` вручную, чтобы обновить DDS и витрины DM. """ @task @@ -174,7 +174,10 @@ def world_next_day(): position = first_day + days Variable.set(WORLD_POSITION, str(position)) logging.info( - "дни %s–%s сыграны; новая позиция — %s", first_day, position - 1, position + "данные за модельные дни %s–%s отправлены; world_position=%s", + first_day, + position - 1, + position, ) first_day = first_unplayed_day() @@ -210,20 +213,23 @@ def world_next_day(): tags=TAGS, ) def world_live_day(): - """Прожить следующий день в темпе модельного времени. + """Постепенно отправить события за следующий модельный день. - Ускорение ×60: модельные сутки укладываются примерно в двадцать четыре - реальные минуты, и суточная волна разворачивается на глазах. Запускается - руками; чтобы мир жил так день за днём сам, есть выключатель `world_live`. - Следующий день шаговый даг берёт из `world_position`. Логическая дата прогона - на выбор дня не влияет. + При ускорении ×60 отправка занимает около 24 реальных минут. Номер дня + берется из `world_position`; Logical date его не меняет. Затем даг + отправляет слепок заказов и ждет его загрузки и завершения `etl_pipeline`. + Для автоматического запуска следующих дней снимите с паузы `world_live`. """ @task def remember_played(first_day: int) -> None: """Позиция ставится по сыгранному дню и только по успеху.""" Variable.set(WORLD_POSITION, str(first_day + 1)) - logging.info("сыгран день %s; новая позиция — %s", first_day, first_day + 1) + logging.info( + "данные за модельный день %s отправлены; world_position=%s", + first_day, + first_day + 1, + ) first_day = first_unplayed_day() played = _generator("play_day", ["live", "--day", PLAYED_DAY]) @@ -254,16 +260,12 @@ def world_live_day(): tags=TAGS, ) def world_live(): - """Выключатель: пока включён, мир живёт день за днём. + """Автоматически запускать модельные дни, пока даг снят с паузы. - Своей работы у выключателя нет — он дёргает `world_live_day` и ждёт конца. - Ожидание тут несущая конструкция, а не вежливость: без него тик шёл бы - независимо от хода дня, лишние прогоны скопились бы очередью, и мир потом - промчался бы по ней без всякого темпа. - - Ждём триггером, а не сенсором: оператор опрашивает тот прогон, который сам - и создал, и ссылка на дочерний прогон видна прямо отсюда. Упал день — - краснеет и выключатель. + Запускает `world_live_day` и ждет его завершения. Пауза останавливает + запуск следующих дней; уже запущенный день завершится. Если он завершился + с ошибкой, этот даг тоже завершится с ошибкой: откройте дочерний запуск + и журнал неуспешной задачи. """ TriggerDagRunOperator( task_id="trigger_live_day", diff --git a/infra/grafana/provisioning/dashboards/cluster.json b/infra/grafana/provisioning/dashboards/cluster.json index 0593e7d..4d29c81 100644 --- a/infra/grafana/provisioning/dashboards/cluster.json +++ b/infra/grafana/provisioning/dashboards/cluster.json @@ -11,7 +11,7 @@ "panels": [ { "type": "timeseries", - "title": "Процессор: сколько ядер занимает нода", + "title": "Процессор: сколько ядер использует узел", "datasource": { "type": "grafana-clickhouse-datasource", "uid": "clickhouse" @@ -51,7 +51,7 @@ }, { "type": "timeseries", - "title": "Память: сколько нода занимает и сколько ей отведено", + "title": "Память: использование и лимит узла", "datasource": { "type": "grafana-clickhouse-datasource", "uid": "clickhouse" @@ -114,7 +114,7 @@ }, { "type": "timeseries", - "title": "Диски: сколько нода читает и пишет", + "title": "Диски: скорость чтения и записи узла", "datasource": { "type": "grafana-clickhouse-datasource", "uid": "clickhouse" @@ -154,7 +154,7 @@ }, { "type": "table", - "title": "Части: сколько их у таблицы и по сколько строк в каждой", + "title": "Части таблиц: количество и среднее число строк в части", "datasource": { "type": "grafana-clickhouse-datasource", "uid": "clickhouse" @@ -199,7 +199,7 @@ }, { "type": "timeseries", - "title": "Слияния: сколько частей они съедают", + "title": "Слияния: число объединенных частей", "datasource": { "type": "grafana-clickhouse-datasource", "uid": "clickhouse" @@ -280,11 +280,12 @@ "min": 0 }, "overrides": [] - } + }, + "description": "Узлы дерева Keeper (znode) — записи со служебными данными координации ClickHouse." }, { "type": "table", - "title": "Распределённый DDL: что прошло по кластеру и чем кончилось", + "title": "Распределенный DDL: результаты команд на узлах кластера", "datasource": { "type": "grafana-clickhouse-datasource", "uid": "clickhouse" diff --git a/infra/grafana/provisioning/dashboards/data.json b/infra/grafana/provisioning/dashboards/data.json index 1c3e16e..cf94d2e 100644 --- a/infra/grafana/provisioning/dashboards/data.json +++ b/infra/grafana/provisioning/dashboards/data.json @@ -11,7 +11,7 @@ "panels": [ { "type": "stat", - "title": "Свежесть: сколько прошло с последней доехавшей строки", + "title": "Свежесть: время с последней загрузки в ODS", "datasource": { "type": "grafana-clickhouse-datasource", "uid": "clickhouse" @@ -48,7 +48,7 @@ }, { "type": "table", - "title": "Брак: сколько приехавшего не разобралось", + "title": "Брак: доля строк с ошибками разбора", "datasource": { "type": "grafana-clickhouse-datasource", "uid": "clickhouse" @@ -72,11 +72,28 @@ "format": 1, "rawSql": "SELECT\n source AS \"Источник\",\n countIf(NOT is_error) AS \"Приехало\",\n countIf(is_error) AS \"Из них в брак\",\n 100 * countIf(is_error) / countIf(NOT is_error) AS \"Доля брака, %\"\nFROM\n(\n SELECT 'события' AS source, 0 AS is_error, _load_ts FROM stg.hits_raw_dist\n UNION ALL\n SELECT 'события', 1, _load_ts FROM ods.event_errors_dist\n UNION ALL\n SELECT 'заказы', 0, _load_ts FROM stg.orders_raw_dist\n UNION ALL\n SELECT 'заказы', 1, _load_ts FROM ods.order_errors_dist\n)\nWHERE $__timeFilter(_load_ts)\nGROUP BY source\nORDER BY source" } - ] + ], + "fieldConfig": { + "defaults": {}, + "overrides": [ + { + "matcher": { + "id": "byName", + "options": "Приехало" + }, + "properties": [ + { + "id": "displayName", + "value": "Загружено в STG" + } + ] + } + ] + } }, { "type": "table", - "title": "Дневной слепок заказов: какой доехал, когда и после какой паузы", + "title": "Дневной слепок заказов: дата и время загрузки", "datasource": { "type": "grafana-clickhouse-datasource", "uid": "clickhouse" @@ -100,11 +117,28 @@ "format": 1, "rawSql": "SELECT\n \"Слепок\",\n \"Доехал\",\n dateDiff('second', prev, \"Доехал\") AS \"После прошлого, с\",\n \"Строк\"\nFROM\n(\n SELECT\n JSONExtractString(raw, 'snapshot_date') AS \"Слепок\",\n max(_load_ts) AS \"Доехал\",\n count() AS \"Строк\",\n -- Обнуляемый тип нужен, чтобы у самого раннего слепка пауза осталась\n -- пустой, а не превратилась в разницу с началом эпохи.\n lagInFrame(toNullable(max(_load_ts))) OVER (ORDER BY \"Слепок\") AS prev\n FROM stg.orders_raw_dist\n GROUP BY \"Слепок\"\n)\nORDER BY \"Слепок\" DESC\nLIMIT 10" } - ] + ], + "fieldConfig": { + "defaults": {}, + "overrides": [ + { + "matcher": { + "id": "byName", + "options": "Доехал" + }, + "properties": [ + { + "id": "displayName", + "value": "Время загрузки" + } + ] + } + ] + } }, { "type": "timeseries", - "title": "Поток: сколько строк приезжает в сырой слой", + "title": "Поток: число сообщений, загруженных в STG", "datasource": { "type": "grafana-clickhouse-datasource", "uid": "clickhouse" @@ -136,7 +170,7 @@ }, { "type": "timeseries", - "title": "Сходимость: произведено в Kafka против приземлившегося в ClickHouse", + "title": "Сходимость: число событий в Kafka и ClickHouse", "datasource": { "type": "datasource", "uid": "-- Mixed --" @@ -161,7 +195,7 @@ }, "editorMode": "code", "expr": "sum(kafka_topic_partition_current_offset{topic=\"hits\"}) - sum(kafka_topic_partition_current_offset{topic=\"hits\"} offset $__interval)", - "legendFormat": "произведено в Kafka", + "legendFormat": "поступило в Kafka", "range": true }, { @@ -187,7 +221,7 @@ "properties": [ { "id": "displayName", - "value": "приземлилось в ClickHouse" + "value": "загружено в ClickHouse" } ] } @@ -197,7 +231,7 @@ }, { "type": "timeseries", - "title": "Отставание чтения из Kafka: непрочитанных сообщений", + "title": "Отставание чтения из Kafka: число непрочитанных сообщений", "datasource": { "type": "prometheus", "uid": "prometheus" diff --git a/infra/grafana/provisioning/dashboards/queries.json b/infra/grafana/provisioning/dashboards/queries.json index daf9e23..2c67b87 100644 --- a/infra/grafana/provisioning/dashboards/queries.json +++ b/infra/grafana/provisioning/dashboards/queries.json @@ -11,7 +11,7 @@ "panels": [ { "type": "table", - "title": "Тяжёлые запросы: что прочитало больше всех", + "title": "Запросы с наибольшим числом прочитанных строк", "datasource": { "type": "grafana-clickhouse-datasource", "uid": "clickhouse" @@ -56,7 +56,7 @@ }, { "type": "table", - "title": "Кто спрашивает: сколько запросов и какой у них аппетит", + "title": "Пользователи: число запросов и потребление ресурсов", "datasource": { "type": "grafana-clickhouse-datasource", "uid": "clickhouse" @@ -93,6 +93,10 @@ { "id": "unit", "value": "bytes" + }, + { + "id": "displayName", + "value": "Максимальная память запроса" } ] } @@ -101,7 +105,7 @@ }, { "type": "table", - "title": "Ошибки: на чём запросы споткнулись", + "title": "Ошибки выполнения запросов", "datasource": { "type": "grafana-clickhouse-datasource", "uid": "clickhouse" @@ -128,7 +132,20 @@ ], "fieldConfig": { "defaults": {}, - "overrides": [] + "overrides": [ + { + "matcher": { + "id": "byName", + "options": "Чем кончилось" + }, + "properties": [ + { + "id": "displayName", + "value": "Ошибка" + } + ] + } + ] } } ] diff --git a/infra/superset/import/charts/purchase_vs_orders.yaml b/infra/superset/import/charts/purchase_vs_orders.yaml index 2f41e09..9a82e42 100644 --- a/infra/superset/import/charts/purchase_vs_orders.yaml +++ b/infra/superset/import/charts/purchase_vs_orders.yaml @@ -1,4 +1,4 @@ -slice_name: Сверка purchase и заказов +slice_name: Сверка событий покупки и заказов description: null certified_by: null certification_details: null diff --git a/infra/superset/import/dashboards/business_stories.yaml b/infra/superset/import/dashboards/business_stories.yaml index 59ef41e..45a59e7 100644 --- a/infra/superset/import/dashboards/business_stories.yaml +++ b/infra/superset/import/dashboards/business_stories.yaml @@ -153,7 +153,7 @@ position: chartId: 2 height: 44 width: 12 - sliceName: Сверка purchase и заказов + sliceName: Сверка событий покупки и заказов uuid: 19600000-0000-4000-8000-000000000102 ROW-traffic: id: ROW-traffic diff --git a/infra/superset/import/datasets/ClickHouse/daily_traffic_v.yaml b/infra/superset/import/datasets/ClickHouse/daily_traffic_v.yaml index 8b126f9..64fe498 100644 --- a/infra/superset/import/datasets/ClickHouse/daily_traffic_v.yaml +++ b/infra/superset/import/datasets/ClickHouse/daily_traffic_v.yaml @@ -37,7 +37,7 @@ metrics: extra: null warning_text: null - metric_name: known_share - verbose_name: Доля известных + verbose_name: Доля известных пользователей metric_type: null expression: SUM(known_users) / nullIf(SUM(visitors), 0) description: null diff --git a/infra/superset/import/datasets/ClickHouse/purchase_vs_orders_v.yaml b/infra/superset/import/datasets/ClickHouse/purchase_vs_orders_v.yaml index 52cceab..7231931 100644 --- a/infra/superset/import/datasets/ClickHouse/purchase_vs_orders_v.yaml +++ b/infra/superset/import/datasets/ClickHouse/purchase_vs_orders_v.yaml @@ -55,7 +55,7 @@ columns: datetime_format: null extra: null - column_name: declared_revenue - verbose_name: Выручка purchase + verbose_name: Заявленная сумма покупки из события is_dttm: false is_active: true type: NULLABLE(DECIMAL(18, 2)) @@ -120,7 +120,7 @@ columns: datetime_format: null extra: null - column_name: order_id - verbose_name: Заказ + verbose_name: Номер заказа is_dttm: false is_active: true type: STRING diff --git a/infra/superset/import/datasets/ClickHouse/revenue_daily_v.yaml b/infra/superset/import/datasets/ClickHouse/revenue_daily_v.yaml index 1c47107..e8c9acf 100644 --- a/infra/superset/import/datasets/ClickHouse/revenue_daily_v.yaml +++ b/infra/superset/import/datasets/ClickHouse/revenue_daily_v.yaml @@ -125,7 +125,7 @@ columns: datetime_format: null extra: null - column_name: units - verbose_name: Товары + verbose_name: Продано единиц товара is_dttm: false is_active: true type: UINT64 diff --git a/scripts/run-dag.sh b/scripts/run-dag.sh index 4eba13f..ef90bf1 100755 --- a/scripts/run-dag.sh +++ b/scripts/run-dag.sh @@ -24,10 +24,10 @@ compose() { next_step() { case "$DAG_ID" in world_initialize) - printf 'посмотрите журнал; повторите make up, а для недействительного мира — make rebuild-storage' + printf 'проверьте состояние запуска в Airflow и журнал задачи с ошибкой; после устранения причины повторите make up' ;; - world_recreate) printf 'устраните причину и повторите make rebuild-storage' ;; - *) printf 'проверьте журнал дага и повторите команду' ;; + world_recreate) printf 'проверьте состояние запуска в Airflow и журнал задачи с ошибкой; после устранения причины повторите make rebuild-storage' ;; + *) printf 'проверьте состояние запуска в Airflow и журнал задачи с ошибкой; после устранения причины повторите команду' ;; esac } @@ -67,7 +67,7 @@ for ((attempt = 1; attempt <= ATTEMPTS; attempt++)); do done jq -e --arg dag_id "$DAG_ID" '.dag_id == $dag_id' \ >/dev/null 2>&1 <<<"$dag" || \ - fail "даг ${DAG_ID} не появился за $((ATTEMPTS * PAUSE_SECONDS)) с; $(next_step)" + fail "даг ${DAG_ID} не появился в API за $((ATTEMPTS * PAUSE_SECONDS)) с; проверьте ошибки импорта в Airflow и журнал службы airflow-dag-processor" # GET читает DagModel по идентификатору, а POST запуска принимает только запись # с `is_stale = false`. На холодном старте GET уже может видеть даг, пока POST @@ -105,7 +105,7 @@ for ((attempt = 1; attempt <= TRIGGER_ATTEMPTS; attempt++)); do fail "Airflow не запустил даг ${DAG_ID}: HTTP ${http_status}${detail:+: ${detail}}; $(next_step)" done ((triggered == 1)) || \ - fail "даг ${DAG_ID} не стал доступен для запуска за $((TRIGGER_ATTEMPTS * PAUSE_SECONDS)) с; $(next_step)" + fail "даг ${DAG_ID} не стал доступен для запуска за $((TRIGGER_ATTEMPTS * PAUSE_SECONDS)) с; проверьте ошибки импорта в Airflow и журнал службы airflow-dag-processor" run_id="$(jq -r '.dag_run_id // empty' <<<"$response")" state="$(jq -r '.state // "queued"' <<<"$response")" [[ -n "$run_id" ]] || fail "Airflow не вернул идентификатор запуска ${DAG_ID}" @@ -121,7 +121,7 @@ for ((attempt = 1; attempt <= ATTEMPTS; attempt++)); do state="$(jq -r '.state // empty' <<<"$response" 2>/dev/null || true)" case "$state" in success) - printf 'ЗЕЛЁНО: даг %s, запуск %s, состояние success.\n' "$DAG_ID" "$run_id" + printf 'УСПЕХ: даг %s, запуск %s, состояние success.\n' "$DAG_ID" "$run_id" exit 0 ;; failed)