> ## Documentation Index > Fetch the complete documentation index at: https://clickhouse.com/docs/llms.txt > Use this file to discover all available pages before exploring further. > استخدام محرك جدول Kafka # استخدام محرك جدول Kafka export const Image = ({img, alt, size = "lg"}) => { const normalizedSize = ["sm", "md", "lg"].includes(size) ? size : "lg"; return
{alt}
; }; يمكن استخدام محرك جدول Kafka [**لقراءة** البيانات من](#kafka-to-clickhouse) و[**لكتابة** البيانات إلى](#clickhouse-to-kafka) Apache Kafka وغيره من الوسطاء المتوافقين مع واجهة برمجة تطبيقات Kafka (مثل Redpanda وAmazon MSK).
### من Kafka إلى ClickHouse
إذا كنت تستخدم ClickHouse Cloud، فننصحك باستخدام [ClickPipes](/docs/ar/integrations/clickpipes/home) بدلاً منه. يوفّر ClickPipes دعمًا أصليًا لاتصالات الشبكة الخاصة، ولتوسيع موارد الإدخال وموارد العنقود بشكل مستقل، ولمراقبة شاملة لإدخال بيانات Kafka المتدفقة إلى ClickHouse. لاستخدام محرك جدول Kafka، ينبغي أن تكون لديك معرفة عامة بـ [العروض المادية في ClickHouse](/docs/ar/concepts/features/materialized-views/cascading-materialized-views).
#### نظرة عامة
نركّز أولًا على حالة الاستخدام الأكثر شيوعًا: استخدام محرك جدول Kafka لإدراج البيانات في ClickHouse من Kafka. يتيح محرك جدول Kafka لـ ClickHouse القراءة مباشرةً من topic Kafka. وعلى الرغم من فائدته في عرض الرسائل الموجودة في topic، فإن هذا المحرك، بحكم تصميمه، لا يتيح سوى الاسترجاع لمرة واحدة؛ أي عند تنفيذ استعلام على الجدول، يستهلك البيانات من الطابور ويزيد إزاحة المستهلك قبل إعادة النتائج إلى الجهة الطالبة. وبذلك، لا يمكن عمليًا إعادة قراءة البيانات من دون إعادة تعيين هذه الإزاحات. وللاحتفاظ بهذه البيانات بعد قراءتها من محرك الجدول، نحتاج إلى وسيلة لالتقاطها وإدراجها في جدول آخر. وتوفّر العروض المادية المعتمدة على المشغّلات هذه الوظيفة بشكل أصيل. ويبدأ العرض المادي قراءةً من محرك الجدول، ثم يتلقى دفعات من المستندات. وتحدّد العبارة `TO` وجهة البيانات، والتي تكون عادةً جدولًا من [عائلة MergeTree](/docs/ar/reference/engines/table-engines/mergetree-family/index). يوضّح الشكل التالي هذه العملية: مخطط معمارية محرك جدول Kafka
#### الخطوات
إذا كانت لديك بيانات مُعبّأة في topic الهدف، فيمكنك تكييف ما يلي لاستخدامه في مجموعة بياناتك. وبدلًا من ذلك، تتوفر مجموعة بيانات GitHub تجريبية [هنا](https://datasets-documentation.s3.eu-west-3.amazonaws.com/kafka/github_all_columns.ndjson). تُستخدم مجموعة البيانات هذه في الأمثلة أدناه، وتعتمد مخطط GitHub المبسّط وجزءًا فقط من الصفوف (وتحديدًا، نقتصر على أحداث GitHub المتعلقة بـ[مستودع ClickHouse](https://github.com/ClickHouse/ClickHouse))، مقارنةً بمجموعة البيانات الكاملة المتاحة [هنا](https://ghe.clickhouse.tech/)، اختصارًا. ومع ذلك، يظل هذا كافيًا لكي تعمل معظم الاستعلامات [المنشورة مع مجموعة البيانات](https://ghe.clickhouse.tech/). هذه الخطوة مطلوبة إذا كنت تتصل بـ Kafka آمن. لا يمكن تمرير هذه الإعدادات عبر أوامر SQL DDL، ويجب تهيئتها في ملف ClickHouse ‏config.xml. نفترض أنك تتصل بمثيل مؤمّن باستخدام SASL. وهذه هي أبسط طريقة عند التفاعل مع Confluent Cloud. ```xml theme={null} username password sasl_ssl PLAIN ``` إما أن تضع المقتطف أعلاه في ملف جديد داخل دليل `conf.d/` لديك أو تدمجه في ملفات التهيئة الحالية. للاطلاع على الإعدادات التي يمكن تهيئتها، انظر [هنا](/docs/ar/reference/engines/table-engines/integrations/kafka#configuration). سننشئ أيضًا قاعدة بيانات باسم `KafkaEngine` لاستخدامها في هذا الدليل العملي: ```sql theme={null} CREATE DATABASE KafkaEngine; ``` بمجرد إنشاء قاعدة البيانات، ستحتاج إلى الانتقال إليها: ```sql theme={null} USE KafkaEngine; ``` جهّز جدول الوجهة. في المثال أدناه، نستخدم مخطط GitHub المبسّط اختصارًا. لاحظ أنه رغم أننا نستخدم محرك الجدول MergeTree، يمكن تكييف هذا المثال بسهولة للعمل مع أي محرك ضمن [عائلة MergeTree](/docs/ar/reference/engines/table-engines/mergetree-family/index). ```sql theme={null} CREATE TABLE github ( file_time DateTime, event_type Enum('CommitCommentEvent' = 1, 'CreateEvent' = 2, 'DeleteEvent' = 3, 'ForkEvent' = 4, 'GollumEvent' = 5, 'IssueCommentEvent' = 6, 'IssuesEvent' = 7, 'MemberEvent' = 8, 'PublicEvent' = 9, 'PullRequestEvent' = 10, 'PullRequestReviewCommentEvent' = 11, 'PushEvent' = 12, 'ReleaseEvent' = 13, 'SponsorshipEvent' = 14, 'WatchEvent' = 15, 'GistEvent' = 16, 'FollowEvent' = 17, 'DownloadEvent' = 18, 'PullRequestReviewEvent' = 19, 'ForkApplyEvent' = 20, 'Event' = 21, 'TeamAddEvent' = 22), actor_login LowCardinality(String), repo_name LowCardinality(String), created_at DateTime, updated_at DateTime, action Enum('none' = 0, 'created' = 1, 'added' = 2, 'edited' = 3, 'deleted' = 4, 'opened' = 5, 'closed' = 6, 'reopened' = 7, 'assigned' = 8, 'unassigned' = 9, 'labeled' = 10, 'unlabeled' = 11, 'review_requested' = 12, 'review_request_removed' = 13, 'synchronize' = 14, 'started' = 15, 'published' = 16, 'update' = 17, 'create' = 18, 'fork' = 19, 'merged' = 20), comment_id UInt64, path String, ref LowCardinality(String), ref_type Enum('none' = 0, 'branch' = 1, 'tag' = 2, 'repository' = 3, 'unknown' = 4), creator_user_login LowCardinality(String), number UInt32, title String, labels Array(LowCardinality(String)), state Enum('none' = 0, 'open' = 1, 'closed' = 2), assignee LowCardinality(String), assignees Array(LowCardinality(String)), closed_at DateTime, merged_at DateTime, merge_commit_sha String, requested_reviewers Array(LowCardinality(String)), merged_by LowCardinality(String), review_comments UInt32, member_login LowCardinality(String) ) ENGINE = MergeTree ORDER BY (event_type, repo_name, created_at) ``` بعد ذلك، سننشئ topic. هناك عدة أدوات يمكننا استخدامها لهذا الغرض. إذا كنا نشغّل Kafka محليًا على أجهزتنا أو داخل حاوية Docker، فإن [RPK](https://docs.redpanda.com/current/get-started/rpk-install/) يعمل بكفاءة. يمكننا إنشاء topic باسم `github` مع 5 partitions عبر تشغيل الأمر التالي: ```bash theme={null} rpk topic create -p 5 github --brokers : ``` إذا كنّا نشغّل Kafka على Confluent Cloud، فقد نفضّل استخدام [Confluent CLI](https://docs.confluent.io/platform/current/tutorials/examples/clients/docs/kcat.html#produce-records): ```bash theme={null} confluent kafka topic create --if-not-exists github ``` نحتاج الآن إلى إرسال بعض البيانات إلى هذا الـ topic، وسنفعل ذلك باستخدام [kcat](https://github.com/edenhill/kcat). يمكننا تشغيل أمر مشابه لما يلي إذا كنا نشغّل Kafka محليًا مع تعطيل المصادقة: ```bash theme={null} cat github_all_columns.ndjson | kcat -P \ -b : \ -t github ``` أو ما يلي إذا كان عنقود Kafka لدينا يستخدم SASL للمصادقة: ```bash theme={null} cat github_all_columns.ndjson | kcat -P \ -b : \ -t github -X security.protocol=sasl_ssl \ -X sasl.mechanisms=PLAIN \ -X sasl.username= \ -X sasl.password= \ ``` تحتوي مجموعة البيانات على 200,000 صف، لذا يُفترض إدخالها خلال بضع ثوانٍ فقط. إذا كنت تريد العمل مع مجموعة بيانات أكبر، فألقِ نظرة على [قسم مجموعات البيانات الكبيرة](https://github.com/ClickHouse/kafka-samples/tree/main/producer#large-datasets) في مستودع GitHub ‏[ClickHouse/kafka-samples](https://github.com/ClickHouse/kafka-samples). ينشئ المثال أدناه محرك جدول له المخطط نفسه لجدول MergeTree. وليس هذا ضروريًا تمامًا، إذ يمكن أن يكون لديك اسم مستعار أو أعمدة مؤقتة في جدول الوجهة. لكن الإعدادات مهمة — لاحظ استخدام `JSONEachRow` بوصفه نوع البيانات لاستهلاك JSON من موضوع Kafka. وتمثّل القيمتان `github` و`clickhouse` اسم الموضوع واسم مجموعة المستهلكين، على الترتيب. ويمكن أن تكون الموضوعات في الواقع قائمة من القيم. ```sql theme={null} CREATE TABLE github_queue ( file_time DateTime, event_type Enum('CommitCommentEvent' = 1, 'CreateEvent' = 2, 'DeleteEvent' = 3, 'ForkEvent' = 4, 'GollumEvent' = 5, 'IssueCommentEvent' = 6, 'IssuesEvent' = 7, 'MemberEvent' = 8, 'PublicEvent' = 9, 'PullRequestEvent' = 10, 'PullRequestReviewCommentEvent' = 11, 'PushEvent' = 12, 'ReleaseEvent' = 13, 'SponsorshipEvent' = 14, 'WatchEvent' = 15, 'GistEvent' = 16, 'FollowEvent' = 17, 'DownloadEvent' = 18, 'PullRequestReviewEvent' = 19, 'ForkApplyEvent' = 20, 'Event' = 21, 'TeamAddEvent' = 22), actor_login LowCardinality(String), repo_name LowCardinality(String), created_at DateTime, updated_at DateTime, action Enum('none' = 0, 'created' = 1, 'added' = 2, 'edited' = 3, 'deleted' = 4, 'opened' = 5, 'closed' = 6, 'reopened' = 7, 'assigned' = 8, 'unassigned' = 9, 'labeled' = 10, 'unlabeled' = 11, 'review_requested' = 12, 'review_request_removed' = 13, 'synchronize' = 14, 'started' = 15, 'published' = 16, 'update' = 17, 'create' = 18, 'fork' = 19, 'merged' = 20), comment_id UInt64, path String, ref LowCardinality(String), ref_type Enum('none' = 0, 'branch' = 1, 'tag' = 2, 'repository' = 3, 'unknown' = 4), creator_user_login LowCardinality(String), number UInt32, title String, labels Array(LowCardinality(String)), state Enum('none' = 0, 'open' = 1, 'closed' = 2), assignee LowCardinality(String), assignees Array(LowCardinality(String)), closed_at DateTime, merged_at DateTime, merge_commit_sha String, requested_reviewers Array(LowCardinality(String)), merged_by LowCardinality(String), review_comments UInt32, member_login LowCardinality(String) ) ENGINE = Kafka('kafka_host:9092', 'github', 'clickhouse', 'JSONEachRow') SETTINGS kafka_thread_per_consumer = 0, kafka_num_consumers = 1; ``` نناقش أدناه إعدادات المحرك وضبط الأداء. عند هذه المرحلة، ينبغي أن يقرأ استعلام select بسيط على الجدول `github_queue` بعض الصفوف. لاحظ أن هذا سيؤدي إلى تقديم offsets الخاصة بالمستهلك، مما يمنع إعادة قراءة هذه الصفوف دون [إعادة تعيين](#common-operations). لاحظ أيضًا قيمة limit والمعلمة المطلوبة `stream_like_engine_allow_direct_select.` سيربط العرض المادي بين الجدولين اللذين أُنشئا سابقًا، إذ يقرأ البيانات من محرك الجدول Kafka ويُدرجها في جدول MergeTree الهدف. يمكننا إجراء عدد من تحويلات البيانات. سنجري عملية قراءة وإدراج بسيطة. ويفترض استخدام \* أن أسماء الأعمدة متطابقة تمامًا (مع مراعاة حالة الأحرف). ```sql theme={null} CREATE MATERIALIZED VIEW github_mv TO github AS SELECT * FROM github_queue; ``` عند إنشائه، يتصل العرض المادي بمحرك Kafka ويبدأ القراءة، مُدرجًا الصفوف في جدول الوجهة. وستستمر هذه العملية بلا توقف، مع استهلاك أي رسائل جديدة تُدرج في Kafka. يمكنك إعادة تشغيل برنامج الإدراج النصي لإدراج المزيد من الرسائل في Kafka. تأكد من أن البيانات موجودة في جدول الوجهة: ```sql theme={null} SELECT count() FROM github; ``` يُفترض أن ترى 200,000 صفًا: ```response theme={null} ┌─count()─┐ │ 200000 │ └─────────┘ ```
#### العمليات الشائعة
##### إيقاف استهلاك الرسائل & إعادة تشغيله
لإيقاف استهلاك الرسائل، يمكنك فصل جدول محرك Kafka: ```sql theme={null} DETACH TABLE github_queue; ``` لن يؤثر هذا في إزاحات مجموعة المستهلك. ولاستئناف الاستهلاك ومتابعته من الإزاحة السابقة، أعد إرفاق الجدول. ```sql theme={null} ATTACH TABLE github_queue; ```
##### إضافة البيانات الوصفية لـ Kafka
قد يكون من المفيد تتبّع البيانات الوصفية لرسائل Kafka الأصلية بعد إدخالها إلى ClickHouse. على سبيل المثال، قد نرغب في معرفة مقدار ما استهلكناه من topic أو partition معيّنين. ولهذا الغرض، يوفّر محرك جدول Kafka عدة [أعمدة افتراضية](/docs/ar/reference/engines/table-engines/index#table_engines-virtual_columns). ويمكن حفظها كأعمدة في الجدول الهدف من خلال تعديل المخطط وعبارة select الخاصة بالعرض المادي. أولاً، ننفّذ عملية الإيقاف الموضّحة أعلاه قبل إضافة أعمدة إلى الجدول الهدف. ```sql theme={null} DETACH TABLE github_queue; ``` فيما يلي نضيف أعمدة معلومات لتحديد الـ topic المصدر والـ partition الذي جاء منه الصف. ```sql theme={null} ALTER TABLE github ADD COLUMN topic String, ADD COLUMN partition UInt64; ``` بعد ذلك، نحتاج إلى التأكد من تعيين الأعمدة الافتراضية على النحو المطلوب. تبدأ الأعمدة الافتراضية بالبادئة `_`. يمكن العثور على قائمة كاملة بالأعمدة الافتراضية [هنا](/docs/ar/reference/engines/table-engines/integrations/kafka#virtual-columns). لتحديث جدولنا بالأعمدة الافتراضية، سنحتاج إلى حذف العرض المادي، ثم إعادة إرفاق جدول محرك Kafka، ثم إعادة إنشاء العرض المادي. ```sql theme={null} DROP VIEW github_mv; ``` ```sql theme={null} ATTACH TABLE github_queue; ``` ```sql theme={null} CREATE MATERIALIZED VIEW github_mv TO github AS SELECT *, _topic AS topic, _partition as partition FROM github_queue; ``` يجب أن تتضمن الصفوف التي جرى استهلاكها حديثًا البيانات الوصفية. ```sql theme={null} SELECT actor_login, event_type, created_at, topic, partition FROM github LIMIT 10; ``` تكون النتيجة كما يلي: | actor\_login | event\_type | created\_at | topic | partition | | :------------ | :----------------- | :------------------ | :----- | :-------- | | IgorMinar | CommitCommentEvent | 2011-02-12 02:22:00 | github | 0 | | queeup | CommitCommentEvent | 2011-02-12 02:23:23 | github | 0 | | IgorMinar | CommitCommentEvent | 2011-02-12 02:23:24 | github | 0 | | IgorMinar | CommitCommentEvent | 2011-02-12 02:24:50 | github | 0 | | IgorMinar | CommitCommentEvent | 2011-02-12 02:25:20 | github | 0 | | dapi | CommitCommentEvent | 2011-02-12 06:18:36 | github | 0 | | sourcerebels | CommitCommentEvent | 2011-02-12 06:34:10 | github | 0 | | jamierumbelow | CommitCommentEvent | 2011-02-12 12:21:40 | github | 0 | | jpn | CommitCommentEvent | 2011-02-12 12:24:31 | github | 0 | | Oxonium | CommitCommentEvent | 2011-02-12 12:31:28 | github | 0 |
##### تعديل إعدادات محرك Kafka
نوصي بإسقاط جدول محرك Kafka ثم إعادة إنشائه بالإعدادات الجديدة. لا حاجة إلى تعديل العرض المادي أثناء هذه العملية، إذ سيُستأنف استهلاك الرسائل بمجرد إعادة إنشاء جدول محرك Kafka.
##### استكشاف المشكلات وإصلاحها
لا يتم الإبلاغ عن الأخطاء، مثل مشكلات المصادقة، في الاستجابات لأوامر DDL الخاصة بمحرك Kafka. لتشخيص المشكلات، نوصي باستخدام ملف السجل الرئيسي لـ ClickHouse، وهو clickhouse-server.err.log. ويمكن تمكين تسجيل تتبّع إضافي لمكتبة عميل Kafka الأساسية [librdkafka](https://github.com/edenhill/librdkafka) من خلال التكوين. ```xml theme={null} all ```
##### التعامل مع الرسائل غير السليمة
غالبًا ما يُستخدم Kafka بوصفه "مكبًا" للبيانات. ويؤدي ذلك إلى احتواء الموضوعات على تنسيقات رسائل مختلطة وأسماء حقول غير متسقة. تجنّب ذلك، واستفد من ميزات Kafka مثل Kafka Streams أو ksqlDB لضمان أن تكون الرسائل سليمة ومتسقة قبل إدراجها في Kafka. وإذا لم تكن هذه الخيارات ممكنة، فإن ClickHouse يوفّر بعض الميزات التي يمكن أن تساعد. * تعامل مع حقل الرسالة على أنه سلاسل نصية. ويمكن استخدام الدوال في عبارة `materialized view` لإجراء التنظيف وCAST عند الحاجة. لا ينبغي أن يُعدّ هذا حلًا لبيئة production، لكنه قد يساعد في الإدخال لمرة واحدة. * إذا كنت تستهلك JSON من topic، باستخدام التنسيق JSONEachRow، فاستخدم الإعداد [`input_format_skip_unknown_fields`](/docs/ar/reference/settings/formats#input_format_skip_unknown_fields). عند كتابة البيانات، يرفع ClickHouse افتراضيًا استثناءً إذا كانت بيانات الإدخال تحتوي على أعمدة غير موجودة في الجدول الهدف. ومع ذلك، إذا كان هذا الخيار مفعّلًا، فسيتم تجاهل هذه الأعمدة الزائدة. ومرة أخرى، لا يُعدّ هذا حلًا على مستوى production وقد يسبب التباسًا للآخرين. * ضع في اعتبارك الإعداد `kafka_skip_broken_messages`. يتطلب هذا من المستخدم تحديد مستوى التحمّل لكل كتلة تجاه الرسائل غير السليمة، وذلك في سياق kafka\_max\_block\_size. وإذا تم تجاوز هذا التحمّل (ويُقاس بعدد الرسائل المطلق)، فسيعود سلوك الاستثناء المعتاد، وسيتم تخطي الرسائل الأخرى.
##### دلالات التسليم والتحديات المرتبطة بالتكرارات
يستخدم محرك جدول Kafka دلالات تسليم من نوع "مرة واحدة على الأقل". ومن الممكن حدوث تكرارات في عدة حالات نادرة معروفة. فعلى سبيل المثال، قد تُقرأ الرسائل من Kafka وتُدرج بنجاح في ClickHouse. لكن قبل اعتماد الإزاحة الجديدة، قد ينقطع الاتصال بـ Kafka. وفي هذه الحالة، يلزم إعادة محاولة الكتلة. وقد تتم [إزالة التكرار من ](/docs/ar/reference/engines/table-engines/mergetree-family/replication)الكتلة باستخدام distributed table أو ReplicatedMergeTree كجدول هدف. ومع أن ذلك يقلل من احتمال تكرار الصفوف، فإنه يعتمد على كتل متطابقة. وقد تؤدي أحداث مثل Kafka rebalancing إلى إبطال هذا الافتراض، مما يسبب تكرارات في حالات نادرة.
##### عمليات الإدراج القائمة على النصاب
قد تحتاج إلى [عمليات الإدراج القائمة على النصاب](/docs/ar/reference/settings/session-settings#insert_quorum) في الحالات التي تتطلب ضمانات أعلى لوصول البيانات في ClickHouse. لا يمكن تعيين هذا على العرض المادي أو الجدول الهدف. ومع ذلك، يمكن تعيينه لملفات تعريف المستخدمين، على سبيل المثال. ```xml theme={null} 2 ```
### ClickHouse إلى Kafka
على الرغم من أن هذا يُعد حالة استخدام أقل شيوعًا، يمكن أيضًا الاحتفاظ ببيانات ClickHouse في Kafka. على سبيل المثال، سنُدرج الصفوف يدويًا في محرك جدول Kafka. سيقرأ محرك Kafka نفسه هذه البيانات، ثم يضعها عرضه المادي في جدول MergeTree. وأخيرًا، نوضّح كيفية استخدام العروض المادية مع عمليات الإدراج إلى Kafka لقراءة الجداول من الجداول المصدر الموجودة.
#### الخطوات
يمكن توضيح هدفنا الأولي بأفضل شكل كما يلي: مخطط محرك جدول Kafka مع عمليات الإدراج نفترض أنك أنشأت الجداول وطرق العرض ضمن خطوات [Kafka إلى ClickHouse](#kafka-to-clickhouse)، وأن الـ topic قد استُهلك بالكامل. أولًا، تحقّق من عدد صفوف الجدول الهدف. ```sql theme={null} SELECT count() FROM github; ``` يُفترض أن يكون لديك 200,000 صف: ```response theme={null} ┌─count()─┐ │ 200000 │ └─────────┘ ``` أدرِج الآن صفوفًا من الجدول الهدف GitHub مرة أخرى في محرك جدول Kafka ‏github\_queue. لاحظ كيف نستخدم تنسيق JSONEachRow وكيف نستخدم LIMIT لتقييد استعلام SELECT إلى 100. ```sql theme={null} INSERT INTO github_queue SELECT * FROM github LIMIT 100 FORMAT JSONEachRow ``` أعِد عدّ الصفوف في GitHub للتأكد من أنها قد زادت بمقدار 100. وكما هو موضح في المخطط أعلاه، أُدرِجت الصفوف في Kafka عبر محرك جدول Kafka قبل أن يعيد المحرّك نفسه قراءتها ويُدرجها عرضنا المادي في جدول GitHub الهدف! ```sql theme={null} SELECT count() FROM github; ``` يُفترض أن ترى 100 صف إضافيًا: ```response theme={null} ┌─count()─┐ │ 200100 │ └─────────┘ ``` يمكننا استخدام العروض المادية لإرسال الرسائل إلى محرك Kafka (وإلى topic) عند إدراج المستندات في جدول. وعند إدراج صفوف في جدول GitHub، يتم تشغيل عرض مادي، ما يؤدي إلى إدراج الصفوف مرة أخرى في محرك Kafka وفي topic جديد. ويوضح الشكل التالي ذلك بشكل أفضل: مخطط محرك جدول Kafka مع العروض المادية أنشئ Kafka topic جديدًا باسم `github_out` أو ما يعادله. وتأكد من أن محرك جدول Kafka باسم `github_out_queue` يشير إلى هذا الـ topic. ```sql theme={null} CREATE TABLE github_out_queue ( file_time DateTime, event_type Enum('CommitCommentEvent' = 1, 'CreateEvent' = 2, 'DeleteEvent' = 3, 'ForkEvent' = 4, 'GollumEvent' = 5, 'IssueCommentEvent' = 6, 'IssuesEvent' = 7, 'MemberEvent' = 8, 'PublicEvent' = 9, 'PullRequestEvent' = 10, 'PullRequestReviewCommentEvent' = 11, 'PushEvent' = 12, 'ReleaseEvent' = 13, 'SponsorshipEvent' = 14, 'WatchEvent' = 15, 'GistEvent' = 16, 'FollowEvent' = 17, 'DownloadEvent' = 18, 'PullRequestReviewEvent' = 19, 'ForkApplyEvent' = 20, 'Event' = 21, 'TeamAddEvent' = 22), actor_login LowCardinality(String), repo_name LowCardinality(String), created_at DateTime, updated_at DateTime, action Enum('none' = 0, 'created' = 1, 'added' = 2, 'edited' = 3, 'deleted' = 4, 'opened' = 5, 'closed' = 6, 'reopened' = 7, 'assigned' = 8, 'unassigned' = 9, 'labeled' = 10, 'unlabeled' = 11, 'review_requested' = 12, 'review_request_removed' = 13, 'synchronize' = 14, 'started' = 15, 'published' = 16, 'update' = 17, 'create' = 18, 'fork' = 19, 'merged' = 20), comment_id UInt64, path String, ref LowCardinality(String), ref_type Enum('none' = 0, 'branch' = 1, 'tag' = 2, 'repository' = 3, 'unknown' = 4), creator_user_login LowCardinality(String), number UInt32, title String, labels Array(LowCardinality(String)), state Enum('none' = 0, 'open' = 1, 'closed' = 2), assignee LowCardinality(String), assignees Array(LowCardinality(String)), closed_at DateTime, merged_at DateTime, merge_commit_sha String, requested_reviewers Array(LowCardinality(String)), merged_by LowCardinality(String), review_comments UInt32, member_login LowCardinality(String) ) ENGINE = Kafka('host:port', 'github_out', 'clickhouse_out', 'JSONEachRow') SETTINGS kafka_thread_per_consumer = 0, kafka_num_consumers = 1; ``` أنشئ الآن عرضًا ماديًا جديدًا `github_out_mv` يشير إلى جدول GitHub، بحيث يُدرِج الصفوف في المحرّك المذكور أعلاه عند تفعيله. ونتيجةً لذلك، ستُدفَع الإضافات إلى جدول GitHub إلى Kafka topic الجديد لدينا. ```sql theme={null} CREATE MATERIALIZED VIEW github_out_mv TO github_out_queue AS SELECT file_time, event_type, actor_login, repo_name, created_at, updated_at, action, comment_id, path, ref, ref_type, creator_user_login, number, title, labels, state, assignee, assignees, closed_at, merged_at, merge_commit_sha, requested_reviewers, merged_by, review_comments, member_login FROM github FORMAT JsonEachRow; ``` إذا أجريت عملية insert في topic الأصلي المسمى github، والذي أُنشئ كجزء من [Kafka to ClickHouse](#kafka-to-clickhouse)، فستظهر المستندات تلقائيًا في topic "github\_clickhouse". أكِّد ذلك باستخدام أدوات Kafka الأصلية. على سبيل المثال، نُدرج أدناه 100 صف في topic ‏github باستخدام [kcat](https://github.com/edenhill/kcat) لـ topic مستضاف على Confluent Cloud: ```sql theme={null} head -n 10 github_all_columns.ndjson | kcat -P \ -b : \ -t github -X security.protocol=sasl_ssl \ -X sasl.mechanisms=PLAIN \ -X sasl.username= \ -X sasl.password= ``` من المفترض أن تؤكد قراءة من topic ‏`github_out` تسليم الرسائل. ```sql theme={null} kcat -C \ -b : \ -t github_out \ -X security.protocol=sasl_ssl \ -X sasl.mechanisms=PLAIN \ -X sasl.username= \ -X sasl.password= \ -e -q | wc -l ``` على الرغم من كون هذا المثال تفصيليًا، إلا أنه يوضح قوة العروض المادية عند استخدامها بالتزامن مع محرك Kafka.
### العناقيد والأداء
#### العمل مع عنقود ClickHouse
من خلال مجموعات مستهلكي Kafka، يمكن لعدة مثيلات من ClickHouse القراءة من topic نفسه. ويُعيَّن كل مستهلك إلى partition واحد من topic بنسبة 1:1. عند توسيع استهلاك ClickHouse باستخدام محرك جدول Kafka، ضع في اعتبارك أن العدد الإجمالي للمستهلكين داخل العنقود لا يمكن أن يتجاوز عدد partitions في topic. لذلك، احرص مسبقًا على تهيئة تقسيم topic بالشكل المناسب. يمكن أيضًا تهيئة عدة مثيلات من ClickHouse للقراءة من topic واحد باستخدام معرّف مجموعة المستهلكين نفسه، والذي يُحدَّد عند إنشاء محرك جدول Kafka. وبذلك، سيقرأ كل مثيل من partition واحد أو أكثر، مع إدراج المقاطع في الجدول الهدف المحلي الخاص به. ويمكن بعد ذلك تهيئة الجداول الهدف لاستخدام ReplicatedMergeTree للتعامل مع تكرار البيانات. يتيح هذا النهج توسيع قراءات Kafka بالتوازي مع عنقود ClickHouse، شريطة توفّر عدد كافٍ من partitions Kafka. مخطط محرك جدول Kafka مع عنقود ClickHouse
#### ضبط الأداء
ضع ما يلي في الاعتبار عند السعي إلى زيادة أداء معدل النقل لجدول Kafka Engine: * يختلف الأداء بحسب حجم الرسائل، والتنسيق، وأنواع الجداول المستهدفة. ويمكن اعتبار 100 ألف صف/ثانية على table engine واحد معدلًا يمكن تحقيقه. افتراضيًا، تُقرأ الرسائل على شكل كتل، ويتحكم في ذلك المعلَمة kafka\_max\_block\_size. افتراضيًا، تُضبط هذه القيمة على [max\_insert\_block\_size](/docs/ar/reference/settings/session-settings#max_insert_block_size)، وقيمتها الافتراضية 1,048,576. وما لم تكن الرسائل كبيرة جدًا، فمن الأفضل تقريبًا دائمًا زيادة هذه القيمة. كما أن القيم بين 500 ألف ومليون ليست غير شائعة. اختبر هذا وقيّم أثره في أداء معدل النقل. * يمكن زيادة عدد المستهلكين لمحرك الجدول باستخدام kafka\_num\_consumers. لكن افتراضيًا، ستُسلسل عمليات الإدراج في thread واحد ما لم يتم تغيير kafka\_thread\_per\_consumer عن القيمة الافتراضية 1. اضبطه على 1 لضمان تنفيذ عمليات التفريغ بالتوازي. لاحظ أن إنشاء جدول Kafka engine بعدد N من المستهلكين (ومع kafka\_thread\_per\_consumer=1) يكافئ منطقيًا إنشاء N من Kafka engines، لكل منها materialized view وkafka\_thread\_per\_consumer=0. * زيادة عدد المستهلكين ليست بلا تكلفة. إذ يحتفظ كل مستهلك بمخازنه المؤقتة وthreads الخاصة به، مما يزيد الحمل الإضافي على الخادم. لذلك، انتبه إلى الحمل الإضافي الناتج عن المستهلكين، ووسّع أفقيًا عبر عنقود أولًا إن أمكن. * إذا كان معدل نقل رسائل Kafka متغيرًا وكانت التأخيرات مقبولة، ففكّر في زيادة stream\_flush\_interval\_ms لضمان تفريغ كتل أكبر. * يحدّد [background\_message\_broker\_schedule\_pool\_size](/docs/ar/reference/settings/server-settings/settings#background_message_broker_schedule_pool_size) عدد threads التي تنفّذ tasks في الخلفية. وتُستخدم هذه threads في Kafka streaming. يُطبَّق هذا الإعداد عند بدء تشغيل ClickHouse server ولا يمكن تغييره ضمن session للمستخدم، وقيمته الافتراضية 16. إذا رأيت حالات timeout في السجلات، فقد يكون من المناسب زيادة هذه القيمة. * في الاتصال مع Kafka، تُستخدم مكتبة librdkafka، وهي بدورها تنشئ threads. لذلك، قد يؤدي العدد الكبير من جداول Kafka أو المستهلكين إلى عدد كبير من عمليات تبديل السياق. وزّع هذا الحمل عبر عنقود، مع الاكتفاء بتكرار الجداول المستهدفة فقط إن أمكن، أو فكّر في استخدام محرك جدول للقراءة من topics متعددة، إذ إن قائمة من القيم مدعومة. كما يمكن قراءة عدة materialized views من جدول واحد، بحيث يرشّح كلٌّ منها البيانات الخاصة بـ topic معيّن. يجب اختبار أي تغييرات في الإعدادات. نوصي بمراقبة تأخر المستهلكين في Kafka لضمان أن التوسّع لديك مناسب.
#### إعدادات إضافية
إلى جانب الإعدادات المذكورة أعلاه، قد تهمّك أيضًا الإعدادات التالية: * [Kafka\_max\_wait\_ms](/docs/ar/reference/settings/session-settings#kafka_max_wait_ms) - مدة الانتظار بالمللي ثانية لقراءة الرسائل من Kafka قبل إعادة المحاولة. يُضبط هذا الإعداد على مستوى profile المستخدم، وتكون قيمته الافتراضية 5000. يمكن أيضًا وضع [جميع الإعدادات ](https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md)الخاصة بمكتبة librdkafka الأساسية داخل ملفات تهيئة ClickHouse ضمن عنصر *kafka* - ويجب أن تكون أسماء الإعدادات عناصر XML مع استبدال النقاط بشرطات سفلية، على سبيل المثال. ```xml theme={null} false ``` هذه إعدادات مخصّصة للخبراء، وننصح بالرجوع إلى وثائق Kafka للحصول على شرحٍ وافٍ.