> ## Documentation Index > Fetch the complete documentation index at: https://private-7c7dfe99-vortex-format.mintlify.site/llms.txt > Use this file to discover all available pages before exploring further. > نقل DAGs الحالية في Airflow من حزمة المجتمع airflow-clickhouse-plugin إلى المزود الرسمي apache-airflow-providers-clickhousedb # الترحيل من airflow-clickhouse-plugin إلى مزود ClickHouse قبل ظهور [مزود ClickHouse الرسمي](/ar/integrations/connectors/data-ingestion/etl-tools/airflow-and-clickhouse)، كان معظم مستخدمي Airflow يتصلون بـ ClickHouse عبر حزمة المجتمع [airflow-clickhouse-plugin](https://github.com/bryzgaloff/airflow-clickhouse-plugin). يستعرض هذا الدليل خطوات ترحيل عملية نشر قائمة إلى `apache-airflow-providers-clickhousedb`. ولمعرفة كيفية عمل المزود نفسه، راجع [ربط Apache Airflow بـ ClickHouse](/ar/integrations/connectors/data-ingestion/etl-tools/airflow-and-clickhouse).

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

المزوّدات (Providers) هي الطريقة التي يتكامل بها Airflow مع الأنظمة الخارجية. ويتم إصدارها واختبارها وتوثيقها مع بقية منظومة Airflow، وهذا المزوّد يتولى صيانته مجتمع Airflow بالتعاون مع ClickHouse team. وهو مبني على [clickhouse-connect](/ar/integrations/language-clients/python/index)، وهو عميل بايثون الذي تطوّره 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` | مجموعة ثابتة من المفاتيح، انظر [خيارات الاتصال الإضافية](/ar/integrations/connectors/data-ingestion/etl-tools/airflow-and-clickhouse#extra-connection-options) | | المُشغِّلات والمستشعرات | `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 أولاً إذا كنت تستخدم إصداراً أقدم. ```bash theme={null} pip install apache-airflow-providers-clickhousedb ``` توجد الحزمتان في مساحتي أسماء مختلفتين في بايثون، لذا يمكن تثبيتهما جنبًا إلى جنب أثناء ترحيل كل DAG على حدة. أزل الـ plugin بعد أن لا يعود أي شيء يستورده: ```bash theme={null} 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](/ar/products/cloud/getting-started/intro) فلا يعرض 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/ apache-airflow-providers-clickhousedb/` ويُلحِق قيمتك بوصفها 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`. | قبل: ```json theme={null} { "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 } } ``` بعد: ```json theme={null} { "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: ```bash theme={null} 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](#step-7-the-commonsql-wrapper-family)).

الخطوة 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](#step-5-clickhousehookexecute-to-dbapihook-methods). | | `settings` | `hook_params={"session_settings": {...}}` | يدعم القوالب عبر `hook_params`. وتُدمج القيم فوق `session_settings` الخاصة بالاتصال. | | `do_xcom_push` | `do_xcom_push` | الراية نفسها، لكن بصيغة مختلفة في المهام متعددة العبارات. كان البرنامج الإضافي يدفع نتيجة آخر عبارة فقط، أما مع قائمة من العبارات فيدفع المزوّد قائمة تتضمن مدخلًا واحدًا لكل عبارة. انظر [نتائج العبارات المتعددة](#multi-statement-results). | | `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](#keeping-column-types). | | `external_tables` | `ClickHouseHook.get_client()` | استخدم `clickhouse_connect.driver.external.ExternalData` مع `client.query`. | | `columnar` | `ClickHouseHook.get_client()` | `client.query(...).result_columns`. | | `types_check` | احذفه | خاص بالبروتوكول الأصلي فقط. | قبل: ```python theme={null} 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}, ) ``` بعد: ```python theme={null} 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. ```python theme={null} 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`: ```python theme={null} 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`: ```python theme={null} 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): قبل: ```python theme={null} 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) ``` بعد: ```python theme={null} 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 يوفّره: ```python theme={null} 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 كاملًا، فإن شيفرة المستشعر القائمة تعتمد على الفهرسة داخله. أزِل هذه الفهرسة عند الترحيل: قبل: ```python theme={null} 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, ) ``` بعد: ```python theme={null} 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`](/ar/reference/system-tables/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 أثبت فعاليته: ```text theme={null} 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=) 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](/ar/integrations/connectors/data-ingestion/etl-tools/airflow-and-clickhouse) * [عميل `clickhouse-connect` لبايثون](/ar/integrations/language-clients/python/index) * [المستندات المرجعية لـ `apache-airflow-providers-clickhousedb`](https://airflow.apache.org/docs/apache-airflow-providers-clickhousedb/) * [airflow-clickhouse-plugin على GitHub](https://github.com/bryzgaloff/airflow-clickhouse-plugin)