""" DAG ETL-процесса STG -> ODS -> DDS -> DM для учебного проекта. Принципы реализации: - SQL выполняется явными task на ClickHouseOperator; - SQL-файлы вызываются по фиксированным путям; - Python используется только для управляющей логики (branch/wait/assert). """ from __future__ import annotations import re import time 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.empty import EmptyOperator from airflow.operators.python import BranchPythonOperator, PythonOperator 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 # ----------------------------------------------------------------------------- # Базовые настройки DAG # ----------------------------------------------------------------------------- default_args = { "owner": "airflow", "depends_on_past": False, "email_on_failure": False, "email_on_retry": False, "retries": 1, "retry_delay": timedelta(minutes=2), } # ----------------------------------------------------------------------------- # SQL-файлы проекта # ----------------------------------------------------------------------------- SQL_ROOT = Path(__file__).resolve().parents[1] / "sql" 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) # ----------------------------------------------------------------------------- # SQL для проверок и технических шагов # ----------------------------------------------------------------------------- SQL_CHECK_CLICKHOUSE = "SELECT 1 AS ok" SQL_CHECK_SCHEMA_READY = """ SELECT (SELECT count() FROM system.tables WHERE database = 'stg' AND name = 'browser_raw') AS stg_browser_raw, (SELECT count() FROM system.tables WHERE database = 'ods' AND name = 'browser_event') AS ods_browser_event, (SELECT count() FROM system.tables WHERE database = 'dds' AND name = 'event') AS dds_event, (SELECT count() FROM system.tables WHERE database = 'dm' AND name = 'v_events_enriched') AS dm_v_events_enriched """ SQL_CHECK_ODS_QUALITY = """ SELECT table_name, total_rows, rows_with_errors, round(if(total_rows = 0, 0, rows_with_errors / total_rows * 100), 2) AS error_pct FROM ( SELECT 'browser_event' AS table_name, toFloat64(count()) AS total_rows, toFloat64(countIf(length(parse_errors) > 0)) AS rows_with_errors FROM ods.browser_event UNION ALL SELECT 'location_event', toFloat64(count()), toFloat64(countIf(length(parse_errors) > 0)) FROM ods.location_event UNION ALL SELECT 'device_by_click', toFloat64(count()), toFloat64(countIf(length(parse_errors) > 0)) FROM ods.device_by_click UNION ALL SELECT 'geo_by_click', toFloat64(count()), toFloat64(countIf(length(parse_errors) > 0)) FROM ods.geo_by_click UNION ALL SELECT 'browser_event_errors', toFloat64(count()), toFloat64(count()) FROM ods.browser_event_errors UNION ALL SELECT 'location_event_errors', toFloat64(count()), toFloat64(count()) FROM ods.location_event_errors UNION ALL SELECT 'device_by_click_errors', toFloat64(count()), toFloat64(count()) FROM ods.device_by_click_errors UNION ALL SELECT 'geo_by_click_errors', toFloat64(count()), toFloat64(count()) FROM ods.geo_by_click_errors ) ORDER BY table_name """ SQL_TRUNCATE_DDS_CLICK = "TRUNCATE TABLE dds.click" SQL_TRUNCATE_DDS_EVENT = "TRUNCATE TABLE dds.event" SQL_CHECK_DDS_INTEGRITY = """ SELECT countIf(click_id IS NOT NULL AND click_id NOT IN (SELECT click_id FROM dds.click)) AS orphan_events FROM dds.event """ SQL_VALIDATE_DM_SUMMARY = "SELECT count() AS dq_rows FROM dm.dq_summary" # ----------------------------------------------------------------------------- # Управляющие функции # ----------------------------------------------------------------------------- def assert_schema_ready(**context) -> None: """Падает, если DDL не применён полностью.""" ti = context["ti"] result = ti.xcom_pull(task_ids="precheck.check_schema_ready_sql") if not result or not result[0] or len(result[0]) != 4: raise AirflowException(f"Некорректный результат check_schema_ready_sql: {result}") if any(value == 0 for value in result[0]): raise AirflowException( "Схема не готова: сначала запустите DAG ddl_init, затем повторите etl_pipeline." ) def wait_for_stg_data(**context) -> None: """ Ожидает появления строк в STG до заданного таймаута. Таймаут берётся из dag_run.conf.wait_stg_timeout_sec (или legacy wait_ods_timeout_sec) либо из params. """ dag_run = context.get("dag_run") conf = dag_run.conf if dag_run else {} timeout_sec = int( conf.get( "wait_stg_timeout_sec", conf.get( "wait_ods_timeout_sec", context["params"]["wait_stg_timeout_sec"], ), ) ) poll_interval_sec = 10 hook = ClickHouseHook(clickhouse_conn_id="clickhouse_default", database="default") started = time.monotonic() while True: rows = hook.execute( """ SELECT (SELECT count() FROM stg.browser_raw) + (SELECT count() FROM stg.location_raw) + (SELECT count() FROM stg.device_raw) + (SELECT count() FROM stg.geo_raw) AS stg_rows_total """ ) count_rows = int(rows[0][0]) if rows else 0 if count_rows > 0: return elapsed = int(time.monotonic() - started) if elapsed >= timeout_sec: raise AirflowException( f"Таймаут ожидания STG истёк ({timeout_sec} сек). " "Таблицы stg.*_raw всё ещё пусты." ) time.sleep(poll_interval_sec) 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"])) return "transform.truncate_dds_click" if full_refresh else "transform.skip_truncate" def assert_dm_summary_not_empty(**context) -> None: """Проверяет, что dm.dq_summary заполнена после загрузки.""" ti = context["ti"] result = ti.xcom_pull(task_ids="transform.validate_dm_summary_sql") if not result or not result[0] or len(result[0]) != 1: raise AirflowException(f"Некорректный результат validate_dm_summary_sql: {result}") dq_rows = int(result[0][0]) if dq_rows <= 0: raise AirflowException("dm.dq_summary пуста после load_dm_summary.") with DAG( dag_id="etl_pipeline", description="ETL STG -> ODS -> DDS -> DM для demo-проекта", default_args=default_args, schedule=None, start_date=datetime(2024, 1, 1), catchup=False, max_active_runs=1, is_paused_upon_creation=True, tags=["etl", "clickhouse", "demo"], params={ "full_refresh": Param(True, type="boolean"), "wait_stg_timeout_sec": Param(600, type="integer", minimum=30), }, ) as dag: with TaskGroup(group_id="precheck") as precheck: check_clickhouse = ClickHouseOperator( task_id="check_clickhouse", sql=SQL_CHECK_CLICKHOUSE, clickhouse_conn_id="clickhouse_default", database="default", ) check_schema_ready_sql = ClickHouseOperator( task_id="check_schema_ready_sql", sql=SQL_CHECK_SCHEMA_READY, clickhouse_conn_id="clickhouse_default", database="default", ) check_schema_ready = PythonOperator( task_id="check_schema_ready", python_callable=assert_schema_ready, ) check_clickhouse >> check_schema_ready_sql >> check_schema_ready with TaskGroup(group_id="transform") as transform: wait_for_stg_data_task = PythonOperator( task_id="wait_for_stg_data", python_callable=wait_for_stg_data, ) load_ods = ClickHouseOperator( task_id="load_ods", sql=load_sql_statements("ods/20_stg_to_ods.sql"), clickhouse_conn_id="clickhouse_default", database="default", ) check_ods_quality = ClickHouseOperator( task_id="check_ods_quality", sql=SQL_CHECK_ODS_QUALITY, clickhouse_conn_id="clickhouse_default", database="default", ) choose_refresh_mode = BranchPythonOperator( task_id="choose_refresh_mode", python_callable=choose_full_refresh, ) truncate_dds_click = ClickHouseOperator( task_id="truncate_dds_click", sql=SQL_TRUNCATE_DDS_CLICK, clickhouse_conn_id="clickhouse_default", database="default", ) truncate_dds_event = ClickHouseOperator( task_id="truncate_dds_event", sql=SQL_TRUNCATE_DDS_EVENT, clickhouse_conn_id="clickhouse_default", database="default", ) skip_truncate = EmptyOperator(task_id="skip_truncate") truncate_complete = EmptyOperator( task_id="truncate_complete", trigger_rule=TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS, ) load_dds = ClickHouseOperator( task_id="load_dds", sql=load_sql_statements("dds/30_ods_to_dds.sql"), clickhouse_conn_id="clickhouse_default", database="default", ) check_dds_integrity = ClickHouseOperator( task_id="check_dds_integrity", sql=SQL_CHECK_DDS_INTEGRITY, clickhouse_conn_id="clickhouse_default", database="default", ) load_dm_summary = ClickHouseOperator( task_id="load_dm_summary", sql=load_sql_statements("dm/40_dds_to_dm.sql"), clickhouse_conn_id="clickhouse_default", database="default", ) validate_dm_summary_sql = ClickHouseOperator( task_id="validate_dm_summary_sql", sql=SQL_VALIDATE_DM_SUMMARY, clickhouse_conn_id="clickhouse_default", database="default", ) validate_dm_summary = PythonOperator( task_id="validate_dm_summary", python_callable=assert_dm_summary_not_empty, ) 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 precheck >> transform