From 10f5bc3510c12db23e5f695e3615ce73c05d05a7 Mon Sep 17 00:00:00 2001 From: Dmitry Dementev Date: Sun, 8 Feb 2026 18:13:22 +0300 Subject: [PATCH 1/5] =?UTF-8?q?feat(airflow):=20=D1=80=D0=B5=D0=B0=D0=BB?= =?UTF-8?q?=D0=B8=D0=B7=D0=BE=D0=B2=D0=B0=D0=BD=20DAG=20kafka=5Fload=20?= =?UTF-8?q?=D0=B4=D0=BB=D1=8F=20=D0=B7=D0=B0=D0=B3=D1=80=D1=83=D0=B7=D0=BA?= =?UTF-8?q?=D0=B8=20=D0=B2=20Kafka=20(=D1=84=D0=B0=D0=B7=D0=B0=202)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Добавлен kafka-python==2.0.6 в airflow/requirements.txt - Создан dags/utils/kafka_helpers.py с функциями: - check_kafka_ready() — проверка доступности брокера - prepare_topics() — создание/сброс топиков через KafkaAdminClient - load_jsonl() — загрузка данных через KafkaProducer (limit=0 = все) - validate_load_params(), check_input_files() — валидация - Создан dags/kafka_load_dag.py с TaskGroup: - precheck: check_kafka, check_input_files, validate_load_params - ingest: prepare_topics, параллельная загрузка 4 потоков, verify_publish_counts - Параметры DAG: limit (0 = все), reset_topics, load_* (выбор потоков) - Обновлена документация: AGENTS.md, README.md, plans/runbook.md, plans/airflow_dags_plan.md, docs/ARCHITECTURE.md Тестирование: - Подключение к Kafka: ✅ (kafka:29092 доступен, брокер 2.6.0) - Загрузка данных: ✅ (1000 сообщений — полный файл browser_events) - Python синтаксис: ✅ (py_compile проходит) - Структура DAG: ✅ (все 9 задач корректно определены) --- .gitignore | 1 + AGENTS.md | 60 ++++++- README.md | 18 +- airflow/requirements.txt | 3 + dags/kafka_load_dag.py | 322 +++++++++++++++++++++++++++++++++ dags/utils/__init__.py | 1 + dags/utils/kafka_helpers.py | 342 ++++++++++++++++++++++++++++++++++++ docs/ARCHITECTURE.md | 21 ++- plans/airflow_dags_plan.md | 3 +- plans/runbook.md | 23 ++- 10 files changed, 778 insertions(+), 16 deletions(-) create mode 100644 .gitignore create mode 100644 dags/kafka_load_dag.py create mode 100644 dags/utils/__init__.py create mode 100644 dags/utils/kafka_helpers.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..bee8a64 --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +__pycache__ diff --git a/AGENTS.md b/AGENTS.md index f479d88..8f72fdd 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -20,7 +20,11 @@ ## Ключевые артефакты ### Исполняемые файлы (текущая структура) -- `dags/` — Airflow DAGs для оркестрации ETL +- `dags/` — Airflow DAGs для оркестрации ETL: + - `dags/ddl_init_dag.py` — инициализация схемы ClickHouse + - `dags/etl_pipeline_dag.py` — ETL процесс STG → ODS → DDS → DM + - `dags/kafka_load_dag.py` — загрузка данных в Kafka из JSONL + - `dags/utils/kafka_helpers.py` — helper-функции для работы с Kafka - `sql/` — SQL по слоям: - `sql/ddl/00_databases.sql` — создание БД stg/ods/dds/dm - `sql/ddl/stg/10_stg.sql` — STG слой (Kafka Engine + MV) @@ -92,13 +96,65 @@ - Bash: шапка с назначением/запуском/требованиями, секции разделены `# -----` - См. существующие файлы как пример (`sql/ddl/ods/20_ods.sql`, `sql/dds/30_ods_to_dds.sql`, `scripts/run_batch.sh`) +## Airflow DAGs + +### `ddl_init` — Инициализация схемы +- Запуск: ручной (Trigger DAG) +- Параметры: `verify_only` (bool, default false) +- Описание: Создаёт БД и таблицы в ClickHouse от 00_databases до 40_dm + +### `kafka_load` — Загрузка в Kafka +- Запуск: ручный (Trigger DAG with config) +- Параметры: + - `limit` (int, default 0) — количество строк (0 = все) + - `reset_topics` (bool, default true) — пересоздать топики + - `load_browser/load_location/load_device/load_geo` — выбор потоков +- Примеры запуска: + ```json + // Полная загрузка (по умолчанию) + {} + // Ограниченная загрузка — 100 строк + {"limit": 100} + // Только browser_events + {"limit": 100, "load_location": false, "load_device": false, "load_geo": false} + ``` + +### `etl_pipeline` — ETL процесс +- Запуск: ручной (Trigger DAG with config) +- Параметры: + - `full_refresh` (bool, default true) — очистить DDS перед загрузкой +- Зависимость: требует наличия данных в STG (от `kafka_load` или `make data`) + ## Быстрые проверки - Kafka ingest: наличие данных в `stg.*` и типизированных строк в `ods.*`. - Мониторинг: доступность `/metrics` у ClickHouse и скрейп в Prometheus. -- **Airflow: `http://localhost:8080` должен показывать UI и DAG `ddl_init` и `etl_pipeline`.** +- **Airflow: `http://localhost:8080` должен показывать UI и DAG `ddl_init`, `kafka_load` и `etl_pipeline`.** - BI: витрина `dm.v_events_enriched` должна отвечать за разумное время при фильтре по дате. +## Сценарий работы с Airflow (фаза 2) + +```bash +# 1. Запуск инфраструктуры +make up + +# 2. Инициализация схемы (один раз) +# Airflow UI → DAGs → ddl_init → Trigger DAG + +# 3. Загрузка данных через Airflow (вместо make data) +# Airflow UI → DAGs → kafka_load → Trigger DAG with config +# Параметры по умолчанию: limit=50, все потоки + +# 4. Запуск ETL +# Airflow UI → DAGs → etl_pipeline → Trigger DAG with config +# {"full_refresh": true} + +# 5. Проверка результатов +docker compose exec -T clickhouse clickhouse-client --user=default --password=123456 --query "SELECT count() FROM ods.browser_event" +docker compose exec -T clickhouse clickhouse-client --user=default --password=123456 --query "SELECT count() FROM dds.event" +docker compose exec -T clickhouse clickhouse-client --user=default --password=123456 --query "SELECT * FROM dm.dq_summary" +``` + ## Связанная документация - [README.md](./README.md) — пользовательская документация (быстрый старт, архитектура) diff --git a/README.md b/README.md index af8e27e..1cf85fa 100644 --- a/README.md +++ b/README.md @@ -33,10 +33,18 @@ docker compose ps docker compose exec -T airflow-webserver airflow dags trigger ddl_init ``` -Загрузка небольшого среза данных в Kafka: +Загрузка данных в Kafka (фаза 2 — через Airflow): ```bash -make data # по умолчанию первые 50 строк -# или: FULL=1 make data # полный датасет (1000 строк) +# Вариант 1: Через Airflow DAG (рекомендуется) — полная загрузка по умолчанию +docker compose exec -T airflow-webserver airflow dags trigger kafka_load \ + --conf '{"reset_topics": true}' + +# Ограниченная загрузка — первые 100 строк +docker compose exec -T airflow-webserver airflow dags trigger kafka_load \ + --conf '{"limit": 100, "reset_topics": true}' + +# Вариант 2: Через shell-скрипт (устаревший) +make data # полная загрузка ``` Запуск batch-трансформации (STG -> ODS -> DDS -> DM) в Airflow (если DAG выключен, сначала unpause): @@ -95,8 +103,8 @@ flowchart TB end subgraph Airflow["Airflow"] - DAG[DAG: ddl_init / etl_pipeline] - end + DAG[DAG: ddl_init / kafka_load / etl_pipeline] + end Sources -->|make data| Kafka -->|MV| STG -->|Batch SQL| ODS -->|Batch SQL| DDS -->|VIEW| DM DAG -.->|оркестрация| STG & ODS & DDS & DM diff --git a/airflow/requirements.txt b/airflow/requirements.txt index feae4fb..9504b3c 100644 --- a/airflow/requirements.txt +++ b/airflow/requirements.txt @@ -5,3 +5,6 @@ psycopg2-binary==2.9.9 # ClickHouse operator/hook для DAG'ов airflow-clickhouse-plugin==1.6.0 + +# Kafka client для загрузки данных (DAG kafka_load) +kafka-python==2.0.6 diff --git a/dags/kafka_load_dag.py b/dags/kafka_load_dag.py new file mode 100644 index 0000000..7baa4ce --- /dev/null +++ b/dags/kafka_load_dag.py @@ -0,0 +1,322 @@ +""" +DAG для загрузки данных в Kafka из JSONL-файлов. + +Функциональность: +- Проверка доступности Kafka и наличия файлов +- Создание/сброс топиков +- Параллельная загрузка 4 потоков данных +- Проверка результатов через XCom + +Параметры (через Trigger DAG with config): + limit (int): Количество строк для загрузки (по умолчанию 50) + full_load (bool): Загрузить всё, игнорируя limit (по умолчанию False) + reset_topics (bool): Пересоздать топики (по умолчанию True) + load_browser (bool): Загружать browser_events (по умолчанию True) + load_location (bool): Загружать location_events (по умолчанию True) + load_device (bool): Загружать device_events (по умолчанию True) + load_geo (bool): Загружать geo_events (по умолчанию True) +""" + +from __future__ import annotations + +from datetime import datetime, timedelta +from pathlib import Path + +from airflow import DAG +from airflow.exceptions import AirflowException +from airflow.models.param import Param +from airflow.operators.python import PythonOperator +from airflow.utils.task_group import TaskGroup + +# Импортируем helper-функции +from utils.kafka_helpers import ( + check_input_files, + check_kafka_ready, + get_topic_file_mapping, + load_jsonl, + prepare_topics, + validate_load_params, +) + +# ----------------------------------------------------------------------------- +# Базовые настройки DAG +# ----------------------------------------------------------------------------- +default_args = { + "owner": "airflow", + "depends_on_past": False, + "email_on_failure": False, + "email_on_retry": False, + "retries": 1, + "retry_delay": timedelta(minutes=2), +} + +DATA_DIR = Path("/opt/airflow/data") + +# ----------------------------------------------------------------------------- +# Python callable функции для задач +# ----------------------------------------------------------------------------- +def _check_kafka(**context) -> None: + """Проверка доступности Kafka брокера.""" + check_kafka_ready() + + +def _check_files(**context) -> None: + """Проверка наличия входных файлов.""" + conf = context.get("dag_run", {}).conf or {} + load_browser = bool(conf.get("load_browser", context["params"]["load_browser"])) + load_location = bool(conf.get("load_location", context["params"]["load_location"])) + load_device = bool(conf.get("load_device", context["params"]["load_device"])) + load_geo = bool(conf.get("load_geo", context["params"]["load_geo"])) + + check_input_files( + data_dir=DATA_DIR, + load_browser=load_browser, + load_location=load_location, + load_device=load_device, + load_geo=load_geo, + ) + + +def _validate_params(**context) -> None: + """Валидация параметров загрузки.""" + conf = context.get("dag_run", {}).conf or {} + limit = int(conf.get("limit", context["params"]["limit"])) + load_browser = bool(conf.get("load_browser", context["params"]["load_browser"])) + load_location = bool(conf.get("load_location", context["params"]["load_location"])) + load_device = bool(conf.get("load_device", context["params"]["load_device"])) + load_geo = bool(conf.get("load_geo", context["params"]["load_geo"])) + + validate_load_params( + limit=limit, + load_browser=load_browser, + load_location=load_location, + load_device=load_device, + load_geo=load_geo, + ) + + +def _prepare_topics(**context) -> None: + """Подготовка топиков Kafka (создание/сброс).""" + conf = context.get("dag_run", {}).conf or {} + reset_topics = bool(conf.get("reset_topics", context["params"]["reset_topics"])) + load_browser = bool(conf.get("load_browser", context["params"]["load_browser"])) + load_location = bool(conf.get("load_location", context["params"]["load_location"])) + load_device = bool(conf.get("load_device", context["params"]["load_device"])) + load_geo = bool(conf.get("load_geo", context["params"]["load_geo"])) + + # Получаем список топиков для загрузки + mapping = get_topic_file_mapping( + load_browser=load_browser, + load_location=load_location, + load_device=load_device, + load_geo=load_geo, + ) + topics = list(mapping.keys()) + + prepare_topics(topics=topics, reset=reset_topics) + + +def _load_events(event_type: str, **context) -> int: + """ + Загружает события определённого типа в Kafka. + + Args: + event_type: Тип события (browser, location, device, geo) + + Returns: + Количество отправленных сообщений + """ + conf = context.get("dag_run", {}).conf or {} + limit = int(conf.get("limit", context["params"]["limit"])) + + # Маппинг типа события на топик и файл + event_mapping = { + "browser": ("browser_events", "browser_events.jsonl"), + "location": ("location_events", "location_events.jsonl"), + "device": ("device_events", "device_events.jsonl"), + "geo": ("geo_events", "geo_events.jsonl"), + } + + topic, filename = event_mapping[event_type] + file_path = DATA_DIR / filename + + # Загружаем данные + sent_count = load_jsonl( + file_path=file_path, + topic=topic, + limit=limit, + ) + + # Сохраняем результат в XCom для verify_publish_counts + context["ti"].xcom_push(key=f"{event_type}_count", value=sent_count) + + return sent_count + + +def _verify_counts(**context) -> None: + """Проверяет, что все загрузки отправили сообщения.""" + conf = context.get("dag_run", {}).conf or {} + load_browser = bool(conf.get("load_browser", context["params"]["load_browser"])) + load_location = bool(conf.get("load_location", context["params"]["load_location"])) + load_device = bool(conf.get("load_device", context["params"]["load_device"])) + load_geo = bool(conf.get("load_geo", context["params"]["load_geo"])) + + ti = context["ti"] + + # Собираем результаты из XCom + results = {} + total_sent = 0 + + if load_browser: + count = ti.xcom_pull(task_ids="ingest.load_browser_events", key="browser_count") + results["browser_events"] = count or 0 + total_sent += count or 0 + + if load_location: + count = ti.xcom_pull(task_ids="ingest.load_location_events", key="location_count") + results["location_events"] = count or 0 + total_sent += count or 0 + + if load_device: + count = ti.xcom_pull(task_ids="ingest.load_device_events", key="device_count") + results["device_events"] = count or 0 + total_sent += count or 0 + + if load_geo: + count = ti.xcom_pull(task_ids="ingest.load_geo_events", key="geo_count") + results["geo_events"] = count or 0 + total_sent += count or 0 + + # Проверяем, что отправлено хотя бы что-то + if total_sent == 0: + raise AirflowException("Не отправлено ни одного сообщения ни в один топик") + + # Логируем итоговую статистику + for topic, count in results.items(): + print(f"✓ {topic}: {count} сообщений") + print(f"\nВсего отправлено: {total_sent} сообщений") + + +# ----------------------------------------------------------------------------- +# Определение DAG +# ----------------------------------------------------------------------------- +with DAG( + dag_id="kafka_load", + default_args=default_args, + description="Загрузка данных из JSONL в Kafka топики", + schedule=None, # Только ручной запуск + start_date=datetime(2024, 1, 1), + catchup=False, + max_active_runs=1, + is_paused_upon_creation=True, + tags=["kafka", "ingest", "experiments"], + params={ + "limit": Param( + default=0, + type="integer", + description="Количество строк для загрузки (0 = все строки, по умолчанию)", + ), + "full_load": Param( + default=False, + type="boolean", + description="Загрузить все данные, игнорируя limit", + ), + "reset_topics": Param( + default=True, + type="boolean", + description="Пересоздать топики перед загрузкой", + ), + "load_browser": Param( + default=True, + type="boolean", + description="Загружать browser_events", + ), + "load_location": Param( + default=True, + type="boolean", + description="Загружать location_events", + ), + "load_device": Param( + default=True, + type="boolean", + description="Загружать device_events", + ), + "load_geo": Param( + default=True, + type="boolean", + description="Загружать geo_events", + ), + }, +) as dag: + + # ------------------------------------------------------------------------- + # TaskGroup: precheck — проверки перед загрузкой + # ------------------------------------------------------------------------- + with TaskGroup(group_id="precheck") as precheck: + check_kafka = PythonOperator( + task_id="check_kafka", + python_callable=_check_kafka, + ) + + check_input_files_task = PythonOperator( + task_id="check_input_files", + python_callable=_check_files, + ) + + validate_params = PythonOperator( + task_id="validate_load_params", + python_callable=_validate_params, + ) + + check_kafka >> check_input_files_task >> validate_params + + # ------------------------------------------------------------------------- + # TaskGroup: ingest — загрузка данных + # ------------------------------------------------------------------------- + with TaskGroup(group_id="ingest") as ingest: + prepare_topics_task = PythonOperator( + task_id="prepare_topics", + python_callable=_prepare_topics, + ) + + load_browser_events = PythonOperator( + task_id="load_browser_events", + python_callable=_load_events, + op_kwargs={"event_type": "browser"}, + ) + + load_location_events = PythonOperator( + task_id="load_location_events", + python_callable=_load_events, + op_kwargs={"event_type": "location"}, + ) + + load_device_events = PythonOperator( + task_id="load_device_events", + python_callable=_load_events, + op_kwargs={"event_type": "device"}, + ) + + load_geo_events = PythonOperator( + task_id="load_geo_events", + python_callable=_load_events, + op_kwargs={"event_type": "geo"}, + ) + + verify_publish_counts = PythonOperator( + task_id="verify_publish_counts", + python_callable=_verify_counts, + ) + + # Зависимости: подготовка -> параллельная загрузка -> проверка + prepare_topics_task >> [ + load_browser_events, + load_location_events, + load_device_events, + load_geo_events, + ] >> verify_publish_counts + + # ------------------------------------------------------------------------- + # Итоговая цепочка + # ------------------------------------------------------------------------- + precheck >> ingest diff --git a/dags/utils/__init__.py b/dags/utils/__init__.py new file mode 100644 index 0000000..29a1ee3 --- /dev/null +++ b/dags/utils/__init__.py @@ -0,0 +1 @@ +# Utils package для Airflow DAG'ов diff --git a/dags/utils/kafka_helpers.py b/dags/utils/kafka_helpers.py new file mode 100644 index 0000000..47988a5 --- /dev/null +++ b/dags/utils/kafka_helpers.py @@ -0,0 +1,342 @@ +""" +Helper-функции для работы с Kafka из Airflow DAG'ов. + +Использует kafka-python: +- KafkaAdminClient — для управления топиками +- KafkaProducer — для публикации сообщений +""" + +from __future__ import annotations + +import logging +from pathlib import Path +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from typing import Optional + +# ----------------------------------------------------------------------------- +# Конфигурация подключения к Kafka +# ----------------------------------------------------------------------------- +KAFKA_BOOTSTRAP_SERVERS = "kafka:29092" +REQUEST_TIMEOUT_MS = 30000 + +# Топики и соответствующие файлы данных +TOPIC_FILE_MAP = { + "browser_events": "browser_events.jsonl", + "location_events": "location_events.jsonl", + "device_events": "device_events.jsonl", + "geo_events": "geo_events.jsonl", +} + +logger = logging.getLogger(__name__) + + +# ----------------------------------------------------------------------------- +# Проверка доступности Kafka +# ----------------------------------------------------------------------------- +def check_kafka_ready( + bootstrap_servers: str = KAFKA_BOOTSTRAP_SERVERS, + timeout_ms: int = REQUEST_TIMEOUT_MS, +) -> None: + """ + Проверяет доступность Kafka брокера. + + Args: + bootstrap_servers: Адрес Kafka брокера + timeout_ms: Таймаут запроса в миллисекундах + + Raises: + AirflowException: Если Kafka недоступна + """ + from kafka import KafkaAdminClient + from kafka.errors import NoBrokersAvailable + from airflow.exceptions import AirflowException + + try: + admin_client = KafkaAdminClient( + bootstrap_servers=bootstrap_servers, + request_timeout_ms=timeout_ms, + ) + # Проверяем связь, запрашивая список топиков + admin_client.list_topics() + admin_client.close() + logger.info("Kafka брокер доступен: %s", bootstrap_servers) + except NoBrokersAvailable as e: + raise AirflowException(f"Kafka брокер недоступен: {bootstrap_servers}") from e + except Exception as e: + raise AirflowException(f"Ошибка подключения к Kafka: {e}") from e + + +# ----------------------------------------------------------------------------- +# Управление топиками +# ----------------------------------------------------------------------------- +def prepare_topics( + topics: Optional[list[str]] = None, + reset: bool = True, + bootstrap_servers: str = KAFKA_BOOTSTRAP_SERVERS, + timeout_ms: int = REQUEST_TIMEOUT_MS, +) -> None: + """ + Создаёт или пересоздаёт топики Kafka. + + Args: + topics: Список топиков для создания (по умолчанию все из TOPIC_FILE_MAP) + reset: Если True — удаляет топики перед созданием + bootstrap_servers: Адрес Kafka брокера + timeout_ms: Таймаут операций в миллисекундах + """ + from kafka import KafkaAdminClient + from kafka.admin import NewTopic + from kafka.errors import TopicAlreadyExistsError, UnknownTopicOrPartitionError + from airflow.exceptions import AirflowException + + if topics is None: + topics = list(TOPIC_FILE_MAP.keys()) + + admin_client = KafkaAdminClient( + bootstrap_servers=bootstrap_servers, + request_timeout_ms=timeout_ms, + ) + + try: + # Удаляем топики если reset=True + if reset: + try: + admin_client.delete_topics(topics, timeout_ms=timeout_ms) + logger.info("Удалены топики: %s", topics) + except UnknownTopicOrPartitionError: + # Топики не существуют — это нормально + logger.info("Топики для удаления не найдены (уже отсутствуют)") + except Exception as e: + logger.warning("Ошибка при удалении топиков: %s", e) + + # Создаём топики + new_topics = [ + NewTopic( + name=topic, + num_partitions=1, # Дефолтное количество партиций + replication_factor=1, + ) + for topic in topics + ] + + try: + admin_client.create_topics(new_topics, timeout_ms=timeout_ms) + logger.info("Созданы топики: %s", topics) + except TopicAlreadyExistsError: + logger.info("Топики уже существуют: %s", topics) + except Exception as e: + raise AirflowException(f"Ошибка создания топиков: {e}") from e + + finally: + admin_client.close() + + +# ----------------------------------------------------------------------------- +# Загрузка данных из JSONL +# ----------------------------------------------------------------------------- +def load_jsonl( + file_path: str | Path, + topic: str, + limit: int = 0, + bootstrap_servers: str = KAFKA_BOOTSTRAP_SERVERS, +) -> int: + """ + Читает JSONL-файл и публикует строки в Kafka топик. + + Формат: 1 строка JSON = 1 сообщение (value), без ключа. + + Args: + file_path: Путь к .jsonl файлу + topic: Имя Kafka топика + limit: Максимальное количество строк (0 = все строки) + bootstrap_servers: Адрес Kafka брокера + + Returns: + Количество отправленных сообщений + + Raises: + AirflowException: Если файл не найден или ошибка отправки + """ + from kafka import KafkaProducer + from kafka.errors import KafkaError + from airflow.exceptions import AirflowException + + file_path = Path(file_path) + if not file_path.is_file(): + raise AirflowException(f"Файл не найден: {file_path}") + + producer = KafkaProducer( + bootstrap_servers=bootstrap_servers, + # Отправляем сырые байты (строки JSON как есть) + value_serializer=lambda v: v.encode("utf-8") if isinstance(v, str) else v, + acks="all", # Ждём подтверждения от всех реплик + retries=3, + batch_size=16384, + linger_ms=10, + ) + + sent_count = 0 + error_count = 0 + + try: + with open(file_path, "r", encoding="utf-8") as f: + for line_num, line in enumerate(f, 1): + # Пропускаем пустые строки + line = line.strip() + if not line: + continue + + # Проверяем лимит + if limit > 0 and sent_count >= limit: + logger.info( + "Достигнут лимит %d строк для %s", limit, topic + ) + break + + # Отправляем сообщение + try: + future = producer.send(topic, value=line) + # Неблокирующая отправка, собираем future для проверки + sent_count += 1 + except KafkaError as e: + error_count += 1 + logger.error("Ошибка отправки строки %d в %s: %s", line_num, topic, e) + if error_count > 10: + raise AirflowException( + f"Слишком много ошибок отправки в {topic}" + ) from e + + # Ждём завершения всех отправок + producer.flush(timeout=60) + + logger.info( + "Загрузка завершена: %s -> %s, отправлено %d сообщений", + file_path.name, + topic, + sent_count, + ) + + if sent_count == 0: + raise AirflowException(f"Не отправлено ни одного сообщения в {topic}") + + return sent_count + + except Exception as e: + if isinstance(e, AirflowException): + raise + raise AirflowException(f"Ошибка загрузки {file_path.name}: {e}") from e + + finally: + producer.close(timeout=30) + + +# ----------------------------------------------------------------------------- +# Утилиты для валидации +# ----------------------------------------------------------------------------- +def validate_load_params( + limit: int, + load_browser: bool, + load_location: bool, + load_device: bool, + load_geo: bool, +) -> None: + """ + Валидирует параметры загрузки данных. + + Args: + limit: Количество строк для загрузки + full_load: Флаг полной загрузки + load_browser: Загружать browser_events + load_location: Загружать location_events + load_device: Загружать device_events + load_geo: Загружать geo_events + + Raises: + AirflowException: Если параметры невалидны + """ + from airflow.exceptions import AirflowException + + # Проверка limit + if not isinstance(limit, int) or limit < 0: + raise AirflowException(f"limit должен быть неотрицательным int, получено: {limit}") + + # Проверка что хотя бы один поток выбран + if not any([load_browser, load_location, load_device, load_geo]): + raise AirflowException("Должен быть выбран хотя бы один поток для загрузки") + + +def check_input_files( + data_dir: str | Path = "/opt/airflow/data", + load_browser: bool = True, + load_location: bool = True, + load_device: bool = True, + load_geo: bool = True, +) -> None: + """ + Проверяет наличие необходимых JSONL-файлов. + + Args: + data_dir: Директория с данными + load_browser: Проверять browser_events.jsonl + load_location: Проверять location_events.jsonl + load_device: Проверять device_events.jsonl + load_geo: Проверять geo_events.jsonl + + Raises: + AirflowException: Если какой-либо файл отсутствует + """ + from airflow.exceptions import AirflowException + + data_dir = Path(data_dir) + files_to_check = [] + + if load_browser: + files_to_check.append("browser_events.jsonl") + if load_location: + files_to_check.append("location_events.jsonl") + if load_device: + files_to_check.append("device_events.jsonl") + if load_geo: + files_to_check.append("geo_events.jsonl") + + missing_files = [] + for filename in files_to_check: + file_path = data_dir / filename + if not file_path.is_file(): + missing_files.append(filename) + + if missing_files: + raise AirflowException( + f"Отсутствуют файлы данных в {data_dir}: {missing_files}" + ) + + logger.info("Все необходимые файлы найдены: %s", files_to_check) + + +def get_topic_file_mapping( + load_browser: bool = True, + load_location: bool = True, + load_device: bool = True, + load_geo: bool = True, +) -> dict[str, str]: + """ + Возвращает маппинг топиков на файлы для выбранных потоков. + + Returns: + Словарь {topic_name: filename} + """ + result = {} + flags = { + "browser_events": load_browser, + "location_events": load_location, + "device_events": load_device, + "geo_events": load_geo, + } + + for topic, filename in TOPIC_FILE_MAP.items(): + if flags.get(topic, True): + result[topic] = filename + + return result diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index e652521..db32e21 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -116,7 +116,7 @@ flowchart TB DM_T["VIEW для BI
(Superset/Grafana)"] end - RAW -->|kafka-console-producer| KAFKA -->|MV| STG_T -->|Batch SQL (Airflow)| ODS_T + RAW -->|kafka_load DAG / kafka-console-producer| KAFKA -->|MV| STG_T -->|Batch SQL (Airflow)| ODS_T ODS_T -->|argMax + JOIN| DDS_T -->|VIEW| DM_T ODS_T -.->|ошибки| DQ ``` @@ -598,21 +598,30 @@ INSERT INTO dm.daily_traffic SELECT * FROM dm.v_daily_traffic; Инфраструктура Airflow развёрнута и готова к использованию: ```python -# dags/ddl_init_dag.py и dags/etl_pipeline_dag.py -# +# dags/ddl_init_dag.py — создание баз/таблиц (ручной запуск при bootstrap) +# dags/kafka_load_dag.py — загрузка JSONL в Kafka (фаза 2, через kafka-python) +# dags/etl_pipeline_dag.py — основной ETL (STG→ODS→DDS→DM) + # Учебный формат: # - DDL и трансформации выполняются явными SQL-task через ClickHouseOperator; # - SQL-файлы вызываются по фиксированным путям; -# - загрузка данных в Kafka (Этап 1) выполняется через `make data`. +# - загрузка данных в Kafka (фаза 2) выполняется через DAG `kafka_load`. # -# Основной demo-сценарий: -# ddl_init -> make data -> etl_pipeline +# Основной demo-сценарий (фаза 2): +# ddl_init -> kafka_load -> etl_pipeline ``` +**DAG `kafka_load`** (фаза 2): +- Загрузка данных из `data/*.jsonl` в Kafka через `kafka-python` +- TaskGroup `precheck`: проверка Kafka, файлов, параметров +- TaskGroup `ingest`: создание топиков → параллельная загрузка 4 потоков → проверка +- Параметры: `limit` (0 = все), `reset_topics`, `load_*` (выбор потоков) + **Подключение к ClickHouse:** - Connection: `clickhouse_default` - URL: `clickhouse://default:123456@clickhouse:9000/default` (native TCP для Airflow plugin) - Provider/интеграция: `airflow-clickhouse-plugin` (в `airflow/requirements.txt`), задачи выполняются через `ClickHouseOperator`. +- Дополнительно: `kafka-python==2.0.6` для работы с Kafka из DAG. - Примечание: Superset подключается к ClickHouse по HTTP (обычно `clickhouse+connect://...:8123/...`). --- diff --git a/plans/airflow_dags_plan.md b/plans/airflow_dags_plan.md index b1ecd10..f32f2f1 100644 --- a/plans/airflow_dags_plan.md +++ b/plans/airflow_dags_plan.md @@ -61,8 +61,7 @@ check_clickhouse >> ddl_00_databases >> ddl_10_stg >> ddl_20_ods >> ddl_30_dds > ## Дизайн DAG `kafka_load` (отдельный независимый контур) ### Params (через Trigger DAG with config) -- `limit`: int, default `50`. -- `full_load`: bool, default `false`. +- `limit`: int, default `0` (0 = все строки). - `reset_topics`: bool, default `true`. - `load_browser`: bool, default `true`. - `load_location`: bool, default `true`. diff --git a/plans/runbook.md b/plans/runbook.md index dd3b5d2..eb0be1c 100644 --- a/plans/runbook.md +++ b/plans/runbook.md @@ -15,8 +15,27 @@ make up ``` -2) (Опционально) Залить данные в Kafka: +2) Залить данные в Kafka (два варианта): +**Вариант А: Через Airflow DAG `kafka_load` (рекомендуется, фаза 2)** +```bash +# Через CLI — полная загрузка по умолчанию +docker compose exec -T airflow-webserver airflow dags trigger kafka_load \ + --conf '{"reset_topics": true}' + +# Ограниченная загрузка — первые 100 строк +docker compose exec -T airflow-webserver airflow dags trigger kafka_load \ + --conf '{"limit": 100, "reset_topics": true}' + +# Или через UI: Airflow → DAGs → kafka_load → Trigger DAG with config +``` + +Параметры `kafka_load`: +- `limit` — количество строк (default: 0 — все строки) +- `reset_topics` — пересоздать топики (default: true) +- `load_browser/load_location/load_device/load_geo` — выбор потоков (default: true) + +**Вариант Б: Через shell-скрипт `make data` (устаревший)** ```bash make data ``` @@ -38,6 +57,8 @@ make ddl План реализации механики заливки (дизайн/решения): `plans/kafka_ingest_plan.md`. +План Airflow DAG'ов: `plans/airflow_dags_plan.md`. + ## Загрузка данных в Kafka (`make data`) ### Топики From 1b9991f5956abad1f38c2cc2bbd35e1d049caabd Mon Sep 17 00:00:00 2001 From: Dmitry Dementev Date: Sun, 8 Feb 2026 18:33:54 +0300 Subject: [PATCH 2/5] fix(airflow): simplify kafka_load params and align docs - Why: - For DE task we only need full ingest or limit-based sample. - load_* and full_load params were redundant and unclear in current flow. - What: - Remove full_load and load_* params from kafka_load DAG contract. - Simplify kafka helpers (validate/check files) to fixed 4-stream ingest. - Sync AGENTS, README, runbook, architecture and airflow plan docs. - Check: - python3 -m py_compile dags/kafka_load_dag.py dags/utils/kafka_helpers.py - Airflow smoke/full runs: ddl_init -> kafka_load -> etl_pipeline (all success). - Legacy path: make data && make transform (success). --- AGENTS.md | 5 +- README.md | 2 +- dags/kafka_load_dag.py | 110 ++++-------------------------------- dags/utils/kafka_helpers.py | 65 +-------------------- docs/ARCHITECTURE.md | 2 +- plans/airflow_dags_plan.md | 16 ++---- plans/runbook.md | 1 - 7 files changed, 22 insertions(+), 179 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 8f72fdd..d03e985 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -108,15 +108,12 @@ - Параметры: - `limit` (int, default 0) — количество строк (0 = все) - `reset_topics` (bool, default true) — пересоздать топики - - `load_browser/load_location/load_device/load_geo` — выбор потоков - Примеры запуска: ```json // Полная загрузка (по умолчанию) {} // Ограниченная загрузка — 100 строк {"limit": 100} - // Только browser_events - {"limit": 100, "load_location": false, "load_device": false, "load_geo": false} ``` ### `etl_pipeline` — ETL процесс @@ -143,7 +140,7 @@ make up # 3. Загрузка данных через Airflow (вместо make data) # Airflow UI → DAGs → kafka_load → Trigger DAG with config -# Параметры по умолчанию: limit=50, все потоки +# Параметры по умолчанию: limit=0 (полная загрузка), reset_topics=true # 4. Запуск ETL # Airflow UI → DAGs → etl_pipeline → Trigger DAG with config diff --git a/README.md b/README.md index 1cf85fa..f991096 100644 --- a/README.md +++ b/README.md @@ -106,7 +106,7 @@ flowchart TB DAG[DAG: ddl_init / kafka_load / etl_pipeline] end - Sources -->|make data| Kafka -->|MV| STG -->|Batch SQL| ODS -->|Batch SQL| DDS -->|VIEW| DM + Sources -->|kafka_load / make data| Kafka -->|MV| STG -->|Batch SQL| ODS -->|Batch SQL| DDS -->|VIEW| DM DAG -.->|оркестрация| STG & ODS & DDS & DM ``` diff --git a/dags/kafka_load_dag.py b/dags/kafka_load_dag.py index 7baa4ce..484f8c8 100644 --- a/dags/kafka_load_dag.py +++ b/dags/kafka_load_dag.py @@ -8,13 +8,8 @@ DAG для загрузки данных в Kafka из JSONL-файлов. - Проверка результатов через XCom Параметры (через Trigger DAG with config): - limit (int): Количество строк для загрузки (по умолчанию 50) - full_load (bool): Загрузить всё, игнорируя limit (по умолчанию False) + limit (int): Количество строк для загрузки (по умолчанию 0 = все) reset_topics (bool): Пересоздать топики (по умолчанию True) - load_browser (bool): Загружать browser_events (по умолчанию True) - load_location (bool): Загружать location_events (по умолчанию True) - load_device (bool): Загружать device_events (по умолчанию True) - load_geo (bool): Загружать geo_events (по умолчанию True) """ from __future__ import annotations @@ -32,7 +27,6 @@ from airflow.utils.task_group import TaskGroup from utils.kafka_helpers import ( check_input_files, check_kafka_ready, - get_topic_file_mapping, load_jsonl, prepare_topics, validate_load_params, @@ -62,58 +56,22 @@ def _check_kafka(**context) -> None: def _check_files(**context) -> None: """Проверка наличия входных файлов.""" - conf = context.get("dag_run", {}).conf or {} - load_browser = bool(conf.get("load_browser", context["params"]["load_browser"])) - load_location = bool(conf.get("load_location", context["params"]["load_location"])) - load_device = bool(conf.get("load_device", context["params"]["load_device"])) - load_geo = bool(conf.get("load_geo", context["params"]["load_geo"])) - - check_input_files( - data_dir=DATA_DIR, - load_browser=load_browser, - load_location=load_location, - load_device=load_device, - load_geo=load_geo, - ) + check_input_files(data_dir=DATA_DIR) def _validate_params(**context) -> None: """Валидация параметров загрузки.""" conf = context.get("dag_run", {}).conf or {} limit = int(conf.get("limit", context["params"]["limit"])) - load_browser = bool(conf.get("load_browser", context["params"]["load_browser"])) - load_location = bool(conf.get("load_location", context["params"]["load_location"])) - load_device = bool(conf.get("load_device", context["params"]["load_device"])) - load_geo = bool(conf.get("load_geo", context["params"]["load_geo"])) - - validate_load_params( - limit=limit, - load_browser=load_browser, - load_location=load_location, - load_device=load_device, - load_geo=load_geo, - ) + validate_load_params(limit=limit) def _prepare_topics(**context) -> None: """Подготовка топиков Kafka (создание/сброс).""" conf = context.get("dag_run", {}).conf or {} reset_topics = bool(conf.get("reset_topics", context["params"]["reset_topics"])) - load_browser = bool(conf.get("load_browser", context["params"]["load_browser"])) - load_location = bool(conf.get("load_location", context["params"]["load_location"])) - load_device = bool(conf.get("load_device", context["params"]["load_device"])) - load_geo = bool(conf.get("load_geo", context["params"]["load_geo"])) - # Получаем список топиков для загрузки - mapping = get_topic_file_mapping( - load_browser=load_browser, - load_location=load_location, - load_device=load_device, - load_geo=load_geo, - ) - topics = list(mapping.keys()) - - prepare_topics(topics=topics, reset=reset_topics) + prepare_topics(reset=reset_topics) def _load_events(event_type: str, **context) -> int: @@ -155,37 +113,16 @@ def _load_events(event_type: str, **context) -> int: def _verify_counts(**context) -> None: """Проверяет, что все загрузки отправили сообщения.""" - conf = context.get("dag_run", {}).conf or {} - load_browser = bool(conf.get("load_browser", context["params"]["load_browser"])) - load_location = bool(conf.get("load_location", context["params"]["load_location"])) - load_device = bool(conf.get("load_device", context["params"]["load_device"])) - load_geo = bool(conf.get("load_geo", context["params"]["load_geo"])) - ti = context["ti"] # Собираем результаты из XCom - results = {} - total_sent = 0 - - if load_browser: - count = ti.xcom_pull(task_ids="ingest.load_browser_events", key="browser_count") - results["browser_events"] = count or 0 - total_sent += count or 0 - - if load_location: - count = ti.xcom_pull(task_ids="ingest.load_location_events", key="location_count") - results["location_events"] = count or 0 - total_sent += count or 0 - - if load_device: - count = ti.xcom_pull(task_ids="ingest.load_device_events", key="device_count") - results["device_events"] = count or 0 - total_sent += count or 0 - - if load_geo: - count = ti.xcom_pull(task_ids="ingest.load_geo_events", key="geo_count") - results["geo_events"] = count or 0 - total_sent += count or 0 + results = { + "browser_events": ti.xcom_pull(task_ids="ingest.load_browser_events", key="browser_count") or 0, + "location_events": ti.xcom_pull(task_ids="ingest.load_location_events", key="location_count") or 0, + "device_events": ti.xcom_pull(task_ids="ingest.load_device_events", key="device_count") or 0, + "geo_events": ti.xcom_pull(task_ids="ingest.load_geo_events", key="geo_count") or 0, + } + total_sent = sum(results.values()) # Проверяем, что отправлено хотя бы что-то if total_sent == 0: @@ -216,36 +153,11 @@ with DAG( type="integer", description="Количество строк для загрузки (0 = все строки, по умолчанию)", ), - "full_load": Param( - default=False, - type="boolean", - description="Загрузить все данные, игнорируя limit", - ), "reset_topics": Param( default=True, type="boolean", description="Пересоздать топики перед загрузкой", ), - "load_browser": Param( - default=True, - type="boolean", - description="Загружать browser_events", - ), - "load_location": Param( - default=True, - type="boolean", - description="Загружать location_events", - ), - "load_device": Param( - default=True, - type="boolean", - description="Загружать device_events", - ), - "load_geo": Param( - default=True, - type="boolean", - description="Загружать geo_events", - ), }, ) as dag: diff --git a/dags/utils/kafka_helpers.py b/dags/utils/kafka_helpers.py index 47988a5..ac142ca 100644 --- a/dags/utils/kafka_helpers.py +++ b/dags/utils/kafka_helpers.py @@ -10,10 +10,6 @@ from __future__ import annotations import logging from pathlib import Path -from typing import TYPE_CHECKING - -if TYPE_CHECKING: - from typing import Optional # ----------------------------------------------------------------------------- # Конфигурация подключения к Kafka @@ -72,7 +68,7 @@ def check_kafka_ready( # Управление топиками # ----------------------------------------------------------------------------- def prepare_topics( - topics: Optional[list[str]] = None, + topics: list[str] | None = None, reset: bool = True, bootstrap_servers: str = KAFKA_BOOTSTRAP_SERVERS, timeout_ms: int = REQUEST_TIMEOUT_MS, @@ -237,21 +233,12 @@ def load_jsonl( # ----------------------------------------------------------------------------- def validate_load_params( limit: int, - load_browser: bool, - load_location: bool, - load_device: bool, - load_geo: bool, ) -> None: """ Валидирует параметры загрузки данных. Args: limit: Количество строк для загрузки - full_load: Флаг полной загрузки - load_browser: Загружать browser_events - load_location: Загружать location_events - load_device: Загружать device_events - load_geo: Загружать geo_events Raises: AirflowException: Если параметры невалидны @@ -262,27 +249,15 @@ def validate_load_params( if not isinstance(limit, int) or limit < 0: raise AirflowException(f"limit должен быть неотрицательным int, получено: {limit}") - # Проверка что хотя бы один поток выбран - if not any([load_browser, load_location, load_device, load_geo]): - raise AirflowException("Должен быть выбран хотя бы один поток для загрузки") - def check_input_files( data_dir: str | Path = "/opt/airflow/data", - load_browser: bool = True, - load_location: bool = True, - load_device: bool = True, - load_geo: bool = True, ) -> None: """ Проверяет наличие необходимых JSONL-файлов. Args: data_dir: Директория с данными - load_browser: Проверять browser_events.jsonl - load_location: Проверять location_events.jsonl - load_device: Проверять device_events.jsonl - load_geo: Проверять geo_events.jsonl Raises: AirflowException: Если какой-либо файл отсутствует @@ -290,16 +265,7 @@ def check_input_files( from airflow.exceptions import AirflowException data_dir = Path(data_dir) - files_to_check = [] - - if load_browser: - files_to_check.append("browser_events.jsonl") - if load_location: - files_to_check.append("location_events.jsonl") - if load_device: - files_to_check.append("device_events.jsonl") - if load_geo: - files_to_check.append("geo_events.jsonl") + files_to_check = list(TOPIC_FILE_MAP.values()) missing_files = [] for filename in files_to_check: @@ -313,30 +279,3 @@ def check_input_files( ) logger.info("Все необходимые файлы найдены: %s", files_to_check) - - -def get_topic_file_mapping( - load_browser: bool = True, - load_location: bool = True, - load_device: bool = True, - load_geo: bool = True, -) -> dict[str, str]: - """ - Возвращает маппинг топиков на файлы для выбранных потоков. - - Returns: - Словарь {topic_name: filename} - """ - result = {} - flags = { - "browser_events": load_browser, - "location_events": load_location, - "device_events": load_device, - "geo_events": load_geo, - } - - for topic, filename in TOPIC_FILE_MAP.items(): - if flags.get(topic, True): - result[topic] = filename - - return result diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index db32e21..18ce86b 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -615,7 +615,7 @@ INSERT INTO dm.daily_traffic SELECT * FROM dm.v_daily_traffic; - Загрузка данных из `data/*.jsonl` в Kafka через `kafka-python` - TaskGroup `precheck`: проверка Kafka, файлов, параметров - TaskGroup `ingest`: создание топиков → параллельная загрузка 4 потоков → проверка -- Параметры: `limit` (0 = все), `reset_topics`, `load_*` (выбор потоков) +- Параметры: `limit` (0 = все), `reset_topics` **Подключение к ClickHouse:** - Connection: `clickhouse_default` diff --git a/plans/airflow_dags_plan.md b/plans/airflow_dags_plan.md index f32f2f1..2b69ed5 100644 --- a/plans/airflow_dags_plan.md +++ b/plans/airflow_dags_plan.md @@ -63,17 +63,13 @@ check_clickhouse >> ddl_00_databases >> ddl_10_stg >> ddl_20_ods >> ddl_30_dds > ### Params (через Trigger DAG with config) - `limit`: int, default `0` (0 = все строки). - `reset_topics`: bool, default `true`. -- `load_browser`: bool, default `true`. -- `load_location`: bool, default `true`. -- `load_device`: bool, default `true`. -- `load_geo`: bool, default `true`. ### TaskGroup `precheck` | Task ID | Что делает | Реализация | |---------|------------|------------| | `check_kafka` | Проверка доступности Kafka broker | `PythonOperator` + `kafka-python` | | `check_input_files` | Проверка наличия `data/*_events.jsonl` | `PythonOperator` | -| `validate_load_params` | Валидация параметров загрузки (`limit`, `full_load`, флаги потоков) | `PythonOperator` | +| `validate_load_params` | Валидация параметров загрузки (`limit`) | `PythonOperator` | ### TaskGroup `ingest` | Task ID | Что делает | Реализация | @@ -83,7 +79,7 @@ check_clickhouse >> ddl_00_databases >> ddl_10_stg >> ddl_20_ods >> ddl_30_dds > | `load_location_events` | Публикация `location_events.jsonl` | `PythonOperator` + KafkaProducer | | `load_device_events` | Публикация `device_events.jsonl` | `PythonOperator` + KafkaProducer | | `load_geo_events` | Публикация `geo_events.jsonl` | `PythonOperator` + KafkaProducer | -| `verify_publish_counts` | Проверка, что отправлено > 0 сообщений в выбранные потоки | `PythonOperator` (по XCom) | +| `verify_publish_counts` | Проверка, что отправлено > 0 сообщений в сумме | `PythonOperator` (по XCom) | Зависимости: ```text @@ -135,7 +131,7 @@ ddl_init -> kafka_load -> etl_pipeline ``` Экспериментальные сценарии: -- `kafka_load` отдельно: проверить разные наборы/параметры загрузки без запуска transform. +- `kafka_load` отдельно: проверить разные параметры загрузки без запуска transform. - `etl_pipeline` отдельно: повторно пересчитать DDS/DM по уже загруженным данным. ## Техническая реализация (приземленно) @@ -156,7 +152,7 @@ ddl_init -> kafka_load -> etl_pipeline - `fetch_one(sql: str) -> tuple` - `dags/utils/kafka_helpers.py`: - `prepare_topics(reset: bool) -> None` - - `load_jsonl(file_path: str, topic: str, limit: int, full_load: bool) -> int` + - `load_jsonl(file_path: str, topic: str, limit: int) -> int` - `check_kafka_ready() -> None` ## Структура файлов @@ -181,7 +177,7 @@ dags/ 2. Этап 2: - Реализовать отдельный DAG `kafka_load` на `kafka-python`. - Перенести загрузку из `make data` в `kafka_load` (функциональный паритет). - - Добавить в `kafka_load` расширенные параметры экспериментов (выбор потоков, сценарии reset/no-reset). + - Добавить в `kafka_load` параметры экспериментов по объёму и reset/no-reset. - Добавить опциональный параметр автотриггера `etl_pipeline` (по умолчанию `false`). 3. Этап 3: - Добавить `dq_monitor` и alert callback (email/Slack/webhook). @@ -225,7 +221,7 @@ docker compose exec -T clickhouse clickhouse-client --user=default --password=12 docker compose exec -T clickhouse clickhouse-client --user=default --password=123456 --query "SELECT count() FROM dm.dq_summary" # 7) Этап 2: после реализации kafka_load -# Trigger DAG kafka_load {"limit": 50, "full_load": false, "reset_topics": true} +# Trigger DAG kafka_load {"limit": 50, "reset_topics": true} # Trigger DAG etl_pipeline {"full_refresh": true} ``` diff --git a/plans/runbook.md b/plans/runbook.md index eb0be1c..ede4616 100644 --- a/plans/runbook.md +++ b/plans/runbook.md @@ -33,7 +33,6 @@ docker compose exec -T airflow-webserver airflow dags trigger kafka_load \ Параметры `kafka_load`: - `limit` — количество строк (default: 0 — все строки) - `reset_topics` — пересоздать топики (default: true) -- `load_browser/load_location/load_device/load_geo` — выбор потоков (default: true) **Вариант Б: Через shell-скрипт `make data` (устаревший)** ```bash From d7588a8caa0c5e17224915dc1706cf857391f064 Mon Sep 17 00:00:00 2001 From: Dmitry Dementev Date: Sun, 8 Feb 2026 18:48:20 +0300 Subject: [PATCH 3/5] =?UTF-8?q?chore:=20=D0=B4=D0=BE=D0=B1=D0=B0=D0=B2?= =?UTF-8?q?=D0=BB=D0=B5=D0=BD=20=D0=BF=D0=BE=D0=BB=D0=BD=D0=BE=D1=86=D0=B5?= =?UTF-8?q?=D0=BD=D0=BD=D1=8B=D0=B9=20.gitignore=20=D0=B4=D0=BB=D1=8F=20ET?= =?UTF-8?q?L-=D0=BF=D1=80=D0=BE=D0=B5=D0=BA=D1=82=D0=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Python: __pycache__, *.pyc, venv - Docker: .env.local, volumes (clickhouse-data/, kafka-data/) - Airflow: logs/, *.pid, airflow.db - ClickHouse: логи сервера - Kafka/Zookeeper: logs/, data dirs - Superset: локальные БД - IDE: VS Code (partial), PyCharm - Секреты: *.pem, *.key, secrets/ - OS: .DS_Store, Thumbs.db - Данные: *.csv.gz, *.parquet, архивы --- .gitignore | 174 ++++++++++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 173 insertions(+), 1 deletion(-) diff --git a/.gitignore b/.gitignore index bee8a64..d4c6094 100644 --- a/.gitignore +++ b/.gitignore @@ -1 +1,173 @@ -__pycache__ +# ============================================================================= +# Python +# ============================================================================= +__pycache__/ +*.py[cod] +*$py.class +*.so +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +*.egg-info/ +.installed.cfg +*.egg +MANIFEST + +# ============================================================================= +# Виртуальное окружение +# ============================================================================= +.env +.venv +env/ +venv/ +ENV/ +env.bak/ +venv.bak/ +.python-version + +# ============================================================================= +# IDE и редакторы +# ============================================================================= +# VS Code +.vscode/* +!.vscode/settings.json +!.vscode/tasks.json +!.vscode/launch.json +!.vscode/extensions.json +*.code-workspace + +# PyCharm / IntelliJ +.idea/ +*.iml +*.iws +*.ipr +out/ + +# Jupyter +.ipynb_checkpoints +*.ipynb + +# ============================================================================= +# Docker +# ============================================================================= +.dockerignore +Dockerfile.*.local +*.env.local +*.env.*.local + +# Docker volumes (local data) +clickhouse-data/ +kafka-data/ +postgres-data/ +redis-data/ +airflow-data/ + +# ============================================================================= +# Airflow +# ============================================================================= +airflow/logs/ +airflow/airflow.db +airflow/airflow.cfg +airflow/webserver_config.py +airflow/*.pid +airflow/*.err +airflow/*.out +airflow/standalone_admin_password.txt + +# ============================================================================= +# ClickHouse +# ============================================================================= +clickhouse-server.log +clickhouse-server.err.log +clickhouse-client.log +*.mergetree* + +# ============================================================================= +# Kafka / Zookeeper +# ============================================================================= +*.log +kafka-logs/ +zookeeper-logs/ + +# ============================================================================= +# Superset +# ============================================================================= +superset/superset.db +superset/*.db +superset/static/ + +# ============================================================================= +# Логи и временные файлы +# ============================================================================= +logs/ +*.log +*.tmp +*.temp +*.bak +*.swp +*.swo +*~ +.DS_Store +Thumbs.db +desktop.ini + +# ============================================================================= +# Тестовые данные и артефакты +# ============================================================================= +# Файлы с данными (но не папки data/ — там исходные данные проекта) +*.csv.gz +*.jsonl.gz +*.parquet +*.avro +*.orc + +# Архивы +*.zip +*.tar.gz +*.tgz +*.rar + +# ============================================================================= +# Секреты и конфиденциальные данные +# ============================================================================= +*.pem +*.key +*.crt +secrets/ +credentials/ +config.local.yml +config.local.yaml + +# ============================================================================= +# OS-специфичные файлы +# ============================================================================= +# macOS +.DS_Store +.AppleDouble +.LSOverride +._* + +# Linux +*~ +.nfs* + +# Windows +Thumbs.db +ehthumbs.db +Desktop.ini +$RECYCLE.BIN/ +*.cab +*.msi +*.msm +*.msp +*.lnk From 0b75da9c08291da489beb242b2013c57c4d3f19f Mon Sep 17 00:00:00 2001 From: Dmitry Dementev Date: Sun, 8 Feb 2026 18:52:44 +0300 Subject: [PATCH 4/5] docs(docs): align docs with airflow-first ingest workflow - Why:\n - User-facing docs mixed Airflow and legacy CLI ingest paths and caused confusion\n- What:\n - Rework README quick start and status to use DAG chain ddl_init -> kafka_load -> etl_pipeline\n - Rewrite runbook as canonical Airflow-first execution flow\n - Sync architecture diagrams/sequence and DQ wording with current SQL and DAG behavior\n- Check:\n - Verified updated sections and removed stale markers with rg in README.md, docs/ARCHITECTURE.md, plans/runbook.md --- README.md | 23 +++----- docs/ARCHITECTURE.md | 57 +++++++++--------- plans/runbook.md | 134 ++++++++++++++++++------------------------- 3 files changed, 95 insertions(+), 119 deletions(-) diff --git a/README.md b/README.md index f991096..2bd1cb8 100644 --- a/README.md +++ b/README.md @@ -9,7 +9,7 @@ Фокус проекта: быстро показать работающий end-to-end сценарий и понятным языком объяснить, как устроены слои и почему пайплайн не падает на "грязных" данных. Коротко про поток: -`data/*.jsonl` -> Kafka (1 строка = 1 сообщение) -> ClickHouse `stg` (сырые JSON) -> Airflow batch `stg -> ods -> dds -> dm` -> Superset. +`data/*.jsonl` -> Airflow DAG `kafka_load` -> Kafka (1 строка = 1 сообщение) -> ClickHouse `stg` (сырые JSON) -> Airflow DAG `etl_pipeline` (`stg -> ods -> dds -> dm`) -> Superset. --- @@ -33,18 +33,15 @@ docker compose ps docker compose exec -T airflow-webserver airflow dags trigger ddl_init ``` -Загрузка данных в Kafka (фаза 2 — через Airflow): +Загрузка данных в Kafka через Airflow DAG: ```bash -# Вариант 1: Через Airflow DAG (рекомендуется) — полная загрузка по умолчанию +# Полная загрузка (по умолчанию limit=0) docker compose exec -T airflow-webserver airflow dags trigger kafka_load \ --conf '{"reset_topics": true}' # Ограниченная загрузка — первые 100 строк docker compose exec -T airflow-webserver airflow dags trigger kafka_load \ --conf '{"limit": 100, "reset_topics": true}' - -# Вариант 2: Через shell-скрипт (устаревший) -make data # полная загрузка ``` Запуск batch-трансформации (STG -> ODS -> DDS -> DM) в Airflow (если DAG выключен, сначала unpause): @@ -106,7 +103,7 @@ flowchart TB DAG[DAG: ddl_init / kafka_load / etl_pipeline] end - Sources -->|kafka_load / make data| Kafka -->|MV| STG -->|Batch SQL| ODS -->|Batch SQL| DDS -->|VIEW| DM + Sources -->|kafka_load| Kafka -->|MV| STG -->|Batch SQL| ODS -->|Batch SQL| DDS -->|VIEW| DM DAG -.->|оркестрация| STG & ODS & DDS & DM ``` @@ -131,14 +128,14 @@ flowchart TB │ ├── ods/ # Batch SQL: STG -> ODS │ ├── dds/ # Batch SQL: ODS -> DDS │ └── dm/ # Batch SQL: DDS -> DM -├── scripts/ # Автоматизация (apply ddl, load data, run batch) +├── scripts/ # Служебные shell-скрипты (legacy fallback, не основной путь) ├── airflow/ # Конфигурация Airflow │ └── requirements.txt ├── docs/ # Документация │ └── ARCHITECTURE.md # Подробное описание слоёв ├── data/ # Исходные JSONL файлы ├── docker-compose.yml -└── Makefile # Команды: up, ddl, data, transform +└── Makefile # Команды: up, ddl, transform ``` --- @@ -149,8 +146,6 @@ flowchart TB |---------|----------| | `make up` | Поднять инфраструктуру | | `make ddl` | Применить DDL в ClickHouse (вне Airflow) | -| `make data` | Загрузить данные в Kafka (50 строк) | -| `FULL=1 make data` | Загрузить полный датасет | | `make transform` | Запустить batch-процесс `STG -> ODS -> DDS -> DM` (вне Airflow) | Примечания про сохранность данных: @@ -214,7 +209,7 @@ flowchart LR ## Частые проблемы - `etl_pipeline` падает с сообщением про схему: сначала запустите `ddl_init`. -- После `docker compose down -v` схема и данные исчезнут: нужно заново `ddl_init` и `make data`. +- После `docker compose down -v` схема и данные исчезнут: нужно заново запустить `ddl_init`, затем `kafka_load`, затем `etl_pipeline`. - Подключения используют разные протоколы: - Airflow (ClickHouseOperator) ходит в ClickHouse по native TCP (порт `9000` внутри сети Docker). - Superset (clickhouse-connect) ходит по HTTP (порт `8123` внутри сети Docker). @@ -223,13 +218,13 @@ flowchart LR ## Статус проекта -Реализовано (Этап 1): +Реализовано: - DAG `ddl_init`: последовательное применение DDL + проверка схемы. +- DAG `kafka_load`: ingest из `.jsonl` в Kafka через `kafka-python` (параметры `limit`, `reset_topics`). - DAG `etl_pipeline`: precheck, ожидание данных в STG, batch-пересчёт ODS/DDS/DM, базовые проверки. - Устойчивость к "грязным" данным: ошибки парсинга сохраняются в ODS, а не валят ingest. В планах (не требуется для MVP задания): -- DAG `kafka_load` (чистый ingest из `.jsonl` в Kafka средствами Airflow). - Инкрементальный batch (watermark вместо `full_refresh`). - DQ мониторинг по расписанию. diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 18ce86b..a89d1d4 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -67,7 +67,7 @@ flowchart LR DM1[v_events_enriched] DM2[v_daily_traffic] DM3[v_utm_effectiveness] - DM4[v_top_pages] + DM4[v_top_pages_daily] end BE --> K1 --> S1 @@ -116,7 +116,7 @@ flowchart TB DM_T["VIEW для BI
(Superset/Grafana)"] end - RAW -->|kafka_load DAG / kafka-console-producer| KAFKA -->|MV| STG_T -->|Batch SQL (Airflow)| ODS_T + RAW -->|Airflow DAG kafka_load| KAFKA -->|MV| STG_T -->|Batch SQL (Airflow)| ODS_T ODS_T -->|argMax + JOIN| DDS_T -->|VIEW| DM_T ODS_T -.->|ошибки| DQ ``` @@ -167,7 +167,7 @@ CREATE TABLE stg.browser_raw ( | `location_event` | event_id | ReplacingMergeTree(src_ingest_ts) | Данные страниц | | `device_by_click` | click_id | ReplacingMergeTree(src_ingest_ts) | Устройства | | `geo_by_click` | click_id | ReplacingMergeTree(src_ingest_ts) | Гео-данные | -| `*_errors` | — | MergeTree | Строки с битыми ключами | +| `*_errors` | — | MergeTree | Строки с критичными ошибками парсинга | **Batch шаг наполнения ODS:** @@ -179,8 +179,8 @@ CREATE TABLE stg.browser_raw ( | `INSERT ... SELECT` в `*_errors` | Сохранение строк с критичными ошибками парсинга | **Логика разделения:** -- **Основная таблица**: строки с валидными ключами (`WHERE key IS NOT NULL`) -- **Таблица ошибок**: строки с невалидными ключами (`WHERE key IS NULL`) +- **Основная таблица**: строки с валидным business key (`WHERE key IS NOT NULL`). +- **Таблица ошибок**: строки с критичными ошибками (невалидный key, timestamp/координаты и т.п.) через отдельный `INSERT ... SELECT`. **Пример структуры:** ```sql @@ -365,7 +365,7 @@ FROM ... ```mermaid sequenceDiagram participant User as Пользователь - participant Make as Makefile + participant Compose as Docker Compose participant Airflow as Airflow participant K as Kafka participant CH as ClickHouse @@ -374,24 +374,24 @@ sequenceDiagram participant DDS as dds.* participant DM as dm.* - User->>Make: make up - Make->>K: docker compose up kafka - Make->>CH: docker compose up clickhouse - K-->>User: ✅ Инфраструктура готова + User->>Compose: make up + Compose->>K: docker compose up -d kafka + Compose->>CH: docker compose up -d clickhouse + Compose->>Airflow: docker compose up -d airflow-* + Compose-->>User: ✅ Инфраструктура готова - User->>Make: make ddl - Make->>CH: sql/ddl/00_databases.sql - Make->>CH: sql/ddl/stg/10_stg.sql (Kafka Engine) - Make->>CH: sql/ddl/ods/20_ods.sql (таблицы ODS + drop legacy MV) - Make->>CH: sql/ddl/dds/30_dds.sql - Make->>CH: sql/ddl/dm/40_dm.sql + User->>Airflow: Trigger ddl_init + Airflow->>CH: sql/ddl/00_databases.sql + Airflow->>CH: sql/ddl/stg/10_stg.sql (Kafka Engine + MV) + Airflow->>CH: sql/ddl/ods/20_ods.sql + Airflow->>CH: sql/ddl/dds/30_dds.sql + Airflow->>CH: sql/ddl/dm/40_dm.sql CH-->>User: ✅ Структура БД создана - User->>Make: make data - Make->>K: load_kafka_data.sh - K->>K: Создание топиков + User->>Airflow: Trigger kafka_load + Airflow->>K: precheck + prepare_topics loop 4 файла - Make->>K: kafka-console-producer + Airflow->>K: KafkaProducer.send(topic, json_line) end K->>CH: Потребление сообщений CH->>STG: INSERT через MV @@ -535,11 +535,11 @@ flowchart LR ### Обработка ошибок в ODS -**Проблема:** Грязные данные с невалидными ключами (NULL event_id/click_id). +**Проблема:** Грязные данные могут содержать не только невалидные ключи, но и невалидные timestamp/координаты/ID. **Решение:** -1. **Основная таблица**: только валидные строки (`WHERE key IS NOT NULL`) -2. **Таблица ошибок**: строки с невалидными ключами через отдельные `INSERT ... SELECT` +1. **Основная таблица**: строки с валидным business key (`WHERE key IS NOT NULL`) +2. **Таблица ошибок**: строки с критичными ошибками парсинга через отдельные `INSERT ... SELECT` 3. **DQ-метрики**: массив `parse_errors` для аудита ```sql @@ -549,7 +549,8 @@ SELECT ... FROM stg.browser_raw WHERE event_id IS NOT NULL; -- Таблица ошибок INSERT INTO ods.browser_event_errors -SELECT ... FROM stg.browser_raw WHERE event_id IS NULL; +SELECT ... FROM stg.browser_raw +WHERE event_id IS NULL OR event_ts IS NULL OR click_id IS NULL; ``` ### Partial data в DDS @@ -599,19 +600,19 @@ INSERT INTO dm.daily_traffic SELECT * FROM dm.v_daily_traffic; ```python # dags/ddl_init_dag.py — создание баз/таблиц (ручной запуск при bootstrap) -# dags/kafka_load_dag.py — загрузка JSONL в Kafka (фаза 2, через kafka-python) +# dags/kafka_load_dag.py — загрузка JSONL в Kafka (через kafka-python) # dags/etl_pipeline_dag.py — основной ETL (STG→ODS→DDS→DM) # Учебный формат: # - DDL и трансформации выполняются явными SQL-task через ClickHouseOperator; # - SQL-файлы вызываются по фиксированным путям; -# - загрузка данных в Kafka (фаза 2) выполняется через DAG `kafka_load`. +# - загрузка данных в Kafka выполняется через DAG `kafka_load`. # -# Основной demo-сценарий (фаза 2): +# Основной demo-сценарий: # ddl_init -> kafka_load -> etl_pipeline ``` -**DAG `kafka_load`** (фаза 2): +**DAG `kafka_load`**: - Загрузка данных из `data/*.jsonl` в Kafka через `kafka-python` - TaskGroup `precheck`: проверка Kafka, файлов, параметров - TaskGroup `ingest`: создание топиков → параллельная загрузка 4 потоков → проверка diff --git a/plans/runbook.md b/plans/runbook.md index ede4616..d380dae 100644 --- a/plans/runbook.md +++ b/plans/runbook.md @@ -1,107 +1,87 @@ -# Runbook: запуск демо и загрузка данных +# Runbook: Airflow-first запуск демо -Этот документ фиксирует порядок действий и `make`‑таргеты. Он не описывает внутренности ClickHouse‑слоёв (это в `plans/clickhouse_ddl.md`). +Документ фиксирует канонический пользовательский сценарий через Airflow DAG'и. +Внутренности слоёв ClickHouse описаны в `plans/clickhouse_ddl.md` и `docs/ARCHITECTURE.md`. ## Предпосылки - Docker + Docker Compose. -- Доступ к Docker daemon (если `docker compose ...` пишет `permission denied ... /var/run/docker.sock`, добавьте пользователя в группу `docker` или запускайте команды с правами, принятыми в вашей среде). +- Доступ к Docker daemon. -## Быстрый сценарий +## Канонический сценарий (через Airflow) 1) Поднять инфраструктуру: ```bash make up +docker compose ps ``` -2) Залить данные в Kafka (два варианта): +2) Инициализировать схему ClickHouse: -**Вариант А: Через Airflow DAG `kafka_load` (рекомендуется, фаза 2)** ```bash -# Через CLI — полная загрузка по умолчанию +docker compose exec -T airflow-webserver airflow dags trigger ddl_init +``` + +3) Загрузить данные в Kafka через DAG `kafka_load`: + +```bash +# Рекомендуется для демо: небольшой срез +docker compose exec -T airflow-webserver airflow dags trigger kafka_load \ + --conf '{"limit": 50, "reset_topics": true}' + +# Полная загрузка (limit=0 по умолчанию) docker compose exec -T airflow-webserver airflow dags trigger kafka_load \ --conf '{"reset_topics": true}' - -# Ограниченная загрузка — первые 100 строк -docker compose exec -T airflow-webserver airflow dags trigger kafka_load \ - --conf '{"limit": 100, "reset_topics": true}' - -# Или через UI: Airflow → DAGs → kafka_load → Trigger DAG with config ``` -Параметры `kafka_load`: -- `limit` — количество строк (default: 0 — все строки) -- `reset_topics` — пересоздать топики (default: true) - -**Вариант Б: Через shell-скрипт `make data` (устаревший)** -```bash -make data -``` - -`make data` не зависит от ClickHouse/DDL — достаточно, чтобы Kafka была поднята. - -3) Применить DDL в ClickHouse: +4) Запустить ETL: ```bash -make ddl +docker compose exec -T airflow-webserver airflow dags trigger etl_pipeline \ + --conf '{"full_refresh": true}' ``` -## Make таргеты - -- `make up` — `docker compose up -d` (поднимает весь стек из `docker-compose.yml`). -- `make ddl` — применяет исполняемые SQL-файлы из `sql/ddl/*` в контейнер ClickHouse через `clickhouse-client`. -- `make data` — пересоздаёт топики (по умолчанию) и публикует события из `data/*.jsonl` в Kafka (1 строка = 1 Kafka message value). -- `make transform` — выполняет batch-процесс `STG -> ODS -> DDS -> DM` через `scripts/run_batch.sh`. - -План реализации механики заливки (дизайн/решения): `plans/kafka_ingest_plan.md`. - -План Airflow DAG'ов: `plans/airflow_dags_plan.md`. - -## Загрузка данных в Kafka (`make data`) - -### Топики - -Скрипт использует фиксированный маппинг: - -- `data/browser_events.jsonl` → `browser_events` -- `data/location_events.jsonl` → `location_events` -- `data/device_events.jsonl` → `device_events` -- `data/geo_events.jsonl` → `geo_events` - -### Режимы загрузки - -- По умолчанию — “debug срез”: первые 50 строк каждого файла. -- Полная загрузка — весь файл. - -Параметры (env): - -- `LIMIT` — сколько строк брать из каждого `.jsonl`. По умолчанию загружаются все записи (весь файл). - Для ограничения используйте `LIMIT=50` или `LIMIT=100`. -- `RESET_TOPICS` — если `RESET_TOPICS=1` (по умолчанию), топики удаляются и создаются заново с теми же именами. -- `BOOTSTRAP_SERVER` — bootstrap для Kafka *изнутри kafka‑контейнера* (по умолчанию `kafka:29092`). - -Примеры: +5) Проверить результат: ```bash -# Загрузить все данные (по умолчанию) -make data - -# Быстрый тест — 50 строк на поток -LIMIT=50 make data - -# Ограниченная загрузка — 100 строк на поток -LIMIT=100 make data - -# Дозалить данные без пересоздания топиков -RESET_TOPICS=0 make data +docker compose exec -T clickhouse clickhouse-client --user=default --password=123456 \ + --query "SELECT count() FROM stg.browser_raw" +docker compose exec -T clickhouse clickhouse-client --user=default --password=123456 \ + --query "SELECT count() FROM ods.browser_event" +docker compose exec -T clickhouse clickhouse-client --user=default --password=123456 \ + --query "SELECT count() FROM dds.event" +docker compose exec -T clickhouse clickhouse-client --user=default --password=123456 \ + --query "SELECT * FROM dm.dq_summary ORDER BY layer, table_name, check_name" ``` -## Применение DDL в ClickHouse (`make ddl`) +## Параметры DAG'ов -Скрипт исполняет SQL-файлы из `sql/ddl/*` по фиксированному порядку. Для `ENGINE = Kafka` важно, чтобы `kafka_broker_list` был доступен из контейнера ClickHouse. +- `ddl_init`: + - `verify_only` (bool, default: `false`) — только проверка схемы, без применения DDL. +- `kafka_load`: + - `limit` (int, default: `0`) — количество строк на поток (`0` = весь файл). + - `reset_topics` (bool, default: `true`) — удалить и создать топики заново. +- `etl_pipeline`: + - `full_refresh` (bool, default: `true`) — очищать DDS перед загрузкой. + - `wait_stg_timeout_sec` (int, default: `600`) — таймаут ожидания данных в STG. -В текущем compose: +## Роль Make-таргетов -- для соединений “контейнер → Kafka” используйте `kafka:29092`; -- `localhost:9092` подходит только для клиентов на хосте. +- `make up` — основной способ поднять стек. +- `make ddl`, `make transform` — технический fallback для низкоуровневой диагностики вне Airflow. +- Загрузка в Kafka в runbook выполняется только через DAG `kafka_load`. + +## Важные примечания + +- Для связей контейнеров используйте `kafka:29092` (не `localhost:9092`). +- После `docker compose down -v` нужно заново выполнить: + 1. `ddl_init` + 2. `kafka_load` + 3. `etl_pipeline` + +## Связанные документы + +- План ingest: `plans/kafka_ingest_plan.md` +- План DAG'ов: `plans/airflow_dags_plan.md` +- Архитектура: `docs/ARCHITECTURE.md` From 60482af66b09c6b81b2cfd2569b3f994583906fc Mon Sep 17 00:00:00 2001 From: Dmitry Dementev Date: Sun, 8 Feb 2026 19:03:11 +0300 Subject: [PATCH 5/5] =?UTF-8?q?docs:=20=D0=BE=D0=B1=D0=BD=D0=BE=D0=B2?= =?UTF-8?q?=D0=BB=D0=B5=D0=BD=D1=8B=20Mermaid-=D0=B4=D0=B8=D0=B0=D0=B3?= =?UTF-8?q?=D1=80=D0=B0=D0=BC=D0=BC=D1=8B=20=D0=B2=20README=20=D0=B8=20ARC?= =?UTF-8?q?HITECTURE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Упрощены диаграммы архитектуры, убраны эмодзи - В README: компактная схема Airflow → Kafka → ClickHouse - В ARCHITECTURE: обновлены общая схема, слои и сборка DDS - Все диаграммы отражают 3 DAG: ddl_init, kafka_load, etl_pipeline --- README.md | 30 ++++++------ docs/ARCHITECTURE.md | 109 +++++++++++++++++-------------------------- 2 files changed, 58 insertions(+), 81 deletions(-) diff --git a/README.md b/README.md index 2bd1cb8..83a1ebf 100644 --- a/README.md +++ b/README.md @@ -80,31 +80,29 @@ docker compose exec -T clickhouse clickhouse-client --user=default --password=12 ## Архитектура (в двух словах) ```mermaid -flowchart TB - subgraph Sources["JSONL файлы"] - BE[browser_events.jsonl] - LE[location_events.jsonl] - DE[device_events.jsonl] - GE[geo_events.jsonl] +flowchart LR + subgraph AF["Airflow"] + D1["ddl_init"] + D2["kafka_load"] + D3["etl_pipeline"] end subgraph Kafka["Kafka"] - KT[Топики] + Topics[4 топика] end subgraph CH["ClickHouse"] - STG["stg: сырьё + Kafka MV"] - ODS["ods: типизация + DQ"] - DDS["dds: сущности"] - DM["dm: витрины (VIEW)"] + STG["STG: сырые данные"] + ODS["ODS: типизация + DQ"] + DDS["DDS: сущности"] + DM["DM: витрины VIEW"] end - subgraph Airflow["Airflow"] - DAG[DAG: ddl_init / kafka_load / etl_pipeline] - end + D2 -->|загрузка JSONL| Kafka -->|Kafka MV| STG + STG -->|batch| ODS -->|batch| DDS -->|VIEW| DM - Sources -->|kafka_load| Kafka -->|MV| STG -->|Batch SQL| ODS -->|Batch SQL| DDS -->|VIEW| DM - DAG -.->|оркестрация| STG & ODS & DDS & DM + D1 -.->|DDL| CH + D3 -.->|batch| ODS & DDS ``` Особенность задания про "грязные данные": парсинг не валит pipeline, ошибки фиксируются в `ods.*_errors` и в поле `parse_errors`. diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index a89d1d4..6503cfd 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -25,100 +25,79 @@ ```mermaid flowchart LR - subgraph Sources["📁 Источники (JSONL)"] - BE[browser_events] - LE[location_events] - DE[device_events] - GE[geo_events] + subgraph AF["Airflow"] + DAG1["ddl_init"] + DAG2["kafka_load"] + DAG3["etl_pipeline"] end - subgraph Kafka["🚀 Kafka"] - K1[browser_events] - K2[location_events] - K3[device_events] - K4[geo_events] + subgraph Kafka["Kafka"] + K[4 топика] end - subgraph STG["📦 STG"] - S1[browser_raw] - S2[location_raw] - S3[device_raw] - S4[geo_raw] + subgraph STG["STG"] + S[*_raw таблицы] end - subgraph AF["⚙️ Airflow"] - B1[load_ods
sql/ods/20_stg_to_ods.sql] - end - - subgraph ODS["🔧 ODS"] - O1[browser_event] - O2[location_event] - O3[device_by_click] - O4[geo_by_click] + subgraph ODS["ODS"] + O[*_event таблицы] OE[error_tables] end - subgraph DDS["🎯 DDS"] - DE1[event] - DC1[click] + subgraph DDS["DDS"] + D1[event] + D2[click] end - subgraph DM["📊 DM"] - DM1[v_events_enriched] - DM2[v_daily_traffic] - DM3[v_utm_effectiveness] - DM4[v_top_pages_daily] + subgraph DM["DM"] + V[витрины VIEW] end - BE --> K1 --> S1 - LE --> K2 --> S2 - DE --> K3 --> S3 - GE --> K4 --> S4 + DAG2 -->|JSONL| K -->|MV| S + S -->|batch| O + O -->|argMax + JOIN| D1 & D2 + O -.->|ошибки| OE + D1 & D2 -->|VIEW| V - S1 & S2 & S3 & S4 -->|Batch SQL| B1 - B1 --> O1 & O2 & O3 & O4 - B1 -.-> OE - O1 --> DE1 - O2 --> DE1 - O3 --> DC1 - O4 --> DC1 - DE1 --> DM1 - DC1 --> DM1 - DE1 --> DM2 & DM3 & DM4 - DC1 --> DM2 & DM3 & DM4 + DAG1 -.->|DDL| STG & ODS & DDS & DM + DAG3 -.->|batch| ODS & DDS ``` ### Слои и их назначение ```mermaid -flowchart TB - subgraph L0["📝 Источники"] - RAW["JSON файлы (1000 строк)"] +flowchart LR + subgraph AF["Airflow"] + DAG1["ddl_init"] + DAG2["kafka_load"] + DAG3["etl_pipeline"] end - subgraph L1["📦 STG - Staging"] - direction LR + subgraph L1["STG"] KAFKA["Kafka Engine"] - STG_T["*_raw таблицы
(MergeTree)"] + STG_T["*_raw таблицы"] end - subgraph L2["🔧 ODS - Операционный слой"] - direction LR - ODS_T["Типизированные таблицы
(ReplacingMergeTree)"] - DQ["parse_errors
DQ-метрики"] + subgraph L2["ODS"] + ODS_T["Типизированные таблицы"] + DQ["parse_errors"] end - subgraph L3["🎯 DDS - Детальный слой"] - DDS_T["event + click
(Batch SQL)"] + subgraph L3["DDS"] + DDS_T["event + click"] end - subgraph L4["📊 DM - Витрины"] - DM_T["VIEW для BI
(Superset/Grafana)"] + subgraph L4["DM"] + DM_T["VIEW"] end - RAW -->|Airflow DAG kafka_load| KAFKA -->|MV| STG_T -->|Batch SQL (Airflow)| ODS_T + DAG2 -->|JSONL| KAFKA -->|MV| STG_T + STG_T -->|batch| ODS_T ODS_T -->|argMax + JOIN| DDS_T -->|VIEW| DM_T ODS_T -.->|ошибки| DQ + + DAG1 -.->|DDL| L1 & L2 & L3 & L4 + DAG3 -.->|batch| ODS_T & DDS_T ``` --- @@ -476,8 +455,8 @@ flowchart LR EV["event"] end - B -->|JOIN по event_id| EV - L -->|JOIN по event_id| EV + B -->|JOIN event_id| EV + L -->|JOIN event_id| EV ``` **click** (device + geo) с поддержкой partial data: @@ -489,7 +468,7 @@ flowchart LR end subgraph BUILD["Batch SQL"] - U["UNION DISTINCT
click_id"] + U["UNION DISTINCT click_id"] J["LEFT JOIN"] end