ربط Apache Airflow بـ ClickHouse

نسّق استعلامات ClickHouse وعمليات تحميل البيانات من Apache Airflow باستخدام موفّر ClickHouse

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

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

هل تستخدم بالفعل airflow-clickhouse-plugin المجتمعي؟ إنه يستخدم driver وprotocol ومنفذًا مختلفين. اتبع دليل الترحيل لنقل DAGs والاتصالات الحالية إلى الموفّر.

​

ثبّت الموفّر

ثبّت الموفّر في البيئة التي يعمل فيها مجدول 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	true	يتحقق من الشهادة الخاصة بـ TLS على خادم عندما تكون قيمة secure هي true. اضبطه على false عند استخدام شهادات موقّعة ذاتيًا.
connect_timeout	مهلة الاتصال (بالثواني)	10	مهلة HTTP اتصال بالثواني.
send_receive_timeout	مهلة الاستعلام (seconds)	300	مهلة القراءة/الكتابة الخاصة بالاستعلام بالثواني. زِد هذه القيمة للاستعلامات التحليلية طويلة التشغيل.
compress	تفعيل ضغط LZ4	true	يفعّل ضغط نتائج LZ4.
client_name	اسم Client	(empty)	تسمية تُلحَق بمعرّف إصدار Airflow في header ‏User-Agent الخاص بـ HTTP، وتُسجَّل في العمود http_user_agent من 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.handlers 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},
)

​

محتوى مرتبط

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