ddmitry/clickstream-ch-kafka-superset-demo

Files

T

ddadmin dd4af822d2 feat(airflow): добавлен пульт управления генератором

- Зачем:
  - нужен основной ручной интерфейс стенда для backfill/import/check без консольной матрицы переменных.
- Что:
  - добавлен DAG generator_control с параметрами Airflow, ветвлением операций и ожиданием ETL.
  - вынесена общая логика запуска и предпроверок генератора для Airflow.
  - обновлены compose-настройки, зависимости, тесты и документация по пульту.
- Проверка:
  - uv run --with pytest --with-requirements generator/requirements.txt pytest generator/tests -q.
  - docker compose config --quiet.

2026-07-04 21:06:30 +03:00

7.8 KiB

Raw Blame History

Runbook: стартовая история стенда

Этот runbook нужен, чтобы один раз создать стартовую историю генератора, сохранить её в файл и быстро восстановить на чистом стенде.

За устройством генератора см. generator/README.md и спеку модельного времени.

Что дёшево

Перезапуск без очистки volumes: Kafka хранит generator_state и manifest.
Live-продолжение после импорта: генератор стартует с T_end, если настройки совпадают.
Восстановление чистого стенда из готового файла: события повторно пишутся в Kafka, ClickHouse наполняется штатным путём.

Что требует нового артефакта

Другая длительность истории или другой профиль запуска.
Другой GEN_SEED, GEN_MODEL_T0, часовой пояс, скорость или настройки генерации.
Осознанный новый мир после несовместимого state: сначала сбросьте state через GEN_STATE_RESET=true или чистые volumes.

Профили запуска

Основной способ — глагол плюс профиль:

Профиль	Для чего	Длительность
`ci`	Быстрая проверка и CI	`6h`
`daily-wave`	История с видимой суточной волной	`2d`

Длительность можно переопределить через GEN_HISTORY_DURATION, например 2d. Команда сама считает GEN_MODEL_T_END от GEN_MODEL_T0.

Пульт в Airflow

Основной ручной путь — DAG generator_control в Airflow UI:

Поднимите стенд: make up.
Если DDL ещё не применён, запустите ddl_init.
Откройте generator_control и выберите operation.

Операции:

backfill — создать стартовую историю. После записи в Kafka DAG сам запускает etl_pipeline, ждёт завершения и выполняет check.
import — прочитать артефакт из artifact_path. Несовместимый артефакт отклоняется до записи в Kafka.
check — сверить ClickHouse с manifest из Kafka.

Поля формы:

profile берётся из профилей генератора.
duration можно оставить пустым, тогда берётся длительность профиля.
seed и model_time_speed — необязательные переопределения мира.
artifact_path: для backfill — куда сохранить файл; пусто — не сохранять. Для import — что читать; пусто — /opt/airflow/data/startup-history-import.json.

Backfill и import работают только на чистом стенде. Если Kafka-топики данных или STG уже непустые, DAG упадёт до записи и подскажет make clean. Это защита от смешивания разных миров.

Границы пульта:

make up, make clean и live-продолжение остаются в консоли.
Операции continue в DAG нет намеренно: live — долгоживущий сервис, а пульт управляет разовыми пакетными операциями.
Airflow не получает доступ к жизненному циклу контейнеров; таски выполняют обычный Python-код генератора.

Если backfill сохраняет файл в ./data, он создаётся пользователем Airflow внутри контейнера. Чтение работает из Airflow и консольных команд, но перезапись чужого файла может потребовать удалить старый файл вручную.

Экспорт

По умолчанию создаётся быстрый 6-часовой артефакт:

make startup-history-export

Файл по умолчанию: /tmp/clickstream-startup-history.json.

История на 2 суток с суточной волной:

ARTIFACT=/tmp/clickstream-startup-history-2d.json \
PROFILE=daily-wave \
make startup-history-export

Команда делает чистый backfill и пишет в файл один связный набор: события Kafka, state и manifest.

Импорт на чистый стенд

make clean
docker compose up -d clickhouse kafka
make ddl

ARTIFACT=/tmp/clickstream-startup-history.json make startup-history-import

sleep 10
make transform
ARTIFACT=/tmp/clickstream-startup-history.json make startup-history-check
make generated-history-check

Импорт не пишет напрямую в ClickHouse. Он воспроизводит события и служебные compact-топики в Kafka. ClickHouse читает данные через свои Kafka-таблицы и Materialized View, затем batch строит ODS, DDS и DM.

Если артефакт создан не профилем ci, импорт запускайте с тем же профилем:

PROFILE=daily-wave ARTIFACT=/tmp/clickstream-startup-history-2d.json \
make startup-history-import

make startup-history-check сверяет контрольные числа DM-витрины с manifest артефакта: события, визиты, пользователей и диапазон event_timestamp. Если data-топики Kafka уже непустые, импорт остановится до публикации событий.

В текущем стеке kafka-python не даёт транзакционный producer для нескольких топиков. Поэтому импорт остаётся clean-stand операцией: при ошибке записи он удаляет import-топики Kafka, чтобы повторный импорт не дописал дубли. Если ClickHouse уже успел прочитать частичные сообщения, очистите стенд через make clean и повторите импорт.

Live-продолжение

После импорта запускайте live с тем же профилем, что был в артефакте:

PROFILE=ci make generator-continue

Если читаемый state есть, но настройки не совпадают, генератор падает с перечнем полей. Это защита от смешения разных миров. Для намеренного нового мира используйте make generator-reset или make clean.

После нестандартного мира make generator-continue нужно запускать с теми же настройками, что были у backfill/import. При расхождении генератор громко покажет поля, которые не совпали.

Старые переменные GEN_RUN_MODE, GEN_STATE_RESET и GEN_MODEL_T_END остаются низкоуровневым способом для отладки и прямого docker compose run.

7.8 KiB Raw Blame History Unescape Escape