diff --git a/test_clickhouse_dag.py b/test_clickhouse_dag.py index 4fceccf..538247e 100644 --- a/test_clickhouse_dag.py +++ b/test_clickhouse_dag.py @@ -1,23 +1,23 @@ from airflow import DAG -from airflow.operators.python import PythonOperator -from datetime import datetime +from airflow.providers.clickhouse.operators.clickhouse import ClickHouseOperator +from airflow.utils.dates import days_ago -# Импортируем hook из установленного пакета airflow-providers-clickhouse -from airflow_clickhouse_plugin.hooks.clickhouse import ClickHouseHook - -def test_conn(): - hook = ClickHouseHook( - clickhouse_conn_id="my_clickhouse_provider" # ваш Connection ID в Airflow - ) - hook.execute("SELECT 1") # у этого пакета метод называется execute +default_args = { + "owner": "airflow", +} with DAG( - dag_id="test_clickhouse_dag", - start_date=datetime(2024, 1, 1), - schedule=None, + dag_id="test_clickhouse_select", + default_args=default_args, + description="Simple SELECT 1 in ClickHouse", + start_date=days_ago(1), + schedule_interval=None, # запуск только вручную catchup=False, + tags=["clickhouse"], ) as dag: - PythonOperator( - task_id="check_clickhouse", - python_callable=test_conn, + + select_1 = ClickHouseOperator( + task_id="select_one", + clickhouse_conn_id="my_clickhouse_provider", # ваш Connection ID + sql="SELECT 1", )