الاستخدام المتقدم

الاستخدام المتقدم مع ClickHouse Connect

​

واجهة برمجة التطبيقات الخام

لحالات الاستخدام التي لا تتطلب تحويلًا بين بيانات ClickHouse وأنواع البيانات والبُنى الأصلية أو الخاصة بجهات خارجية، يوفّر عميل ClickHouse Connect طرقًا لاستخدام اتصال ClickHouse مباشرةً.

​

الطريقة raw_query في Client

تتيح الطريقة Client.raw_query استخدام واجهة استعلام HTTP الخاصة بـ ClickHouse مباشرةً عبر اتصال العميل. وتكون القيمة المُعادة كائن bytes غير مُعالَج. كما توفّر طبقة تغليف ملائمة مع ربط المعلمات، ومعالجة الأخطاء، وإعادات المحاولة، وإدارة الإعدادات من خلال واجهة مبسطة:

المعلمة	النوع	الافتراضي	الوصف
query	str	مطلوب	أي استعلام ClickHouse صالح.
parameters	dict or sequence	None	راجع وسيطة المعلمات.
settings	dict	None	راجع وسيطة الإعدادات.
fmt	str	None	تنسيق الإخراج في ClickHouse. يستخدم ClickHouse تنسيق TSV عند عدم تحديد أي تنسيق.
use_database	bool	True	أدرج قاعدة البيانات المُعدّة على العميل.
external_data	ExternalData	None	ملف خارجي أو بيانات ثنائية. راجع البيانات الخارجية.
transport_settings	dict	None	ترويسات HTTP المُضافة إلى هذا الطلب.

تقع على عاتق المستدعي مسؤولية التعامل مع كائن bytes الناتج. لاحظ أن Client.query_arrow ليس سوى طبقة تغليف خفيفة حول هذه الطريقة باستخدام تنسيق الإخراج Arrow في ClickHouse.

​

طريقة raw_stream في Client

للطريقة المتزامنة Client.raw_stream واجهة برمجة تطبيقات مماثلة لـ raw_query، لكنها تُرجع تدفق io.IOBase من مقاطع بايت. أغلِق التدفق عند انتهاء المعالجة. ويُنتظر AsyncClient.raw_stream ويُرجع StreamContext غير متزامن لاستخدامه مع async with وasync for.

​

طريقة raw_insert في Client

تتيح الطريقة Client.raw_insert إجراء عمليات إدراج مباشرة لكائنات bytes أو مولدات كائنات bytes باستخدام اتصال العميل. ونظرًا لأنها لا تجري أي معالجة لحمولة الإدراج، فهي عالية الكفاءة جدًا. كما توفّر الطريقة خيارات لتحديد الإعدادات وتنسيق الإدراج:

Parameter	Type	Default	Description
table	str	Required	الجدول الهدف البسيط أو المؤهل باسم قاعدة البيانات.
column_names	Sequence[str]	None	أسماء الأعمدة الخاصة بكتلة الإدراج. وهي مطلوبة عندما لا يتضمن fmt الأسماء.
insert_block	str, bytes, generator, or BinaryIO	Required	البيانات المطلوب إدراجها. تُرمَّز السلاسل النصية باستخدام ترميز العميل.
settings	dict	None	راجع وسيطة الإعدادات.
fmt	str	None	تنسيق الإدخال في ClickHouse لحمولة insert_block. يُستخدم Native عندما لا يتم تحديد تنسيق.
compression	str	None	الضغط المُطبَّق مسبقًا على insert_block، مثل "gzip" أو "lz4" أو "zstd".
transport_settings	dict	None	ترويسات HTTP المضافة إلى هذا الطلب.

تقع على عاتق المستدعي مسؤولية التأكد من أن insert_block بالتنسيق المحدد ويستخدم طريقة الضغط المحددة. ويستخدم ClickHouse Connect عمليات الإدراج الخام هذه لرفع الملفات وجداول PyArrow، مع تفويض التحليل إلى خادم ClickHouse.

​

حفظ نتائج الاستعلامات كملفات

يمكنك بث الملفات مباشرةً من ClickHouse إلى نظام الملفات المحلي باستخدام الطريقة raw_stream. على سبيل المثال، إذا كنت ترغب في حفظ نتائج استعلام في ملف CSV، فيمكنك استخدام مقتطف الشيفرة التالي:

import clickhouse_connect

if __name__ == "__main__":
    client = clickhouse_connect.get_client()
    query = (
        "SELECT number, toString(number) AS number_as_str "
        "FROM system.numbers LIMIT 5"
    )
    stream = client.raw_stream(query=query, fmt="CSVWithNames")
    try:
        with open("output.csv", "wb") as file:
            for chunk in stream:
                file.write(chunk)
    finally:
        stream.close()
        client.close()

ينتج عن الشيفرة أعلاه ملف output.csv بالمحتوى التالي:

"number","number_as_str"
0,"0"
1,"1"
2,"2"
3,"3"
4,"4"

وبالمثل، يمكنك حفظ البيانات بتنسيق TabSeparated وبتنسيقات أخرى. راجع تنسيقات بيانات الإدخال والإخراج للاطلاع على نظرة عامة على جميع خيارات التنسيق المتاحة.

​

حالات الاستخدام متعددة الخيوط، ومتعددة العمليات، وغير المتزامنة/القائمة على حلقة الأحداث

يعمل ClickHouse Connect بكفاءة في التطبيقات متعددة الخيوط، ومتعددة العمليات، والتطبيقات غير المتزامنة/القائمة على حلقة الأحداث. تتم جميع عمليات معالجة الاستعلامات والإدراج ضمن خيط واحد، لذا تكون العمليات آمنة على مستوى الخيوط بشكل عام. (قد تُضاف مستقبلًا إمكانية المعالجة المتوازية لبعض العمليات على مستوى منخفض لتجاوز أثر الأداء المترتب على الاعتماد على خيط واحد، ولكن حتى في هذه الحالة ستظل السلامة على مستوى الخيوط محفوظة.)

ولأن كل استعلام أو عملية إدراج يتم تنفيذها يحتفظ كلٌّ منها بحالته داخل الكائن QueryContext أو InsertContext الخاص به، على التوالي، فإن هذه الكائنات المساعدة ليست آمنة على مستوى الخيوط، ولا ينبغي مشاركتها بين عدة تدفقات معالجة. راجع أيضًا المناقشة الإضافية حول كائنات السياق في قسمي QueryContexts وInsertContexts.

إضافةً إلى ذلك، في التطبيق الذي توجد فيه استعلامات و/أو عمليات إدراج، اثنتان أو أكثر، “قيد التنفيذ” في الوقت نفسه، هناك اعتباران إضافيان ينبغي أخذهما في الحسبان. الأول هو “الجلسة” في ClickHouse المرتبطة بالاستعلام/الإدراج، والثاني هو تجمع اتصالات HTTP الذي تستخدمه مثيلات ClickHouse Connect Client.

​

AsyncClient

يوفّر ClickHouse Connect عميلًا أصليًا يعتمد على aiohttp لتطبيقات asyncio. ثبّت التبعية الاختيارية قبل استخدامه:

pip install "clickhouse-connect[async]"

استخدم await مع get_async_client لإنشاء العميل وتهيئته. طرائق الإدخال/الإخراج مثل query وcommand وinsert هي روتينات تعاونية:

import asyncio

import clickhouse_connect


async def main():
    async with await clickhouse_connect.get_async_client() as client:
        result = await client.query(
            "SELECT name FROM system.databases ORDER BY name LIMIT 1"
        )
        print(result.result_rows)


asyncio.run(main())

يتبع العميل غير المتزامن نفس واجهة query وinsert وraw وArrow وstreaming التي يتبعها العميل المتزامن. ويستخدم aiohttp لعمليات I/O على الشبكة. وقد يُشغَّل تحليل تنسيق Native كثيف الاعتماد على CPU داخل executor حتى لا يحجب حلقة الأحداث.

يمتلك العميل غير المتزامن جلسة aiohttp أُنشئت في حلقة أحداث بعينها. قبل نقل العميل إلى حلقة أحداث أخرى، أغلقه في الحلقة المالكة له، ثم استدعِ await client._initialize() في الحلقة الجديدة قبل إرسال أي طلبات. أما إذا كانت الحلقة المالكة قد أُغلقت بالفعل، فاستدعِ await client.close() ثم await client._initialize() في الحلقة الحالية. وقد يظل aiohttp يُبلغ عن ناقل (transport) غير مُغلق إذا لم تبدأ عملية التنظيف إلا بعد إغلاق الحلقة المالكة، لذا احرص على إغلاق العميل قبل نقله كلما أمكن ذلك.

تُنتظر طرق البث غير المتزامنة قبل الدخول إلى السياق المُعاد:

async with await client.query_rows_stream(
    "SELECT number FROM numbers(100000)"
) as stream:
    async for row in stream:
        process(row)

على عكس المصنع المتزامن، يعطّل get_async_client مُعرّفات الجلسات التلقائية افتراضيًا لكي تتمكن coroutines المتزامنة من مشاركة عميل واحد. مرّر session_id محددًا صراحةً أو استخدم autogenerate_session_id=True فقط عند الحاجة إلى حالة الجلسة، مع تجنّب الاستعلامات المتزامنة ضمن تلك الجلسة.

​

إدارة معرّفات الجلسات في ClickHouse

يُنفَّذ كل استعلام في ClickHouse ضمن سياق “جلسة” في ClickHouse. وتُستخدم الجلسات حاليًا لغرضين:

• ربط إعدادات ClickHouse محددة بعدة استعلامات (راجع إعدادات المستخدم). ويُستخدم الأمر SET في ClickHouse لتغيير الإعدادات ضمن نطاق جلسة المستخدم.
• تتبّع الجداول المؤقتة.

بشكل افتراضي، يستخدم Client المتزامن معرّف جلسة يتم إنشاؤه تلقائيًا. ولا تستمر عبارات SET والجداول المؤقتة عبر الطلبات الصادرة من ذلك العميل إلا عندما تصل تلك الطلبات إلى عملية خادم ClickHouse نفسها. ولا تُنشئ الدالة المصنعية غير المتزامنة معرّف جلسة بشكل افتراضي. وتقتصر حالة الجلسات المسمّاة وفحوص التداخل ضمن الجلسة نفسها على العملية المحلية، ويرفع العميل الخطأ ProgrammingError عندما يكتشف تداخلًا محليًا قبل إرسال الطلب. وفي ClickHouse Cloud أو غيرها من عمليات النشر التي تستخدم موازنة الحمل، لا تعتمد على session_id ثابت كحالة موزَّعة أو كقفل mutex موزَّع. وإذا كان التداخل يمثّل مشكلة، فرتّب الطلبات تسلسليًا قبل إرسالها إلى ClickHouse. استخدم أحد الأنماط التالية:

1. أنشئ مثيل Client منفصلًا لكل خيط تنفيذ/process/event handler يحتاج إلى عزل للجلسة. يحافظ ذلك على حالة الجلسة الخاصة بكل عميل (الجداول المؤقتة وقيم SET).
2. استخدم session_id فريدًا لكل استعلام عبر الوسيط settings عند استدعاء query أو command أو insert، إذا لم تكن بحاجة إلى حالة جلسة مشتركة.
3. عطّل الجلسات على عميل مشترك عبر تعيين autogenerate_session_id=False قبل إنشاء العميل (أو مرّره مباشرةً إلى get_client).

import clickhouse_connect
from clickhouse_connect import common

common.set_setting("autogenerate_session_id", False)
client = clickhouse_connect.get_client(
    host="somehost.com",
    username="dbuser",
    password="password",
)

بدلًا من ذلك، مرّر autogenerate_session_id=False مباشرةً إلى get_client(...).

في هذه الحالة، لا يرسل ClickHouse Connect قيمة session_id؛ ولا يعتبر الخادم الطلبات المنفصلة جزءًا من الجلسة نفسها. ولن تستمر الجداول المؤقتة وإعدادات مستوى الجلسة عبر الطلبات.

​

تخصيص تجمع اتصالات HTTP

يستخدم ClickHouse Connect تجمعات اتصالات HTTP urllib3 لإدارة اتصال HTTP الأساسي مع الخادم. افتراضيًا، تشترك جميع مثيلات العميل المتزامنة داخل العملية الواحدة في تجمع اتصالات HTTP نفسه، وهو كافٍ لمعظم حالات الاستخدام. ويحصل كل عامل في المعالجة المتعددة على تجمع افتراضي خاص به ومحلي لعمليته، ويعيد استخدامه لجميع العملاء الذين يُنشَؤون داخل ذلك العامل. أما العميل الذي يُنشأ قبل تنفيذ fork فيحتفظ بتجمع العملية الأم، ولا ينبغي استخدامه في العملية الفرعية. ويحافظ التجمع الافتراضي على ما يصل إلى 8 اتصالات HTTP Keep Alive مع كل خادم ClickHouse يستخدمه التطبيق.

تُفعّل خيارات المقبس الافتراضية آلية TCP keepalive وTCP_NODELAY، بينما يتولى نظام التشغيل إدارة أحجام المخازن المؤقتة للإرسال والاستقبال الخاصة بالمقبس.

بالنسبة إلى التطبيقات الكبيرة متعددة الخيوط، قد يكون من الأنسب استخدام تجمعات اتصالات HTTP منفصلة. ويمكن توفير تجمعات اتصالات HTTP مخصّصة عبر وسيط الكلمة المفتاحية pool_mgr للدالة الرئيسية clickhouse_connect.get_client:

import clickhouse_connect
from clickhouse_connect.driver import httputil

big_pool_mgr = httputil.get_pool_manager(maxsize=16, num_pools=12)

client1 = clickhouse_connect.get_client(pool_mgr=big_pool_mgr)
client2 = clickhouse_connect.get_client(pool_mgr=big_pool_mgr)

يمكن لعدة عملاء مشاركة مدير تجمع واحد، أو يمكن لكل عميل استخدام مدير منفصل. لمزيد من التفاصيل، راجع وثائق urllib3 الخاصة بـ PoolManager.

لتعيين خيارات المقبس، مرّر socket_options إلى httputil.get_pool_manager أو httputil.get_pool_manager_options. يؤدي ذلك إلى استبدال القائمة الافتراضية بأكملها، بما فيها خيارات keepalive وTCP_NODELAY. مرّر [] أو None إذا كنت لا تريد تطبيق أي خيارات مقبس صريحة.

يمتلك العميل غير المتزامن تجمع aiohttp بدلًا من استخدام urllib3. قم بتكوينه من خلال connector_limit وconnector_limit_per_host وkeepalive_timeout في get_async_client. يؤدي استدعاء await async_client.close_connections() إلى تبديل التجمع دوريًا دون مقاطعة الطلبات قيد التنفيذ.

في الاستعلامات وعمليات الإدراج غير المتزامنة، لا توجد مهلة زمنية لانتظار توفّر خانة شاغرة في التجمع. لذا اقرأ الاستجابات المتدفقة بالكامل أو أغلقها لتحرير خانات التجمع التي تشغلها. تبدأ connect_timeout بعد توفّر خانة، وتشمل تحليل أسماء DNS وإعداد اتصال TCP وTLS والتفاوض مع الوكيل. أما send_receive_timeout فتحدّ من مدة عمليات القراءة من المقبس. ولتعيين موعد نهائي للعملية بأكملها، بما في ذلك انتظار التجمع، استخدم asyncio.wait_for، على سبيل المثال: await asyncio.wait_for(client.query("SELECT 13"), timeout=30).