محرك الجدول NATS

يتيح هذا المحرك دمج ClickHouse مع NATS لنشر subject الرسائل أو الاشتراك فيها، ومعالجة الرسائل الجديدة عند توفرها.

يتيح هذا المحرك دمج ClickHouse مع NATS.

يتيح لك NATS ما يلي:

• نشر subject الرسائل أو الاشتراك فيها.
• معالجة الرسائل الجديدة عند توفرها.

​

إنشاء جدول

CREATE TABLE [IF NOT EXISTS] [db.]table_name [ON CLUSTER cluster]
(
    name1 [type1] [DEFAULT|MATERIALIZED|ALIAS expr1],
    name2 [type2] [DEFAULT|MATERIALIZED|ALIAS expr2],
    ...
) ENGINE = NATS SETTINGS
    nats_url = 'host:port',
    nats_subjects = 'subject1,subject2,...',
    nats_format = 'data_format'[,]
    [nats_schema = '',]
    [nats_num_consumers = N,]
    [nats_queue_group = 'group_name',]
    [nats_secure = false,]
    [nats_max_reconnect = N,]
    [nats_reconnect_wait = N,]
    [nats_server_list = 'host1:port1,host2:port2,...',]
    [nats_skip_broken_messages = N,]
    [nats_max_block_size = N,]
    [nats_flush_interval_ms = N,]
    [nats_username = 'user',]
    [nats_password = 'password',]
    [nats_token = 'clickhouse',]
    [nats_credentials = '-----BEGIN NATS USER JWT----- ...',]
    [nats_startup_connect_tries = 5,]
    [nats_max_rows_per_message = 1,]
    [nats_commit_on_select = false,]
    [nats_handle_error_mode = 'default']

المعلمات المطلوبة:

• nats_url – host:port (على سبيل المثال، localhost:4222)..
• nats_subjects – قائمة بالـ subject التي يشترك فيها جدول NATS أو ينشر إليها. يدعم subject أحرف البدل مثل foo.*.bar أو baz.>
• nats_format – تنسيق الرسالة. يستخدم الصياغة نفسها المستخدمة في دالة SQL FORMAT، مثل JSONEachRow. لمزيد من المعلومات، راجع قسم التنسيقات.

المعلمات الاختيارية:

• nats_schema – معلمة يجب استخدامها إذا كان التنسيق يتطلب تعريف schema. على سبيل المثال، يتطلب Cap’n Proto مسار ملف schema واسم الكائن الجذر schema.capnp:Message.
• nats_stream – اسم stream موجود في NATS JetStream.
• nats_consumer_name – اسم durable pull مستهلك موجود في NATS JetStream.
• nats_num_consumers – عدد الـ مستهلكين لكل جدول. القيمة الافتراضية: 1. حدِّد عددًا أكبر من الـ مستهلكين إذا كان معدل النقل لمستهلك واحد غير كافٍ في NATS core فقط.
• nats_queue_group – اسم مجموعة queue لمشتركي NATS. القيمة الافتراضية هي اسم الجدول.
• nats_max_reconnect – مهمل وليس له أي تأثير، إذ تتم إعادة الاتصال بشكل دائم مع مهلة nats_reconnect_wait.
• nats_reconnect_wait – مقدار الوقت بالمللي ثانية للانتظار بين كل محاولة إعادة اتصال. القيمة الافتراضية: 2000.
• nats_server_list - قائمة الخوادم الخاصة بالاتصال. يمكن تحديدها للاتصال بعنقود NATS.
• nats_skip_broken_messages - مدى تحمّل محلل رسائل NATS للرسائل غير المتوافقة مع schema لكل block. القيمة الافتراضية: 0. إذا كان nats_skip_broken_messages = N فسيتخطى المحرك عدد N من رسائل NATS التي يتعذر تحليلها (الرسالة الواحدة تعادل صفًا واحدًا من البيانات).
• nats_max_block_size - عدد الـ row التي تجمعها عملية أو عمليات poll لتفريغ البيانات من NATS. القيمة الافتراضية: max_insert_block_size.
• nats_flush_interval_ms - مهلة تفريغ البيانات المقروءة من NATS. القيمة الافتراضية: stream_flush_interval_ms.
• nats_wait_for_flush_interval - إذا كانت القيمة true، فستبقى دورة البث في الخلفية مفتوحة طوال فترة التفريغ (nats_flush_interval_ms، أو stream_flush_interval_ms بخلاف ذلك) بدلًا من الانتهاء بمجرد تفريغ queue المستهلك، مما يتيح تراكم المزيد من الرسائل في block واحد مقابل زمن استيعاب إضافي يصل إلى فترة تفريغ واحدة. القيمة الافتراضية: false (سلوك تفريغ فوري منخفض الكمون).
• nats_username - اسم مستخدم NATS. عند تخزينه في مجموعة مسماة معرّفة في ملف إعدادات الخادم، لا يمكن لـ query تجاوز nats_url أو nats_server_list الخاصين بالمجموعة.
• nats_password - كلمة مرور NATS. عند تخزينها في مجموعة مسماة معرّفة في ملف إعدادات الخادم، لا يمكن لـ query تجاوز nats_url أو nats_server_list الخاصين بالمجموعة.
• nats_token - رمز مصادقة NATS. عند تخزينه في مجموعة مسماة معرّفة في ملف إعدادات الخادم، لا يمكن لـ query تجاوز nats_url أو nats_server_list الخاصين بالمجموعة.
• nats_credential_file - مسار ملف بيانات اعتماد NATS. لا يُقبل إلا من مجموعة مسماة معرّفة في ملف إعدادات الخادم لا يتم تجاوز nats_url وnats_server_list الخاصين بها بواسطة query، لأن الخادم يفتح المسار بصلاحياته الخاصة. في query، مرّر محتويات الملف في nats_credentials بدلًا من ذلك.
• nats_credentials - محتوى بيانات اعتماد NATS (نفس الحمولة الموجودة في ملف .creds الذي يتضمن JWT المستخدم والبذرة). نظرًا إلى أنه الصيغة الوحيدة التي يمكن لـ query استخدامها، فإنه يستبدل nats_credential_file الموروث من مجموعة مسماة بدلًا من التعارض معه، ما لم يقفل المشغّل ذلك المسار باستخدام <nats_credential_file overridable="false">. لا يمكن تعيين سلسلة فارغة له لإزالة بيانات الاعتماد التي تحملها مجموعة مسماة.
• nats_ca_file - مسار ملف يحتوي على شهادات CA الموثوقة المستخدمة للتحقق من شهادة خادم NATS. يتطلب nats_secure. ومثل nats_credential_file، لا يُقبل إلا من مجموعة مسماة معرّفة في ملف إعدادات الخادم لا يتم تجاوز nats_url وnats_server_list الخاصين بها بواسطة query، لأن الخادم يفتح المسار بصلاحياته الخاصة.
• nats_client_cert_file - مسار شهادة العميل المقدمة إلى خادم NATS. يتطلب nats_secure وnats_client_key_file. يُقبل من المصادر نفسها التي يُقبل منها nats_ca_file.
• nats_client_key_file - مسار المفتاح الخاص بـ nats_client_cert_file. يُقبل من المصادر نفسها التي يُقبل منها nats_ca_file.
• nats_startup_connect_tries - عدد محاولات الاتصال عند بدء التشغيل. القيمة الافتراضية: 5.
• nats_max_rows_per_message — الحد الأقصى لعدد rows المكتوبة في رسالة NATS واحدة للتنسيقات المعتمدة على الصفوف. (القيمة الافتراضية: 1).
• nats_commit_on_select - ثبّت الرسائل عند تنفيذ query. ينطبق على JetStream فقط؛ إذ لا يحتوي NATS core على إقرارات. القيمة الافتراضية: 0.
• nats_handle_error_mode — كيفية التعامل مع الأخطاء في محرك NATS. القيم الممكنة: default (سيتم طرح الاستثناء إذا فشل تحليل رسالة)، stream (ستُحفَظ رسالة الاستثناء والرسالة الخام في أعمدة افتراضية _error و _raw_message).

اتصال SSL:

للاتصال الآمن، استخدم nats_secure = 1. يتحكم متغير البيئة CLICKHOUSE_NATS_TLS_SECURE في التحقق من الشهادة؛ إذا كانت الشهادة منتهية الصلاحية أو موقعة ذاتيًا أو مفقودة أو غير صالحة لسبب آخر، فعطّل التحقق بتعيين CLICKHOUSE_NATS_TLS_SECURE=0.

يتم التحقق من شهادة خادم موقعة بواسطة CA خاصة عبر توجيه nats_ca_file إلى شهادة CA، وهو أفضل من إيقاف التحقق. عندما يتطلب الخادم شهادات العميل، زوّده بها باستخدام nats_client_cert_file وnats_client_key_file. الخيارات الثلاثة كلها إعدادات للمشغّل: إذ تأتي من مجموعة مُسمّاة محددة في ملف إعدادات الخادم. يُقرأ كل ملف عند اتصال الجدول، لذلك يؤدي الملف غير القابل للقراءة أو المشوّه إلى فشل الاستعلام بدلًا من المصافحة.

الكتابة إلى جدول NATS:

إذا كان الجدول يقرأ فقط من subject واحد، فسيُنشَر أي إدراج إلى نفس subject. ومع ذلك، إذا كان الجدول يقرأ من عدة subject، فسنحتاج إلى تحديد الـ subject الذي نرغب في النشر إليه. لذلك، عند الإدراج في جدول يقرأ من عدة subject، يلزم ضبط stream_like_engine_insert_queue. يمكنك اختيار أحد الـ subject التي يقرأ منها الجدول ونشر بياناتك هناك. على سبيل المثال:

CREATE TABLE queue (
    key UInt64,
    value UInt64
  ) ENGINE = NATS
    SETTINGS nats_url = 'localhost:4444',
             nats_subjects = 'subject1,subject2',
             nats_format = 'JSONEachRow';

INSERT INTO queue
SETTINGS stream_like_engine_insert_queue = 'subject2'
VALUES (1, 1);

يمكن أيضًا إضافة إعدادات التنسيق إلى جانب الإعدادات المتعلقة بـ NATS.

مثال:

CREATE TABLE queue (
    key UInt64,
    value UInt64,
    date DateTime
  ) ENGINE = NATS
    SETTINGS nats_url = 'localhost:4444',
             nats_subjects = 'subject1',
             nats_format = 'JSONEachRow',
             date_time_input_format = 'best_effort';

يمكن إضافة إعدادات خادم NATS باستخدام ملف إعدادات ClickHouse. وبشكل أكثر تحديدًا، يمكنك إضافة كلمة مرور NATS لمحرك NATS:

<nats>
    <user>click</user>
    <password>house</password>
    <token>clickhouse</token>
</nats>

​

الوصف

لا تكون SELECT مفيدة كثيرًا لقراءة الرسائل (إلا لأغراض استكشاف الأخطاء وإصلاحها)، لأن كل رسالة لا يمكن قراءتها إلا مرة واحدة. والأكثر عملية هو إنشاء مسارات في الوقت الفعلي باستخدام العروض المادية. للقيام بذلك:

1. استخدم المحرك لإنشاء مستهلك NATS واعتبره stream بيانات.
2. أنشئ جدولًا بالبنية المطلوبة.
3. أنشئ عرضًا ماديًا يحوّل البيانات من المحرك ويضعها في جدول أُنشئ مسبقًا.

عندما يرتبط MATERIALIZED VIEW بالمحرك، يبدأ في جمع البيانات في الخلفية. يتيح لك ذلك الاستمرار في تلقي الرسائل من NATS وتحويلها إلى التنسيق المطلوب باستخدام SELECT. يمكن أن يحتوي جدول NATS واحد على أي عدد تريده من العروض المادية؛ فهي لا تقرأ البيانات من الجدول مباشرةً، بل تستقبل السجلات الجديدة (على شكل كتل)، وبهذه الطريقة يمكنك الكتابة إلى عدة جداول بمستويات مختلفة من التفصيل (مع التجميع وبدونه).

مثال:

CREATE TABLE queue (
    key UInt64,
    value UInt64
  ) ENGINE = NATS
    SETTINGS nats_url = 'localhost:4444',
             nats_subjects = 'subject1',
             nats_format = 'JSONEachRow',
             date_time_input_format = 'best_effort';

CREATE TABLE daily (key UInt64, value UInt64)
    ENGINE = MergeTree() ORDER BY key;

CREATE MATERIALIZED VIEW consumer TO daily
    AS SELECT key, value FROM queue;

SELECT key, value FROM daily ORDER BY key;

لإيقاف استلام بيانات الـ stream أو لتغيير منطق التحويل، افصل العرض المادي:

DETACH TABLE consumer;
ATTACH TABLE consumer;

إذا كنت تريد تغيير الجدول الهدف باستخدام ALTER، فنوصي بتعطيل العرض المادي لتجنّب حدوث اختلافات بين الجدول الهدف والبيانات الواردة من العرض.

​

الأعمدة الافتراضية

• _subject - الـsubject لرسالة NATS. نوع البيانات: String.

أعمدة افتراضية إضافية عندما تكون قيمة nats_handle_error_mode='stream':

• _raw_message - الرسالة الخام التي تعذّر تحليلها بنجاح. نوع البيانات: Nullable(String).
• _error - رسالة الاستثناء التي حدثت أثناء فشل التحليل. نوع البيانات: Nullable(String).

ملاحظة: لا تُملأ الأعمدة الافتراضية _raw_message و _error إلا عند حدوث استثناء أثناء التحليل، وتكون دائمًا NULL عندما يُحلَّلَت الرسالة بنجاح.

​

دعم تنسيقات البيانات

يدعم محرك NATS جميع التنسيقات التي يدعمها ClickHouse. ويعتمد عدد الصفوف في رسالة NATS الواحدة على ما إذا كان التنسيق يعتمد على الصفوف أم على الكتل:

• بالنسبة إلى التنسيقات المعتمدة على الصفوف، يمكن التحكم في عدد الصفوف في رسالة NATS الواحدة عبر ضبط nats_max_rows_per_message.
• أما بالنسبة إلى التنسيقات المعتمدة على الكتل، فلا يمكن تقسيم الكتلة إلى أجزاء أصغر، لكن يمكن التحكم في عدد الصفوف في الكتلة الواحدة من خلال الإعداد العام max_block_size.

​

استخدام JetStream

قبل استخدام محرك NATS مع NATS JetStream، يجب إنشاء stream في NATS وdurable pull مستهلك. ويمكنك لهذا الغرض استخدام أداة nats، على سبيل المثال، من حزمة NATS CLI:

إنشاء stream

$ nats stream add
? Stream Name stream_name
? Subjects stream_subject
? Storage file
? Replication 1
? Retention Policy Limits
? Discard Policy Old
? Stream Messages Limit -1
? Per Subject Messages Limit -1
? Total Stream Size -1
? Message TTL -1
? Max Message Size -1
? Duplicate tracking time window 2m0s
? Allow message Roll-ups No
? Allow message deletion Yes
? Allow purging subjects or the entire stream Yes
Stream stream_name was created

Information for Stream stream_name created 2025-10-03 14:12:51

                Subjects: stream_subject
                Replicas: 1
                 Storage: File

Options:

               Retention: Limits
         Acknowledgments: true
          Discard Policy: Old
        Duplicate Window: 2m0s
              Direct Get: true
       Allows Msg Delete: true
            Allows Purge: true
Allows Per-Message TTL: false
          Allows Rollups: false

Limits:

        Maximum Messages: unlimited
     Maximum Per Subject: unlimited
           Maximum Bytes: unlimited
             Maximum Age: unlimited
    Maximum Message Size: unlimited
       Maximum Consumers: unlimited

State:

                Messages: 0
                   Bytes: 0 B
          First Sequence: 0
           Last Sequence: 0
        Active Consumers: 0

إنشاء durable pull مستهلك

$ nats consumer add
? Select a Stream stream_name
? Consumer name consumer_name
? Delivery target (empty for Pull Consumers)
? Start policy (all, new, last, subject, 1h, msg sequence) all
? Acknowledgment policy explicit
? Replay policy instant
? Filter Stream by subjects (blank for all)
? Maximum Allowed Deliveries -1
? Maximum Acknowledgments Pending 0
? Deliver headers only without bodies No
? Add a Retry Backoff Policy No
Information for Consumer stream_name > consumer_name created 2025-10-03T14:13:51+03:00

Configuration:

                    Name: consumer_name
               Pull Mode: true
          Deliver Policy: All
              Ack Policy: Explicit
                Ack Wait: 30.00s
           Replay Policy: Instant
         Max Ack Pending: 1,000
       Max Waiting Pulls: 512

State:

Last Delivered Message: Consumer sequence: 0 Stream sequence: 0
    Acknowledgment Floor: Consumer sequence: 0 Stream sequence: 0
        Outstanding Acks: 0 out of maximum 1,000
    Redelivered Messages: 0
    Unprocessed Messages: 0
           Waiting Pulls: 0 of maximum 512

بعد إنشاء stream وdurable pull مستهلك، يمكننا إنشاء جدول باستخدام محرك NATS. وللقيام بذلك، يجب تهيئة: nats_stream وnats_consumer_name وnats_subjects:

CREATE TABLE nats_jet_stream (
    key UInt64,
    value UInt64
  ) ENGINE NATS
    SETTINGS  nats_url = 'localhost:4222',
              nats_stream = 'stream_name',
              nats_consumer_name = 'consumer_name',
              nats_subjects = 'stream_subject',
              nats_format = 'JSONEachRow';

توفر جداول JetStream تسليمًا مرة واحدة على الأقل: لا يتم الإقرار بالرسالة إلا بعد إدراجها في العروض المادية التابعة، لذا تظل الرسالة التي يفشل إدراجها أو ينقطع إدراجها من دون إقرار، وتُعاد تسليمها. لا يوفر Core NATS (من دون JetStream) إقرارًا أو إعادة تشغيل، لذا فهو يضمن التسليم مرة واحدة على الأكثر، وتُفقد الرسالة التي ينقطع تسليمها.

​

متانة البيانات

ينطبق هذا القسم على JetStream فقط. لا يوفّر Core NATS إقرارات استلام ويعمل بأسلوب التسليم مرة واحدة على الأكثر، كما هو موضح أعلاه، لذلك لا توجد فترة يمكن خلالها فقدان رسالة تم الإقرار باستلامها.

قد يفقد جدول JetStream بصمت صفوفًا سبق استهلاكها إذا فُقدت ذاكرة تخزين الصفحات المؤقتة لنظام التشغيل قبل كتابة البيانات المُدرجة على القرص. بعد دفع دفعة إلى العروض المادية التابعة، يقرّ المستهلك باستلام تلك الرسائل، مما يسمح للتدفق بتجاوزها. لكن الصفوف المُدرجة لا تصبح دائمة إلا بعد إجراء fsync للجزء الهدف، وهو ما لا يحدث بصورة متزامنة افتراضيًا (fsync_after_insert = 0). إذا فُقدت ذاكرة التخزين المؤقت للصفحات بعد الإقرار بالاستلام وقبل إجراء fsync للجزء الهدف، فلن تُعاد تسليم الرسائل، وبالتالي تُفقد الصفوف دون ظهور أي خطأ، وتكون نتيجة count() أصغر فحسب. لا يكشف إنهاء العملية العادي عن ذلك، لأن النواة تحتفظ بذاكرة التخزين المؤقت للصفحات وتكتبها في النهاية. لكن فقدان ذاكرة التخزين المؤقت للصفحات يكشف هذه المشكلة؛ ومن أمثلته انقطاع الطاقة على مستوى الجهاز وإعادة تشغيل غير نظيفة للمضيف أو النواة.

بالنسبة إلى مسار الاستهلاك الموصى به عبر العرض المادي (لا يُرسل الإقرار بالاستلام إلا بعد اكتمال مسار الإدراج بالكامل)، فإن ضبط fsync_after_insert = 1 (وfsync_part_directory = 1) على جداول MergeTree الهدف يجعل الأجزاء المُدرجة دائمة قبل إرسال الإقرار بالاستلام، مما يضيّق هذه الفترة بدرجة كبيرة. يجب تمكين الإعداد على كل جدول MergeTree تُدرج فيه الدفعة، بما في ذلك أهداف العروض المادية المتسلسلة؛ إذ يمكن لأي جدول من هذا النوع يُترك على الإعداد الافتراضي أن يفقد جزأه. لا تكتسب المكونات الوسيطة غير المتزامنة المتانة من هذا الإعداد وحده: فعلى سبيل المثال، يُدرج الهدف Distributed البيانات في الخلفية عندما يكون distributed_foreground_insert = 0، وهو الإعداد الافتراضي خارج ClickHouse Cloud، لذا يحتاج إلى إعدادات متانة خاصة به أو إلى إدراج متزامن. لا ينطبق هذا التخفيف أيضًا على INSERT ... SELECT ... FROM <nats_table> مباشر مع nats_commit_on_select = 1، حيث يُقر باستلام الرسائل عند وصول القراءة إلى نهايتها، بدلًا من بعد أن تكتب الوجهة جزءًا دائمًا.