feat(airflow): полная реализация фазы 2 — DAG для загрузки данных в Kafka

Слияние ветки с реализацией автоматизированной загрузки JSONL-файлов в Kafka
через Airflow DAG с валидацией, мониторингом и документацией.

- Что добавлено:
  - dags/kafka_load_dag.py: TaskGroup-пайплайн загрузки 4 потоков данных
  - dags/utils/kafka_helpers.py: хелперы для работы с Kafka (проверка,
    создание топиков, загрузка с лимитом)
  - airflow/requirements.txt: зависимость kafka-python==2.0.6
  - .gitignore: полноценный шаблон для ETL-проекта

- Параметры DAG:
  - limit: ограничение строк (0 = все)
  - reset_topics: пересоздание топиков перед загрузкой
  - load_browser/device/geo/location_events: выбор потоков

- Обновлена документация:
  - README.md, AGENTS.md, docs/ARCHITECTURE.md
  - plans/runbook.md, plans/airflow_dags_plan.md

.gitignore

@@ -0,0 +1,173 @@
+# =============================================================================
+# 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

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,62 @@
   - 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) — пересоздать топики
+- Примеры запуска:
+  ```json
+  // Полная загрузка (по умолчанию)
+  {}
+  // Ограниченная загрузка — 100 строк
+  {"limit": 100}
+  ```
+
+### `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=0 (полная загрузка), reset_topics=true
+
+# 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) — пользовательская документация (быстрый старт, архитектура)

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,10 +33,15 @@ docker compose ps
 docker compose exec -T airflow-webserver airflow dags trigger ddl_init
 ```
 
-Загрузка небольшого среза данных в Kafka:
+Загрузка данных в Kafka через Airflow DAG:
 ```bash
-make data            # по умолчанию первые 50 строк
-# или: FULL=1 make data    # полный датасет (1000 строк)
+# Полная загрузка (по умолчанию 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}'
 ```
 
 Запуск batch-трансформации (STG -> ODS -> DDS -> DM) в Airflow (если DAG выключен, сначала unpause):
@@ -75,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 / etl_pipeline]
-    end
+    D2 -->|загрузка JSONL| Kafka -->|Kafka MV| STG
+    STG -->|batch| ODS -->|batch| DDS -->|VIEW| DM
 
-    Sources -->|make data| 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`.
@@ -123,14 +126,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
 ```
 
 ---
@@ -141,8 +144,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) |
 
 Примечания про сохранность данных:
@@ -206,7 +207,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).
@@ -215,13 +216,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 мониторинг по расписанию.
 

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

dags/kafka_load_dag.py

@@ -0,0 +1,234 @@
+"""
+DAG для загрузки данных в Kafka из JSONL-файлов.
+
+Функциональность:
+- Проверка доступности Kafka и наличия файлов
+- Создание/сброс топиков
+- Параллельная загрузка 4 потоков данных
+- Проверка результатов через XCom
+
+Параметры (через Trigger DAG with config):
+    limit (int): Количество строк для загрузки (по умолчанию 0 = все)
+    reset_topics (bool): Пересоздать топики (по умолчанию 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,
+    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:
+    """Проверка наличия входных файлов."""
+    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"]))
+    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"]))
+
+    prepare_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:
+    """Проверяет, что все загрузки отправили сообщения."""
+    ti = context["ti"]
+
+    # Собираем результаты из XCom
+    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:
+        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 = все строки, по умолчанию)",
+        ),
+        "reset_topics": Param(
+            default=True,
+            type="boolean",
+            description="Пересоздать топики перед загрузкой",
+        ),
+    },
+) 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

dags/utils/__init__.py

@@ -0,0 +1 @@
+# Utils package для Airflow DAG'ов

dags/utils/kafka_helpers.py

@@ -0,0 +1,281 @@
+"""
+Helper-функции для работы с Kafka из Airflow DAG'ов.
+
+Использует kafka-python:
+- KafkaAdminClient — для управления топиками
+- KafkaProducer — для публикации сообщений
+"""
+
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+
+# -----------------------------------------------------------------------------
+# Конфигурация подключения к 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: list[str] | None = 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,
+) -> None:
+    """
+    Валидирует параметры загрузки данных.
+
+    Args:
+        limit: Количество строк для загрузки
+
+    Raises:
+        AirflowException: Если параметры невалидны
+    """
+    from airflow.exceptions import AirflowException
+
+    # Проверка limit
+    if not isinstance(limit, int) or limit < 0:
+        raise AirflowException(f"limit должен быть неотрицательным int, получено: {limit}")
+
+
+def check_input_files(
+    data_dir: str | Path = "/opt/airflow/data",
+) -> None:
+    """
+    Проверяет наличие необходимых JSONL-файлов.
+
+    Args:
+        data_dir: Директория с данными
+
+    Raises:
+        AirflowException: Если какой-либо файл отсутствует
+    """
+    from airflow.exceptions import AirflowException
+
+    data_dir = Path(data_dir)
+    files_to_check = list(TOPIC_FILE_MAP.values())
+
+    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)

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<br/>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]
+    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 таблицы<br/>(MergeTree)"]
+        STG_T["*_raw таблицы"]
     end
 
-    subgraph L2["🔧 ODS - Операционный слой"]
-        direction LR
-        ODS_T["Типизированные таблицы<br/>(ReplacingMergeTree)"]
-        DQ["parse_errors<br/>DQ-метрики"]
+    subgraph L2["ODS"]
+        ODS_T["Типизированные таблицы"]
+        DQ["parse_errors"]
     end
 
-    subgraph L3["🎯 DDS - Детальный слой"]
-        DDS_T["event + click<br/>(Batch SQL)"]
+    subgraph L3["DDS"]
+        DDS_T["event + click"]
     end
 
-    subgraph L4["📊 DM - Витрины"]
-        DM_T["VIEW для BI<br/>(Superset/Grafana)"]
+    subgraph L4["DM"]
+        DM_T["VIEW"]
     end
 
-    RAW -->|kafka-console-producer| 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
 ```
 
 ---
@@ -167,7 +146,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 +158,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 +344,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 +353,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
@@ -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<br/>click_id"]
+        U["UNION DISTINCT click_id"]
         J["LEFT JOIN"]
     end
 
@@ -535,11 +514,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 +528,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
@@ -598,21 +578,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 (через kafka-python)
+# dags/etl_pipeline_dag.py — основной ETL (STG→ODS→DDS→DM)
+
 # Учебный формат:
 # - DDL и трансформации выполняются явными SQL-task через ClickHouseOperator;
 # - SQL-файлы вызываются по фиксированным путям;
-# - загрузка данных в Kafka (Этап 1) выполняется через `make data`.
+# - загрузка данных в Kafka выполняется через DAG `kafka_load`.
 #
 # Основной demo-сценарий:
-# ddl_init -> make data -> etl_pipeline
+# ddl_init -> kafka_load -> etl_pipeline
 ```
 
+**DAG `kafka_load`**:
+- Загрузка данных из `data/*.jsonl` в Kafka через `kafka-python`
+- TaskGroup `precheck`: проверка Kafka, файлов, параметров
+- TaskGroup `ingest`: создание топиков → параллельная загрузка 4 потоков → проверка
+- Параметры: `limit` (0 = все), `reset_topics`
+
 **Подключение к 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/...`).
 
 ---

plans/airflow_dags_plan.md

@@ -61,20 +61,15 @@ 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`.
-- `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 | Что делает | Реализация |
@@ -84,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
@@ -136,7 +131,7 @@ ddl_init -> kafka_load -> etl_pipeline
 ```
 
 Экспериментальные сценарии:
-- `kafka_load` отдельно: проверить разные наборы/параметры загрузки без запуска transform.
+- `kafka_load` отдельно: проверить разные параметры загрузки без запуска transform.
 - `etl_pipeline` отдельно: повторно пересчитать DDS/DM по уже загруженным данным.
 
 ## Техническая реализация (приземленно)
@@ -157,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`
 
 ## Структура файлов
@@ -182,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).
@@ -226,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}
 ```
 

plans/runbook.md

@@ -1,87 +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:
 
 ```bash
-make data
+docker compose exec -T airflow-webserver airflow dags trigger ddl_init
 ```
 
-`make data` не зависит от ClickHouse/DDL — достаточно, чтобы Kafka была поднята.
-
-3) Применить DDL в ClickHouse:
+3) Загрузить данные в Kafka через DAG `kafka_load`:
 
 ```bash
-make ddl
+# Рекомендуется для демо: небольшой срез
+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}'
 ```
 
-## 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`.
-
-## Загрузка данных в 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`).
-
-Примеры:
+4) Запустить ETL:
 
 ```bash
-# Загрузить все данные (по умолчанию)
-make data
-
-# Быстрый тест — 50 строк на поток
-LIMIT=50 make data
-
-# Ограниченная загрузка — 100 строк на поток
-LIMIT=100 make data
-
-# Дозалить данные без пересоздания топиков
-RESET_TOPICS=0 make data
+docker compose exec -T airflow-webserver airflow dags trigger etl_pipeline \
+  --conf '{"full_refresh": true}'
 ```
 
-## Применение DDL в ClickHouse (`make ddl`)
+5) Проверить результат:
 
-Скрипт исполняет SQL-файлы из `sql/ddl/*` по фиксированному порядку. Для `ENGINE = Kafka` важно, чтобы `kafka_broker_list` был доступен из контейнера ClickHouse.
+```bash
+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"
+```
 
-В текущем compose:
+## Параметры DAG'ов
 
-- для соединений “контейнер → Kafka” используйте `kafka:29092`;
-- `localhost:9092` подходит только для клиентов на хосте.
+- `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.
+
+## Роль Make-таргетов
+
+- `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`