ربط Apache Airflow بـ ClickHouse

تُعد Apache Airflow منصة مفتوحة المصدر لتأليف سير العمل وجدولته ومراقبته على شكل شيفرة. ويُعرَّف سير العمل على هيئة رسوم بيانية موجهة لا دورية (DAGs) تتكوّن من مهام مكتوبة بلغة بايثون.

يربط موفّر apache-airflow-providers-clickhousedb بين Airflow وClickHouse، مما يتيح لك تشغيل الاستعلامات وإنشاء الجداول وتحميل البيانات كجزء من DAG. ويتصل عبر واجهة HTTP باستخدام عميل clickhouse-connect، كما يوفّر ClickHouse من خلال إطار SQL الشائع في Airflow، بحيث يتولى SQLExecuteQueryOperator القياسي تنفيذ استعلامات DDL وDML والاستعلامات التحليلية من دون الحاجة إلى مشغّل خاص بـ ClickHouse.

ثبّت الموفّر في البيئة التي يعمل فيها مجدول Airflow والعمّال لديك:

pip install apache-airflow-providers-clickhousedb

يعتمد هذا الموفّر على apache-airflow-providers-common-sql وclickhouse-connect، ويُثبَّتان معه. لتمرير نتائج الاستعلام إلى كائنات DataFrame في pandas أو polars، ثبّت الإضافات الاختيارية التالية:

pip install 'apache-airflow-providers-common-sql[pandas,polars]'

يسجّل الموفّر نوع اتصال باسم clickhouse. أنشئ اتصالًا من واجهة Airflow ضمن Admin > Connections، أو عرّفه عبر CLI أو من خلال متغير بيئة.

في واجهة المستخدم، اختر ClickHouse كنوع الاتصال واملأ الحقول التالية:

بالنسبة إلى ClickHouse Cloud أو أي عنقود مستضاف ذاتيًا مع تمكين TLS، اضبط secure على true في حقل Extra واستخدم منفذ TLS (8443).

يوفّر الموفّر خيارات إضافية كحقول مخصّصة في نموذج الاتصال. وإذا عرّفت الاتصال باستخدام URI أو JSON أو متغيّر بيئة بدلًا من ذلك، فأرسلها كمفاتيح ضمن كائن JSON ‏extra. جميعها اختيارية:

اضبط الاتصال عبر متغيّر بيئة. ويشمل تنسيق URI كلًا من المضيف وبيانات الاعتماد وقاعدة البيانات:

export AIRFLOW_CONN_CLICKHOUSE_DEFAULT='clickhouse://default:password@localhost:8123/my_database'

يجب ترميز جميع مكوّنات URI باستخدام ترميز URL. بالنسبة إلى TLS ومهلات الانتظار وإعدادات الجلسة، استخدم صيغة JSON التي تُظهر حقول Extra:

export AIRFLOW_CONN_CLICKHOUSE_DEFAULT='{
    "conn_type": "clickhouse",
    "host": "abc123.clickhouse.cloud",
    "port": 8443,
    "login": "default",
    "password": "secret",
    "schema": "my_database",
    "extra": {
        "secure": true,
        "session_settings": {
            "max_execution_time": 300,
            "max_memory_usage": 10000000000
        }
    }
}'

تستخدم جميع الـ hooks والـ operators معرّف الاتصال clickhouse_default ما لم تحدد معرّفًا آخر.

اضبط قيمة conn_id للمشغّل بحيث تشير إلى اتصال ClickHouse لديك. ينشئ DAG التالي جدولًا، ويدرج صفوفًا، ثم يقرؤها مرة أخرى، ثم يحذف الجدول:

from datetime import datetime

from airflow import DAG
from airflow.providers.common.sql.hooks.sql import fetch_all_handler
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator

CLICKHOUSE_CONN_ID = "clickhouse_default"
CLICKHOUSE_TABLE = "airflow_example"

with DAG(
    dag_id="example_clickhouse",
    start_date=datetime(2021, 1, 1),
    default_args={"conn_id": CLICKHOUSE_CONN_ID},
    schedule="@once",
    catchup=False,
) as dag:
    create_table = SQLExecuteQueryOperator(
        task_id="create_table",
        sql=f"""
            CREATE TABLE IF NOT EXISTS {CLICKHOUSE_TABLE} (
                id   UInt32,
                name String,
                ts   DateTime DEFAULT now()
            ) ENGINE = MergeTree()
            ORDER BY id
        """,
    )

    insert_rows = SQLExecuteQueryOperator(
        task_id="insert_rows",
        sql=f"""
            INSERT INTO {CLICKHOUSE_TABLE} (id, name) VALUES
                (1, 'Alice'),
                (2, 'Bob'),
                (3, 'Charlie')
        """,
    )

    read_rows = SQLExecuteQueryOperator(
        task_id="read_rows",
        sql=f"SELECT id, name FROM {CLICKHOUSE_TABLE} ORDER BY id",
        handler=fetch_all_handler,
    )

    drop_table = SQLExecuteQueryOperator(
        task_id="drop_table",
        sql=f"DROP TABLE IF EXISTS {CLICKHOUSE_TABLE}",
    )

    create_table >> insert_rows >> read_rows >> drop_table

تُسترجَع نتائج الاستعلام باستخدام handler الافتراضي (fetch_all_handler). ولإرجاع شيء آخر غير مجموعة النتائج الكاملة، مرِّر معالجًا مختلفًا، مثل fetch_one_handler لإرجاع الصف الأول فقط.

عندما يكون اتصال واحد موجّهًا إلى عنقود وتنفّذ المهام الفردية استعلامات على قواعد بيانات مختلفة، فعيّن قاعدة البيانات عبر hook_params بدلًا من إنشاء اتصال منفصل:

read_rows = SQLExecuteQueryOperator(
    task_id="read_rows",
    conn_id=CLICKHOUSE_CONN_ID,
    sql="SELECT count() FROM events",
    hook_params={"database": "analytics"},
)

بالنسبة إلى الأعمال التي لا تناسب مُعامل SQL — مثل الإدراجات المجمّعة، أو التدفق، أو استدعاءات العميل الخاصة بـ ClickHouse — استخدم ClickHouseHook داخل مهمة في بايثون.

تستخدم الطريقة bulk_insert_rows الخاصة بالخطاف مسار الإدراج العمودي الأصلي في clickhouse-connect، وهو أسرع بكثير من الإدراج صفًا بصف عند التعامل مع مجموعات بيانات كبيرة. اضبط batch_size للحد من ذروة الذاكرة عند التعامل مع مدخلات كبيرة جدًا:

from airflow.providers.clickhousedb.hooks.clickhouse import ClickHouseHook

hook = ClickHouseHook(clickhouse_conn_id="clickhouse_default")

hook.bulk_insert_rows(
    table="events",
    rows=[("user1", "click"), ("user2", "view")],
    column_names=["user_id", "action"],
    batch_size=1000,
)

استدعِ get_client() للوصول إلى عميل clickhouse-connect الداخلي لأي شيء لا يتيحه الخطاف مباشرةً:

client = hook.get_client()
total = client.query("SELECT count() FROM events").result_rows[0][0]

مرِّر إعدادات الجلسة عند إنشاء الـ خطاف، إما مباشرةً أو عبر hook_params الخاص بالمشغّل. وتُدمَج الإعدادات المُمرَّرة إلى المُنشئ مع أي session_settings مُعرَّفة في حقل Extra الخاص بالاتصال، وتكون لقيم المُنشئ الأولوية عند تعارض المفاتيح:

hook = ClickHouseHook(
    clickhouse_conn_id="clickhouse_default",
    session_settings={"max_execution_time": 60, "max_threads": 4},
)

• clickhouse-connect عميل بايثون
• واجهة ClickHouse HTTP
• مرجع إعدادات جلسة ClickHouse
• الوثائق المرجعية لـ apache-airflow-providers-clickhousedb
• حزمة الموفّر على PyPI