ربط Apache Airflow بـ ClickHouse - ClickHouse Documentation

تُعد 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

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

الحقل	الوصف	القيمة الافتراضية
Host	اسم مضيف خادم ClickHouse، على سبيل المثال `abc123.clickhouse.cloud`	`localhost`
Port	منفذ HTTP(S)	`8123` (عادي)، `8443` (TLS)
Login	اسم مستخدم ClickHouse	`default`
Password	كلمة مرور مستخدم ClickHouse	(فارغة)
Database	قاعدة البيانات الافتراضية للاتصال. تعرض واجهة المستخدم هذا الحقل باسم Database؛ وهو الحقل `schema` عند تعريف الاتصال باستخدام URI أو JSON.	`default`

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

خيارات الاتصال الإضافية

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

`extra` key	UI field	Default	Description
`secure`	استخدام TLS ‏(HTTPS)	`false`	يفعّل HTTPS/TLS.
`verify`	التحقق من SSL certificate	`true`	يتحقق من certificate الخاصة بـ TLS على server عندما تكون قيمة `secure` هي `true`. اضبطه على `false` عند استخدام شهادات موقّعة ذاتيًا.
`connect_timeout`	مهلة الاتصال (بالثواني)	`10`	مهلة HTTP اتصال بالثواني.
`send_receive_timeout`	مهلة الاستعلام (seconds)	`300`	مهلة القراءة/الكتابة الخاصة بالاستعلام بالثواني. زِد هذه القيمة للاستعلامات التحليلية طويلة التشغيل.
`compress`	تفعيل ضغط LZ4	`true`	يفعّل ضغط نتائج LZ4.
`client_name`	اسم Client	(empty)	تسمية تُلحَق بمعرّف إصدار Airflow في `User-Agent` الخاص بـ ClickHouse وفي العمود `client_name` من `system.query_log`.
`session_settings`	إعدادات الجلسة (JSON)	(empty)	إعدادات الجلسة في ClickHouse التي تُطبَّق على كل استعلام في الاتصال، مثل `{"max_execution_time": 300, "max_threads": 8}`.
`client_kwargs`	Client kwargs (JSON)	(empty)	keyword arguments إضافية تُمرَّر إلى `clickhouse_connect.get_client()`، مثل `http_proxy`.

حدِّد اتصالًا من دون واجهة المستخدم

اضبط الاتصال عبر متغيّر بيئة. ويشمل تنسيق 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 ما لم تحدد معرّفًا آخر.

تشغيل الاستعلامات باستخدام SQLExecuteQueryOperator

اضبط قيمة 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

​خيارات الاتصال الإضافية

​حدِّد اتصالًا من دون واجهة المستخدم

​تشغيل الاستعلامات باستخدام SQLExecuteQueryOperator

​استهدف قاعدة بيانات مختلفة لكل مهمة

​استخدم الخطاف مباشرة

​تطبيق إعدادات الجلسة

​محتوى مرتبط