feat(airflow): добавлен гейт целостности DDS для урока 4

- Зачем:
  - урок 4 должен показывать не только измерение сирот в DDS, но и остановку Airflow DAG при нарушении связи dds.event -> dds.click.
- Что:
  - добавлен assert_dds_integrity в etl_pipeline и документация управляемого красного сценария.
  - вынесены общие helper'ы для SQL-split и boolean-параметров Airflow.
  - добавлен урок 4 и обновлены навигация курса, план обучения и operations notes.
- Проверка:
  - python3 -m py_compile airflow/dags/etl_pipeline_dag.py airflow/dags/ddl_init_dag.py airflow/dags/kafka_load_dag.py airflow/dags/utils/airflow_params.py airflow/dags/utils/sql_helpers.py.
  - docker compose exec -T airflow-webserver airflow dags test etl_pipeline 2026-06-05T18:00:00 -c '{"full_refresh": true}'.

airflow/dags/ddl_init_dag.py

@@ -9,7 +9,6 @@ DAG инициализации DDL в ClickHouse.
 
 from __future__ import annotations
 
-import re
 from datetime import datetime, timedelta
 from pathlib import Path
 
@@ -20,6 +19,8 @@ from airflow.operators.empty import EmptyOperator
 from airflow.operators.python import BranchPythonOperator, PythonOperator
 from airflow.utils.trigger_rule import TriggerRule
 from airflow_clickhouse_plugin.operators.clickhouse import ClickHouseOperator
+from utils.airflow_params import parse_bool_param
+from utils.sql_helpers import load_sql_statements as load_sql_file_statements
 
 
 # -----------------------------------------------------------------------------
@@ -55,23 +56,7 @@ SQL_ROOT = resolve_sql_root()
 
 def load_sql_statements(relative_path: str) -> tuple[str, ...]:
     """Читает SQL-файл и делит его на отдельные команды по ';'."""
-    file_path = SQL_ROOT / relative_path
-    if not file_path.is_file():
-        raise AirflowException(f"SQL-файл не найден: {file_path}")
-
-    sql_text = file_path.read_text(encoding="utf-8")
-    statements: list[str] = []
-    for segment in sql_text.split(";"):
-        # Убираем блочные и строковые комментарии, чтобы не отправлять "пустые" запросы.
-        no_block_comments = re.sub(r"/\*.*?\*/", "", segment, flags=re.S)
-        lines = [line for line in no_block_comments.splitlines() if not line.strip().startswith("--")]
-        cleaned = "\n".join(lines).strip()
-        if cleaned:
-            statements.append(cleaned)
-
-    if not statements:
-        raise AirflowException(f"SQL-файл пустой: {file_path}")
-    return tuple(statements)
+    return load_sql_file_statements(SQL_ROOT, relative_path)
 
 
 # -----------------------------------------------------------------------------
@@ -96,7 +81,10 @@ def choose_ddl_mode(**context) -> str:
     """Выбирает ветку выполнения: full DDL или только verify."""
     dag_run = context.get("dag_run")
     conf = dag_run.conf if dag_run else {}
-    verify_only = bool(conf.get("verify_only", context["params"]["verify_only"]))
+    verify_only = parse_bool_param(
+        conf.get("verify_only", context["params"]["verify_only"]),
+        "verify_only",
+    )
     return "skip_ddl" if verify_only else "ddl_00_databases"
 
 

airflow/dags/etl_pipeline_dag.py

@@ -1,6 +1,9 @@
 """
 DAG ETL-процесса STG -> ODS -> DDS -> DM для учебного проекта.
 
+Поток задач:
+  precheck -> transform: wait -> ods -> dq -> branch -> dds -> integrity -> dm -> validate
+
 Принципы реализации:
 - SQL выполняется явными task на ClickHouseOperator;
 - SQL-файлы вызываются по фиксированным путям;
@@ -9,7 +12,6 @@ DAG ETL-процесса STG -> ODS -> DDS -> DM для учебного про
 
 from __future__ import annotations
 
-import re
 import time
 from datetime import datetime, timedelta
 from pathlib import Path
@@ -23,6 +25,8 @@ from airflow.utils.task_group import TaskGroup
 from airflow.utils.trigger_rule import TriggerRule
 from airflow_clickhouse_plugin.hooks.clickhouse import ClickHouseHook
 from airflow_clickhouse_plugin.operators.clickhouse import ClickHouseOperator
+from utils.airflow_params import parse_bool_param
+from utils.sql_helpers import load_sql_statements as load_sql_file_statements
 
 
 # -----------------------------------------------------------------------------
@@ -58,23 +62,7 @@ SQL_ROOT = resolve_sql_root()
 
 def load_sql_statements(relative_path: str) -> tuple[str, ...]:
     """Читает SQL-файл и делит его на отдельные команды по ';'."""
-    file_path = SQL_ROOT / relative_path
-    if not file_path.is_file():
-        raise AirflowException(f"SQL-файл не найден: {file_path}")
-
-    sql_text = file_path.read_text(encoding="utf-8")
-    statements: list[str] = []
-    for segment in sql_text.split(";"):
-        # Убираем блочные и строковые комментарии, чтобы не отправлять "пустые" запросы.
-        no_block_comments = re.sub(r"/\*.*?\*/", "", segment, flags=re.S)
-        lines = [line for line in no_block_comments.splitlines() if not line.strip().startswith("--")]
-        cleaned = "\n".join(lines).strip()
-        if cleaned:
-            statements.append(cleaned)
-
-    if not statements:
-        raise AirflowException(f"SQL-файл пустой: {file_path}")
-    return tuple(statements)
+    return load_sql_file_statements(SQL_ROOT, relative_path)
 
 
 # -----------------------------------------------------------------------------
@@ -228,7 +216,10 @@ def choose_full_refresh(**context) -> str:
     """Ветвление: делать TRUNCATE DDS или пропустить."""
     dag_run = context.get("dag_run")
     conf = dag_run.conf if dag_run else {}
-    full_refresh = bool(conf.get("full_refresh", context["params"]["full_refresh"]))
+    full_refresh = parse_bool_param(
+        conf.get("full_refresh", context["params"]["full_refresh"]),
+        "full_refresh",
+    )
     return "transform.truncate_dds_click" if full_refresh else "transform.skip_truncate"
 
 
@@ -245,6 +236,22 @@ def assert_dm_summary_not_empty(**context) -> None:
         raise AirflowException("dm.dq_summary пуста после load_dm_summary.")
 
 
+def assert_dds_integrity(**context) -> None:
+    """Падает, если события ссылаются на отсутствующие клики."""
+    ti = context["ti"]
+    result = ti.xcom_pull(task_ids="transform.check_dds_integrity")
+
+    if not result or not result[0] or len(result[0]) != 1:
+        raise AirflowException(f"Некорректный результат check_dds_integrity: {result}")
+
+    orphan_events = int(result[0][0])
+    if orphan_events > 0:
+        raise AirflowException(
+            f"DDS integrity check failed: orphan_events={orphan_events}. "
+            "Есть события, чей click_id отсутствует в dds.click."
+        )
+
+
 with DAG(
     dag_id="etl_pipeline",
     description="ETL STG -> ODS -> DDS -> DM для demo-проекта",
@@ -342,6 +349,12 @@ with DAG(
             database="default",
         )
 
+        assert_dds_integrity_task = PythonOperator(
+            task_id="assert_dds_integrity",
+            python_callable=assert_dds_integrity,
+            retries=0,
+        )
+
         load_dm_summary = ClickHouseOperator(
             task_id="load_dm_summary",
             sql=load_sql_statements("dm/40_dds_to_dm.sql"),
@@ -364,6 +377,14 @@ with DAG(
         wait_for_stg_data_task >> load_ods >> check_ods_quality >> choose_refresh_mode
         choose_refresh_mode >> truncate_dds_click >> truncate_dds_event >> truncate_complete
         choose_refresh_mode >> skip_truncate >> truncate_complete
-        truncate_complete >> load_dds >> check_dds_integrity >> load_dm_summary >> validate_dm_summary_sql >> validate_dm_summary
+        (
+            truncate_complete
+            >> load_dds
+            >> check_dds_integrity
+            >> assert_dds_integrity_task
+            >> load_dm_summary
+            >> validate_dm_summary_sql
+            >> validate_dm_summary
+        )
 
     precheck >> transform

airflow/dags/kafka_load_dag.py

@@ -24,6 +24,7 @@ from airflow.operators.python import PythonOperator
 from airflow.utils.task_group import TaskGroup
 
 # Импортируем helper-функции
+from utils.airflow_params import parse_bool_param
 from utils.kafka_helpers import (
     check_input_files,
     check_kafka_ready,
@@ -69,7 +70,10 @@ def _validate_params(**context) -> None:
 def _prepare_topics(**context) -> None:
     """Подготовка топиков Kafka (создание/сброс)."""
     conf = context.get("dag_run", {}).conf or {}
-    reset_topics = bool(conf.get("reset_topics", context["params"]["reset_topics"]))
+    reset_topics = parse_bool_param(
+        conf.get("reset_topics", context["params"]["reset_topics"]),
+        "reset_topics",
+    )
 
     prepare_topics(reset=reset_topics)
 

airflow/dags/utils/airflow_params.py

@@ -0,0 +1,33 @@
+"""Общие helper-функции для параметров Airflow DAG."""
+
+from __future__ import annotations
+
+TRUE_VALUES = {"1", "true", "t", "yes", "y", "on"}
+FALSE_VALUES = {"0", "false", "f", "no", "n", "off"}
+
+
+def _param_error(message: str) -> Exception:
+    try:
+        from airflow.exceptions import AirflowException
+
+        return AirflowException(message)
+    except ModuleNotFoundError:
+        return ValueError(message)
+
+
+def parse_bool_param(value: object, name: str) -> bool:
+    """Преобразует bool-параметр из dag_run.conf/params в явный boolean."""
+    if isinstance(value, bool):
+        return value
+
+    if isinstance(value, int) and value in (0, 1):
+        return bool(value)
+
+    if isinstance(value, str):
+        normalized = value.strip().lower()
+        if normalized in TRUE_VALUES:
+            return True
+        if normalized in FALSE_VALUES:
+            return False
+
+    raise _param_error(f"Параметр {name} должен быть boolean, получено: {value!r}")

airflow/dags/utils/sql_helpers.py

@@ -0,0 +1,153 @@
+"""Общие helper-функции для чтения SQL-файлов из Airflow DAG."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+
+def _sql_error(message: str) -> Exception:
+    try:
+        from airflow.exceptions import AirflowException
+
+        return AirflowException(message)
+    except ModuleNotFoundError:
+        return ValueError(message)
+
+
+def strip_sql_comments(sql_text: str) -> str:
+    """Удаляет SQL-комментарии, не трогая строки в кавычках."""
+    result: list[str] = []
+    i = 0
+    in_single_quote = False
+    in_double_quote = False
+
+    while i < len(sql_text):
+        char = sql_text[i]
+        next_char = sql_text[i + 1] if i + 1 < len(sql_text) else ""
+
+        if in_single_quote:
+            result.append(char)
+            if char == "'" and next_char == "'":
+                result.append(next_char)
+                i += 2
+                continue
+            if char == "'" and (i == 0 or sql_text[i - 1] != "\\"):
+                in_single_quote = False
+            i += 1
+            continue
+
+        if in_double_quote:
+            result.append(char)
+            if char == '"' and (i == 0 or sql_text[i - 1] != "\\"):
+                in_double_quote = False
+            i += 1
+            continue
+
+        if char == "'":
+            in_single_quote = True
+            result.append(char)
+            i += 1
+            continue
+
+        if char == '"':
+            in_double_quote = True
+            result.append(char)
+            i += 1
+            continue
+
+        if char == "-" and next_char == "-":
+            i += 2
+            while i < len(sql_text) and sql_text[i] not in "\r\n":
+                i += 1
+            continue
+
+        if char == "/" and next_char == "*":
+            i += 2
+            while (
+                i < len(sql_text) - 1
+                and not (sql_text[i] == "*" and sql_text[i + 1] == "/")
+            ):
+                if sql_text[i] in "\r\n":
+                    result.append(sql_text[i])
+                i += 1
+            i += 2
+            continue
+
+        result.append(char)
+        i += 1
+
+    return "".join(result)
+
+
+def split_sql_statements(sql_text: str) -> tuple[str, ...]:
+    """Делит SQL на команды по ';' вне строковых литералов."""
+    statements: list[str] = []
+    current: list[str] = []
+    cleaned_sql = strip_sql_comments(sql_text)
+    in_single_quote = False
+    in_double_quote = False
+    i = 0
+
+    while i < len(cleaned_sql):
+        char = cleaned_sql[i]
+        next_char = cleaned_sql[i + 1] if i + 1 < len(cleaned_sql) else ""
+
+        if in_single_quote:
+            current.append(char)
+            if char == "'" and next_char == "'":
+                current.append(next_char)
+                i += 2
+                continue
+            if char == "'" and (i == 0 or cleaned_sql[i - 1] != "\\"):
+                in_single_quote = False
+            i += 1
+            continue
+
+        if in_double_quote:
+            current.append(char)
+            if char == '"' and (i == 0 or cleaned_sql[i - 1] != "\\"):
+                in_double_quote = False
+            i += 1
+            continue
+
+        if char == "'":
+            in_single_quote = True
+            current.append(char)
+            i += 1
+            continue
+
+        if char == '"':
+            in_double_quote = True
+            current.append(char)
+            i += 1
+            continue
+
+        if char == ";":
+            statement = "".join(current).strip()
+            if statement:
+                statements.append(statement)
+            current = []
+            i += 1
+            continue
+
+        current.append(char)
+        i += 1
+
+    statement = "".join(current).strip()
+    if statement:
+        statements.append(statement)
+
+    return tuple(statements)
+
+
+def load_sql_statements(sql_root: Path, relative_path: str) -> tuple[str, ...]:
+    """Читает SQL-файл и возвращает отдельные команды."""
+    file_path = sql_root / relative_path
+    if not file_path.is_file():
+        raise _sql_error(f"SQL-файл не найден: {file_path}")
+
+    statements = split_sql_statements(file_path.read_text(encoding="utf-8"))
+    if not statements:
+        raise _sql_error(f"SQL-файл пустой: {file_path}")
+
+    return statements