الترحيل من airflow-clickhouse-plugin إلى مزود ClickHouse

نقل DAGs الحالية في Airflow من حزمة المجتمع airflow-clickhouse-plugin إلى المزود الرسمي apache-airflow-providers-clickhousedb

قبل ظهور مزود ClickHouse الرسمي، كان معظم مستخدمي Airflow يتصلون بـ ClickHouse عبر حزمة المجتمع airflow-clickhouse-plugin. يستعرض هذا الدليل خطوات ترحيل عملية نشر قائمة إلى apache-airflow-providers-clickhousedb. ولمعرفة كيفية عمل المزود نفسه، راجع ربط Apache Airflow بـ ClickHouse.

​

لماذا مزوّد أصلي؟

المزوّدات (Providers) هي الطريقة التي يتكامل بها Airflow مع الأنظمة الخارجية. ويتم إصدارها واختبارها وتوثيقها مع بقية منظومة Airflow، وهذا المزوّد يتولى صيانته مجتمع Airflow بالتعاون مع ClickHouse team. وهو مبني على clickhouse-connect، وهو عميل بايثون الذي تطوّره ClickHouse وتدعمه بنفسها، لا على driver يصونه المجتمع. وبذلك تصل ميزات الخادم الجديدة وإصلاحاته إلى مستخدمي Airflow عبر مسار مدعوم. ويمنحك الانتقال إلى المزوّد package له موطن رسمي، وعوامل common.sql القياسية ومستشعراتها، ونوع اتصال يظهر في UI الخاص بـ Airflow كما هو الحال مع أي database أخرى.

ولا يقتصر الاختلاف بين الحزمتين على مسارات import. فالـ plugin يتواصل مع ClickHouse عبر بروتوكول TCP الأصلي باستخدام clickhouse-driver، أما المزوّد فيتواصل عبر HTTP(S) باستخدام clickhouse-connect ويندمج مع عوامل common.sql العامة بدلاً من توفير عوامل مخصصة لـ ClickHouse.

اقرأ الدليل بالكامل مرة واحدة قبل تغيير أي شيء. فتغيير الاتصال على وجه الخصوص يؤثر على كل DAG في الوقت نفسه.

​

لمحة سريعة

	airflow-clickhouse-plugin	apache-airflow-providers-clickhousedb
جذر الاستيراد	airflow_clickhouse_plugin	airflow.providers.clickhousedb
برنامج التشغيل (Driver)	clickhouse-driver	clickhouse-connect
البروتوكول / المنفذ الافتراضي	Native TCP، 9000 (9440 مع TLS)	HTTP، 8123 (8443 مع TLS)
نوع الاتصال	لا يوجد نوع مُسجَّل؛ أي نوع يفي بالغرض	clickhouse
إضافات الاتصال	تُمرَّر حرفيًا إلى clickhouse_driver.Client	مجموعة ثابتة من المفاتيح، انظر خيارات الاتصال الإضافية
المُشغِّلات والمستشعرات	ClickHouseOperator وClickHouseSensor إضافة إلى أغلفة common.sql المسبوقة بـ ClickHouse	مُشغِّلات ومستشعرات common.sql تُستخدم مباشرةً
الخطاف	ClickHouseHook (BaseHook) وClickHouseDbApiHook (DbApiHook)	خطاف واحد ClickHouseHook (DbApiHook)
الحد الأدنى لإصدار Airflow	2.0	2.11

​

الخطوة 1: التحقق من المتطلبات المسبقة والتثبيت

يتطلب المزوّد إصدار Airflow 2.11 أو أحدث، والإصدار 1.32.0 أو أحدث من apache-airflow-providers-common-sql. قم بترقية Airflow أولاً إذا كنت تستخدم إصداراً أقدم.

pip install apache-airflow-providers-clickhousedb

توجد الحزمتان في مساحتي أسماء مختلفتين في بايثون، لذا يمكن تثبيتهما جنبًا إلى جنب أثناء ترحيل كل DAG على حدة. أزل الـ plugin بعد أن لا يعود أي شيء يستورده:

pip uninstall airflow-clickhouse-plugin clickhouse-driver

​

الخطوة 2: تحديث الاتصالات

هذه هي الخطوة التي يؤدي تجاوزها إلى تعطّل الأمور. فكل اتصال ClickHouse قائم يشير إلى المنفذ native، بينما يحتاج provider إلى منفذ HTTP.

Field	Plugin	Provider
نوع الاتصال	أي شيء (غالبًا sqlite أو generic)	clickhouse
المنفذ، عادي	9000	8123
المنفذ، TLS	9440	8443
Login	القيمة الافتراضية للـ driver وهي default	نفسها
Schema	قاعدة البيانات	نفسها

إذا كان ClickHouse خلف firewall أو load balancer، فتأكّد من إمكانية الوصول إلى منفذ HTTP من الـ workers قبل التبديل. وتحقّق من أن HTTP interface مُمكّن على الخادم (http_port أو https_port في تهيئة الخادم). أما ClickHouse Cloud فلا يعرض HTTPS إلا على المنفذ 8443.

أما الاتصالات المخزّنة كعناوين URI (clickhouse://user:pass@host:9000/db?secure=true) فهي تحمل بالفعل نوع الاتصال clickhouse لأن Airflow يستنتجه من الـ scheme في عنوان URI. ولا يتغيّر فيها سوى المنفذ والـ keys الإضافية؛ وتُحلَّل قيم سلسلة الاستعلام على أنها JSON، لذا تبقى secure=true قيمة منطقية.

​

إضافات الاتصال

كان الـ plugin يمرّر كل مفتاح في extra مباشرةً إلى clickhouse_driver.Client، لذا قد تحمل الاتصالات أي وسيطات كلمات مفتاحية خاصة بـ clickhouse-driver. أما الـ provider فلا يقرأ سوى مجموعة ثابتة من المفاتيح ويمرّر ما عداها عبر client_kwargs. حوّلها على النحو التالي:

إضافة الـ plugin (clickhouse-driver)	إضافة الـ provider	ملاحظات
secure	secure	دون تغيير. تذكّر تغيير الـ Port أيضاً.
verify	verify	دون تغيير.
settings	session_settings	المحتوى نفسه، مفتاح جديد.
compression	compress	قيمة Boolean. تختلف أسماء الخوارزميات بين الـ drivers؛ وtrue هو الخيار الآمن.
connect_timeout	connect_timeout	دون تغيير.
send_receive_timeout	send_receive_timeout	دون تغيير.
client_name	client_name	تتغيّر الدلالة. يرسل الـ provider دائماً apache-airflow/<version> apache-airflow-providers-clickhousedb/<version> ويُلحِق قيمتك بوصفها label.
ca_certs	client_kwargs.ca_cert	الـ path إلى الـ CA bundle.
certfile / keyfile	client_kwargs.client_cert / client_kwargs.client_cert_key	mutual TLS.
server_hostname	client_kwargs.server_host_name	SNI override الخاص بـ TLS.
alt_hosts، round_robin	لا مكافئ	يتصل العميل بالمضيف الواحد المحدّد في الاتصال. إن كنت تعتمد على قائمة الـ failover، فوجّه الاتصال إلى الـ load balancer الخاص بك أو إلى ClickHouse Cloud نقطة نهاية.
sync_request_timeout، tcp_keepalive، compress_block_size	حذف	خاصة بالبروتوكول الأصلي فقط.
ssl_version، ciphers، use_numpy، client_revision، settings_is_important، opentelemetry_traceparent، opentelemetry_tracestate	حذف	لا مكافئ لها في clickhouse-connect.

قبل:

{
    "conn_type": "sqlite",
    "host": "ch.example.com",
    "port": 9440,
    "login": "airflow",
    "password": "secret",
    "schema": "analytics",
    "extra": {
        "secure": true,
        "settings": {"max_execution_time": 300},
        "compression": true
    }
}

بعد:

{
    "conn_type": "clickhouse",
    "host": "ch.example.com",
    "port": 8443,
    "login": "airflow",
    "password": "secret",
    "schema": "analytics",
    "extra": {
        "secure": true,
        "session_settings": {"max_execution_time": 300},
        "compress": true
    }
}

تحقّق من كل اتصال مُرحَّل قبل تعديل DAGs:

airflow connections test clickhouse_default

​

الخطوة 3: استبدال عمليات الاستيراد

صنف الـ plugin	البديل في الـ provider
airflow_clickhouse_plugin.hooks.clickhouse.ClickHouseHook	airflow.providers.clickhousedb.hooks.clickhouse.ClickHouseHook
airflow_clickhouse_plugin.hooks.clickhouse_dbapi.ClickHouseDbApiHook	airflow.providers.clickhousedb.hooks.clickhouse.ClickHouseHook
airflow_clickhouse_plugin.operators.clickhouse.ClickHouseOperator	airflow.providers.common.sql.operators.sql.SQLExecuteQueryOperator
airflow_clickhouse_plugin.sensors.clickhouse.ClickHouseSensor	airflow.providers.common.sql.sensors.sql.SqlSensor
airflow_clickhouse_plugin.operators.clickhouse_dbapi.ClickHouseSQLExecuteQueryOperator	airflow.providers.common.sql.operators.sql.SQLExecuteQueryOperator
...clickhouse_dbapi.ClickHouseSQLCheckOperator	...common.sql.operators.sql.SQLCheckOperator
...clickhouse_dbapi.ClickHouseSQLValueCheckOperator	...common.sql.operators.sql.SQLValueCheckOperator
...clickhouse_dbapi.ClickHouseSQLIntervalCheckOperator	...common.sql.operators.sql.SQLIntervalCheckOperator
...clickhouse_dbapi.ClickHouseSQLThresholdCheckOperator	...common.sql.operators.sql.SQLThresholdCheckOperator
...clickhouse_dbapi.ClickHouseSQLColumnCheckOperator	...common.sql.operators.sql.SQLColumnCheckOperator
...clickhouse_dbapi.ClickHouseSQLTableCheckOperator	...common.sql.operators.sql.SQLTableCheckOperator
...clickhouse_dbapi.ClickHouseBranchSQLOperator	...common.sql.operators.sql.BranchSQLOperator
airflow_clickhouse_plugin.sensors.clickhouse_dbapi.ClickHouseSqlSensor	airflow.providers.common.sql.sensors.sql.SqlSensor

لم يكن الغرض من الـ wrappers الخاصة بـ common.sql والمسبوقة بـ ClickHouse سوى حقن الخطاف الخاص بالـ plugin. أما الـ provider فيسجّل نوع الاتصال clickhouse، ومن ثم تستطيع أصناف common.sql غير المسبوقة تحديد الخطاف من الاتصال بنفسها. فإذا كنت تستخدم تلك الـ wrappers، يقتصر الترحيل عادةً على تعديل سطر الـ import، وحذف السابقة ClickHouse، وتمرير conn_id بشكل صريح (انظر الخطوة 7).

​

الخطوة 4: من ClickHouseOperator إلى SQLExecuteQueryOperator

معامل ClickHouseOperator	المكافئ في SQLExecuteQueryOperator	ملاحظات
sql	sql	سلسلة نصية، أو قائمة سلاسل نصية، أو مسار ملف .sql. يدعم القوالب في كليهما.
clickhouse_conn_id	conn_id	كان البرنامج الإضافي يستخدم clickhouse_default كقيمة افتراضية. أما SQLExecuteQueryOperator فليست له قيمة افتراضية: مرّر conn_id في كل مهمة أو عيّنه مرة واحدة في default_args.
database	database	دون تغيير.
parameters لعبارة SELECT	parameters	تظل العناصر النائبة %(name)s تعمل. كما يتوفر الربط على جانب الخادم بصيغة {name:Type}.
parameters لعبارة INSERT (قائمة صفوف)	SQLInsertRowsOperator	مرّر الصفوف (أو قيمة XCom) عبر rows، وسمِّ الـ column باستخدام columns، واضبط insert_args={"executemany": True} كي يستخدم clickhouse-connect الإدراج الأصلي الخاص به. أو بدلًا من ذلك، استدعِ ClickHouseHook.bulk_insert_rows من داخل @task، انظر الخطوة 5.
settings	hook_params={"session_settings": {...}}	يدعم القوالب عبر hook_params. وتُدمج القيم فوق session_settings الخاصة بالاتصال.
do_xcom_push	do_xcom_push	الراية نفسها، لكن بصيغة مختلفة في المهام متعددة العبارات. كان البرنامج الإضافي يدفع نتيجة آخر عبارة فقط، أما مع قائمة من العبارات فيدفع المزوّد قائمة تتضمن مدخلًا واحدًا لكل عبارة. انظر نتائج العبارات المتعددة.
query_id	hook_params={"session_settings": {"query_id": "..."}}	يدعم القوالب عبر hook_params. وكما هو الحال مع البرنامج الإضافي، يُرسل المعرّف نفسه لكل عبارة ضمن مهمة متعددة العبارات. وعند تركه دون تعيين، يولّد clickhouse-connect معرّفًا فريدًا لكل عبارة؛ ويمكن لـ handler قراءة آخر معرّف من cursor.summary[-1]["query_id"].
with_column_types	handler	مرّر معالجًا يقرأ cursor.description إلى جانب الصفوف. انظر الحفاظ على أنواع الـ column.
external_tables	ClickHouseHook.get_client()	استخدم clickhouse_connect.driver.external.ExternalData مع client.query.
columnar	ClickHouseHook.get_client()	client.query(...).result_columns.
types_check	احذفه	خاص بالبروتوكول الأصلي فقط.

قبل:

from airflow_clickhouse_plugin.operators.clickhouse import ClickHouseOperator

update_income_aggregate = ClickHouseOperator(
    task_id="update_income_aggregate",
    clickhouse_conn_id="clickhouse_test",
    database="default",
    sql=(
        """
        INSERT INTO aggregate
        SELECT eventDt, sum(price * qty) AS income FROM sales
        WHERE eventDt = '{{ ds }}' GROUP BY eventDt
        """,
        """
        SELECT sum(income) FROM aggregate
        WHERE eventDt BETWEEN
            '{{ data_interval_start | ds }}' AND '{{ data_interval_end | ds }}'
        """,
    ),
    settings={"max_execution_time": 600},
)

بعد:

from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator

update_income_aggregate = SQLExecuteQueryOperator(
    task_id="update_income_aggregate",
    conn_id="clickhouse_test",
    database="default",
    sql=[
        """
        INSERT INTO aggregate
        SELECT eventDt, sum(price * qty) AS income FROM sales
        WHERE eventDt = '{{ ds }}' GROUP BY eventDt
        """,
        """
        SELECT sum(income) FROM aggregate
        WHERE eventDt BETWEEN
            '{{ data_interval_start | ds }}' AND '{{ data_interval_end | ds }}'
        """,
    ],
    hook_params={"session_settings": {"max_execution_time": 600}},
)

​

نتائج العبارات المتعددة

كان الـ plugin يدفع نتيجة آخر عبارة إلى XCom. أما SQLExecuteQueryOperator فيُعيد نتيجة واحدة لكل عبارة عندما يكون sql قائمة، لذا فإن المثال أعلاه يدفع [[], [(12345.0,)]] بينما كان الـ plugin يدفع [(12345.0,)]. اختر أحد الخيارين التاليين:

• غيّر xcom_pull في المرحلة اللاحقة ليأخذ العنصر الأخير.
• مرّر العبارات كسلسلة نصية واحدة مفصولة بـ ; واضبط split_statements=True. ومع القيمة الافتراضية return_last=True يدفع المُشغّل عندئذٍ صفوف آخر عبارة فقط، بما يطابق سلوك الـ plugin.

SQLExecuteQueryOperator(
    task_id="update_income_aggregate",
    conn_id="clickhouse_test",
    sql="""
        INSERT INTO aggregate
        SELECT eventDt, sum(price * qty) AS income FROM sales
        WHERE eventDt = '{{ ds }}' GROUP BY eventDt;
        SELECT sum(income) FROM aggregate WHERE eventDt = '{{ ds }}'
    """,
    split_statements=True,
)

إن ClickHouseOperator الذي كان يُدرج قائمة من الصفوف عبر parameters يصبح SQLInsertRowsOperator:

from airflow.providers.common.sql.operators.sql import SQLInsertRowsOperator

load_rows = SQLInsertRowsOperator(
    task_id="load_rows",
    conn_id="clickhouse_default",
    table_name="some_ch_table",
    columns=["id", "name"],
    rows=extract_task.output,
    insert_args={"executemany": True},
)

مرّر دائماً columns؛ فبدونها يبحث الـ operator عن الجدول عبر SQLAlchemy.

​

الحفاظ على column types

كانت with_column_types=True تُعيد (rows, [(name, type), ...]). يمكن تحقيق الأمر نفسه باستخدام handler، حيث يعرض مؤشر clickhouse-connect أسماء أنواع ClickHouse في cursor.description:

def fetch_with_column_types(cursor):
    return cursor.fetchall(), [(col[0], col[1]) for col in cursor.description]


SQLExecuteQueryOperator(
    task_id="typed_query",
    conn_id="clickhouse_default",
    sql="SELECT id, name FROM users LIMIT 10",
    handler=fetch_with_column_types,
)

​

الخطوة 5: من ClickHouseHook.execute إلى طرق DbApiHook

كان خطاف الـ plugin يوفّر طريقة واحدة، execute، تُحاكي clickhouse_driver.Client.execute. أما خطاف الـ provider فهو DbApiHook، ولذلك يحصل على الطرق القياسية المتوفرة في كل provider SQL آخر: run وget_records وget_first وget_pandas_df وget_df وinsert_rows وtest_connection. ولم تتغيّر وسيطتا المُنشئ clickhouse_conn_id وdatabase.

Plugin	Provider
hook.execute("SELECT ...")	hook.get_records("SELECT ...")
hook.execute("SELECT ...", params={"d": ds})	hook.get_records("SELECT ...", parameters={"d": ds})
hook.execute("SELECT count() ...")[0][0]	hook.get_first("SELECT count() ...")[0]
hook.execute("INSERT INTO t VALUES", rows)	hook.bulk_insert_rows("t", rows, column_names=[...])
hook.execute(["SET ...", "INSERT ...", "SELECT ..."])	hook.run([...], handler=fetch_all_handler, return_last=True)
hook.execute(..., settings={...})	ClickHouseHook(session_settings={...})
hook.execute(..., external_tables=..., columnar=..., query_id=...)	hook.get_client().query(...)
hook.get_conn() تُعيد clickhouse_driver.Client	hook.get_client() تُعيد clickhouse_connect Client

يُستورد fetch_all_handler وبقية الـ handlers من airflow.providers.common.sql.hooks.handlers.

أكثر أنماط استخدام الخطاف شيوعًا في شيفرة عصر الـ plugin هو الإدراج بالجملة (bulk insert). ولا يمكن تنفيذه باستدعاء run بسيط، لأن مؤشر DB-API سيحاول تضمين الصفوف داخل سلسلة SQL. استخدم بدلًا من ذلك الإدراج الأصلي (native insert):

قبل:

from airflow_clickhouse_plugin.hooks.clickhouse import ClickHouseHook


def sqlite_to_clickhouse():
    records = SqliteHook().get_records("SELECT id, name FROM some_sqlite_table")
    ClickHouseHook().execute("INSERT INTO some_ch_table VALUES", records)

بعد:

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


def sqlite_to_clickhouse():
    records = SqliteHook().get_records("SELECT id, name FROM some_sqlite_table")
    ClickHouseHook().bulk_insert_rows(
        "some_ch_table", records, column_names=["id", "name"], batch_size=100_000
    )

تتطلب bulk_insert_rows تمرير column_names. أما batch_size فهي اختيارية وتحدّ من استهلاك الذاكرة مع المدخلات الضخمة جدًا. كذلك تنتهي الدالة العامة insert_rows(table, rows, target_fields=[...], executemany=True) بعملية إدخال أصلية (native insert)، لكن فقط عند استخدام executemany=True؛ إذ يرسل السلوك الافتراضي طلب HTTP واحدًا لكل صف.

ولكل ما لا تغطيه واجهة DB-API، تُعيد get_client() عميل clickhouse-connect الخام المُهيّأ انطلاقًا من اتصال Airflow، وهو البديل لكل argument خاص بـ clickhouse-driver كان الـ plugin يوفّره:

from clickhouse_connect.driver.external import ExternalData

hook = ClickHouseHook()
with hook.get_client() as client:
    ext = ExternalData(file_name="ids", structure="id UInt64", data=b"1\n2\n3\n")
    result = client.query(
        "SELECT * FROM events WHERE id IN ids",
        external_data=ext,
        settings={"query_id": "my-traceable-id"},
    )
    columns = result.result_columns

​

الخطوة 6: من ClickHouseSensor إلى SqlSensor

هذا هو الاستبدال الوحيد الذي يتغيّر فيه دخل الدالة القابلة للاستدعاء.

	ClickHouseSensor	SqlSensor
الدالة القابلة للاستدعاء عند النجاح	is_success(result)	success(cell)
الدالة القابلة للاستدعاء عند الفشل	is_failure(result)	failure(cell)
ما تستقبله الدالة	النتيجة الكاملة لآخر عبارة، وهي list من row tuples	أول column من أول row افتراضيًا (selector=itemgetter(0))
النجاح الافتراضي	bool(result)، وتكون true عند إرجاع أي row	bool(first cell) عند إرجاع rows، وFalse عند عدم إرجاع أي منها
sql	سلسلة نصية أو قائمة من العبارات، إضافة إلى جميع وسيطات ClickHouseOperator	سلسلة نصية واحدة. استخدم hook_params لإعدادات قاعدة البيانات وإعدادات الجلسة.
معرّف الاتصال	clickhouse_conn_id، وقيمته الافتراضية clickhouse_default	conn_id، مطلوب
عدم إرجاع أي rows	is_success([])، وFalse مع القيمة الافتراضية	False، أو خطأ مع fail_on_empty=True

ولأن الـ plugin كان يمرّر الـ result set كاملًا، فإن شيفرة المستشعر القائمة تعتمد على الفهرسة داخله. أزِل هذه الفهرسة عند الترحيل:

قبل:

from airflow_clickhouse_plugin.sensors.clickhouse import ClickHouseSensor

ClickHouseSensor(
    task_id="poke_events_count",
    database="monitor",
    sql="SELECT count() FROM warnings WHERE eventDate = '{{ ds }}'",
    is_success=lambda result: result[0][0] > 10000,
)

بعد:

from airflow.providers.common.sql.sensors.sql import SqlSensor

SqlSensor(
    task_id="poke_events_count",
    conn_id="clickhouse_default",
    hook_params={"database": "monitor"},
    sql="SELECT count() FROM warnings WHERE eventDate = '{{ ds }}'",
    success=lambda cnt: cnt > 10000,
)

إذا كان الكائن القابل للاستدعاء يحتاج إلى الصف بأكمله، فمرّر selector=lambda row: row. وإذا كان يحتاج إلى جميع الصفوف، فاكتب الـ check في SQL بحيث يُعيد الاستعلام قيمة منطقية واحدة أو عدداً.

​

الخطوة 7: عائلة الأغلفة common.sql

الشيفرة التي استخدمت ClickHouseSQLExecuteQueryOperator وClickHouseSqlSensor وبقية الأغلفة المسبوقة بـ ClickHouse تحتاج إلى أقل قدر من العمل:

• غيّر الاستيراد إلى وحدة common.sql واحذف السابقة ClickHouse من اسم الصنف.
• مرّر conn_id بشكل صريح. كانت الأغلفة تتعامل مع conn_id الغائب أو None على أنه clickhouse_default، أما أصناف common.sql فليست لها قيمة افتراضية وتفشل بدونه. ويكفي default_args={"conn_id": "clickhouse_default"} لتغطية DAG بأكمله.
• لا يزال database= في المشغّلات وhook_params={"schema": ...} في المستشعر يعملان، إذ يتعامل خطاف المزوّد مع schema كاسم بديل لـ database.
• يصبح ClickHouseDbApiHook هو ClickHouseHook. ولا يزال وسيط الباني schema مقبولًا كاسم بديل، أما database فهي الصيغة الأصلية في ClickHouse ولها الأسبقية عند تمرير الاثنين معًا.
• لا يزال الاتصال بحاجة إلى تغييرات المنفذ والإضافات الواردة في الخطوة 2، فالأغلفة كانت تستخدم البروتوكول الأصلي أيضًا.

​

اختلافات السلوك التي يجب مراجعتها

حتى بعد نجاح ترجمة الشيفرة، تختلف بعض السلوكيات في وقت التشغيل.

قيمة XCom لمهام INSERT. كان الـ plugin يدفع ما يُرجعه clickhouse-driver، وهو في حالة الإدخال بصيغة VALUES مع معاملات عدد الصفوف المُدخلة. أما الـ provider فيدفع مجموعة نتائج فارغة للعبارات التي لا تُرجع صفوفًا. لذا على المهام اللاحقة التي تقرأ عدد الصفوف من XCom أن تحصل عليه بطريقة أخرى، مثل تنفيذ SELECT count() لاحق.

إعدادات الجلسة مقابل عبارات SET. تُنفّذ كلتا الحزمتين قائمة عبارات متعددة عبر اتصال واحد، وclickhouse-connect ينشئ جلسة لكل عميل افتراضيًا، لذا فإن عبارة SET في بداية القائمة ينبغي أن تنطبق على العبارات اللاحقة. ومع ذلك، يُفضّل استخدام session_settings؛ فهو صريح وقابل للقولبة، ويعمل بالطريقة ذاتها سواء حافظ الخادم أو أي proxy وسيط على الجلسة أم لا. تحقّق من السلوك في بيئتك إذا كانت مخططات DAG لديك تعتمد على SET.

الاستثناءات. أصبحت الأخطاء الآن clickhouse_connect.driver.exceptions.DatabaseError أو OperationalError أو ProgrammingError بدلًا من clickhouse_driver.errors.ServerException وNetworkError. حدّث عبارات except وشيفرة on_failure_callback ومنطق إعادة المحاولة الذي يفحص أنواع الاستثناءات.

الضغط. كان clickhouse-driver يُبقي الضغط معطّلًا ما لم يُضبط compression؛ أما clickhouse-connect فيُمكّن ضغط استجابات HTTP افتراضيًا ويتفاوض على الخوارزمية مع الخادم. اضبط "compress": false في حقل extra الخاص بالاتصال لاستعادة السلوك القديم. ولم تعد حزمة clickhouse-cityhash التي كان الـ plugin يحتاجها للضغط الأصلي مطلوبة؛ بينما تبقى lz4 مثبّتة كتابعة لـ clickhouse-connect.

تعيين الأنواع. يُرجع كلا الـ driver أنواع Python أصلية، لكنهما قاعدتا شيفرة مختلفتان. راجع المهام التي تعتمد على أنواع محددة بدقة مع DateTime64 ذات المناطق الزمنية، وDecimal، وUUID، وcolumn من نوع Nullable، وقيم Array أو Map المتداخلة، خصوصًا حين تُدفع النتيجة إلى XCom وتُستهلك في مراحل لاحقة.

تعريف الاستعلامات في system.query_log. تصل الاستعلامات الآن عبر HTTP interface، لذا تظهر بالقيمة interface = 2 بدلًا من 1، ويحمل الـ column http_user_agent في system.query_log إصدارات Airflow والـ provider إضافةً إلى حقل client_name الإضافي إن كان مضبوطًا. وأي مراقبة كانت تُرشّح بناءً على البروتوكول الأصلي أو على اسم عميل clickhouse-driver تحتاج إلى تحديث. كما أن استعلام SELECT الذي لا يُرجع صفوفًا يُنتج مدخلًا ثانيًا: إذ يُنفّذ مؤشر DB-API العبارة SELECT * FROM (...) LIMIT 0 لاسترجاع البيانات الوصفية للـ column.

المُهل الزمنية عبر HTTP. أصبح send_receive_timeout الآن هو مهلة قراءة HTTP، وأي proxy أو موازن حمل بين الـ worker وClickHouse يطبّق مهلة السكون الخاصة به على الطلب. لذا قد تحتاج العبارات التي كانت تُنفَّذ لدقائق طويلة عبر البروتوكول الأصلي إلى رفع تلك الحدود.

التعامل مع الاتصالات. يُنشئ الخطاف عميل clickhouse-connect لكل استدعاء run أو get_client، على غرار الطريقة التي كان الـ plugin يفتح بها اتصالًا أصليًا جديدًا لكل execute. ويتشارك العملاء مجمّع اتصالات HTTP على مستوى العملية بأكملها، لذا فإن استدعاء close() على عميل ناتج عن get_client() يُعدّ ممارسة جيدة لا شرطًا إلزاميًا؛ فالمجمّع يُحرَّر عند انتهاء عملية المهمة. والعميل هو context manager، لذا تبقى with hook.get_client() as client: أنظف صيغة.

​

قائمة التحقق

1. إصدار Airflow هو 2.11 أو أحدث.
2. منفذ HTTP يمكن الوصول إليه من الـ workers؛ وشهادات TLS صالحة لنقطة نهاية HTTP.
3. كل اتصال ClickHouse: النوع clickhouse، المنفذ 8123 أو 8443، والحقول الإضافية مُحوَّلة وفق الخطوة 2، ومُتحقَّق منها باستخدام airflow connections test.
4. استبدال الـ imports وفق الخطوة 3؛ وحذف بادئات ClickHouse من wrappers الخاصة بـ common.sql.
5. إعادة تسمية clickhouse_conn_id إلى conn_id في الـ operators والـ sensors، وتعيين conn_id في كل task كان يعتمد على القيمة الافتراضية للـ plugin.
6. نقل settings= إلى داخل hook_params={"session_settings": ...}.
7. استبدال hook.execute("INSERT ... VALUES", rows) بـ bulk_insert_rows؛ واستبدال ClickHouseOperator(parameters=rows) بـ SQLInsertRowsOperator.
8. تعديل الدوال القابلة للنداء في الـ sensors من result[0][0] إلى قيمة الـ cell المجردة.
9. إعادة كتابة استخدامات with_column_types وexternal_tables وcolumnar وquery_id وtypes_check باستخدام handler أو get_client().
10. مراجعة الجهات المستهلكة اللاحقة لـ XComs الناتجة عن الـ tasks متعددة العبارات وtasks الإدخال (INSERT).
11. تحديث الشيفرة التي تعترض استثناءات clickhouse_driver.
12. إلغاء تثبيت airflow-clickhouse-plugin وclickhouse-driver.

​

استخدام مساعد برمجي يعمل بالذكاء الاصطناعي

صُمِّم الـ mapping أعلاه ليكون آليًا بحتًا بشكل متعمّد، بحيث يستطيع المساعد البرمجي تطبيقه على repository خاص بـ DAG. وفيما يلي prompt أثبت فعاليته:

Migrate this repository from airflow-clickhouse-plugin to
apache-airflow-providers-clickhousedb following
https://clickhouse.com/docs/integrations/airflow/migrating-from-airflow-clickhouse-plugin

- Replace every airflow_clickhouse_plugin import per the class mapping table.
- Rename clickhouse_conn_id to conn_id on operators and sensors, not on hooks; add
  conn_id="clickhouse_default" wherever a task relied on the plugin's default.
- Move settings= into hook_params={"session_settings": ...}.
- Replace hook.execute(...) with get_records, get_first, run or bulk_insert_rows
  according to the hook table; never pass a list of rows as parameters. Replace
  ClickHouseOperator(parameters=<rows>) with SQLInsertRowsOperator.
- Rewrite sensor callables to receive the first cell instead of the full result.
- Flag, but do not silently rewrite, any use of with_column_types, external_tables,
  columnar, query_id or types_check, and any XCom consumer of an INSERT task.
- List every Airflow connection that must change port and extras; do not edit
  connections yourself.
Show the diff and a summary of items flagged for human review.

راجع الفروقات؛ فتغييرات الاتصال ودلالات المستشعر ومستهلكو XCom هي المواضع التي تخطئ فيها عمليات إعادة الكتابة الآلية.

​

محتوى ذو صلة

• ربط Apache Airflow بـ ClickHouse
• عميل clickhouse-connect لبايثون
• المستندات المرجعية لـ apache-airflow-providers-clickhousedb
• airflow-clickhouse-plugin على GitHub