`). تُطبَّق التهيئة العامة أولًا، ثم تُطبَّق التهيئة على مستوى الـموضوع (إن وُجدت).
```xml theme={null}
cgrp
3000
logs
4000
smallest
logs
100000
stats
50000
logs
250
stats
400
```
للاطلاع على قائمة بخيارات التهيئة المتاحة، راجع [مرجع تكوين librdkafka](https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md). استخدم الشرطة السفلية (`_`) بدلًا من النقطة في تكوين ClickHouse. على سبيل المثال، ستُكتب `check.crcs=true` على النحو التالي: `true`.
### دعم Kerberos
للتعامل مع Kafka المتوافق مع Kerberos، أضِف العنصر الفرعي `security_protocol` بالقيمة `sasl_plaintext`. ويكفي الحصول على تذكرة منح التذاكر الخاصة بـ Kerberos وتخزينها مؤقتًا عبر إمكانات نظام التشغيل.
يمكن لـ ClickHouse الاحتفاظ ببيانات اعتماد Kerberos باستخدام ملف keytab. راعِ العناصر الفرعية `sasl_kerberos_service_name` و`sasl_kerberos_keytab` و`sasl_kerberos_principal`.
مثال:
```xml theme={null}
SASL_PLAINTEXT
/home/kafkauser/kafkauser.keytab
kafkauser/kafkahost@EXAMPLE.COM
```
## الأعمدة الافتراضية
* `_topic` — موضوع Kafka. نوع البيانات: `LowCardinality(String)`.
* `_key` — مفتاح الرسالة. نوع البيانات: `String`.
* `_offset` — إزاحة الرسالة. نوع البيانات: `UInt64`.
* `_timestamp` — الطابع الزمني للرسالة. نوع البيانات: `Nullable(DateTime)`.
* `_timestamp_ms` — الطابع الزمني للرسالة بالمللي ثانية. نوع البيانات: `Nullable(DateTime64(3))`.
* `_partition` — قسم موضوع Kafka. نوع البيانات: `UInt64`.
* `_headers.name` — مصفوفة مفاتيح ترويسات الرسالة. نوع البيانات: `Array(String)`.
* `_headers.value` — مصفوفة قيم ترويسات الرسالة. نوع البيانات: `Array(String)`.
أعمدة افتراضية إضافية عندما تكون قيمة `kafka_handle_error_mode='stream'`:
* `_raw_message` - الرسالة الخام التي تعذّر تحليلها بنجاح. نوع البيانات: `String`.
* `_error` - رسالة الاستثناء التي حدثت أثناء تعذّر التحليل. نوع البيانات: `String`.
ملاحظة: لا تتم تعبئة العمودين الافتراضيين `_raw_message` و `_error` إلا عند حدوث استثناء أثناء التحليل، ويظلان فارغين دائمًا إذا تم تحليل الرسالة بنجاح.
## ربط الأعمدة بالبيانات الوصفية لرسائل Kafka
عند إنتاج الرسائل باستخدام `INSERT INTO`، يستخدم محرك Kafka دائمًا عمودًا باسم `_key` (من النوع `String`) كمفتاح لرسالة Kafka وعمودًا باسم `_timestamp` (من النوع `DateTime`) كطابع زمني لرسالة Kafka، إذا كانت هذه الأعمدة موجودة في الجدول. وبشكل افتراضي، تظهر هذه الأعمدة أيضًا في حمولة الرسالة المُنتَجة إلى جانب الأعمدة الأخرى.
عند استخدام `kafka_map_virtual_columns_on_write = 1`، يتغير السلوك:
* `_key` (النوع `String`) — يُربَط بمفتاح رسالة Kafka.
* `_timestamp` (النوع `DateTime`) — يُربَط بالطابع الزمني لرسالة Kafka.
* `_headers.name` (النوع `Array(String)`) و `_headers.value` (النوع `Array(String)`) — يُربَطان بترويسات رسالة Kafka. ويتحوّل كل زوج `(_headers.name[i], _headers.value[i])` إلى ترويسة Kafka واحدة. ونظرًا إلى أن `_headers.name` و `_headers.value` يشتركان في البادئة Nested `_headers`، يشترط ClickHouse أن يكون حجم المصفوفتين متطابقًا في كل صف.
لا تُستبعَد الأعمدة التي تحمل هذه الأسماء من **حمولة الرسالة** إلا إذا كانت أنواعها مطابقة للأنواع المذكورة أعلاه؛ وإلا فإنها تبقى ضمن الحمولة، لذلك تظل المخططات التي صادف أنها تعيد استخدام هذه الأسماء لبيانات غير مرتبطة تعمل كما هي.
مثال:
```sql theme={null}
CREATE TABLE kafka_out
(
event_json String,
`_key` String,
`_timestamp` DateTime,
`_headers.name` Array(String),
`_headers.value` Array(String)
)
ENGINE = Kafka
SETTINGS
kafka_broker_list = 'broker:9092',
kafka_topic_list = 'events',
kafka_group_name = 'events-producer',
kafka_format = 'JSONEachRow',
kafka_map_virtual_columns_on_write = 1;
INSERT INTO kafka_out VALUES
('{"a":1}', 'session-42', now(), ['source', 'trace_id'], ['api', 'abc-123']);
```
تحتوي رسالة Kafka الناتجة على البيانات `{"event_json":"{\"a\":1}"}`، والمفتاح `session-42`، والطابع الزمني الحالي، ورأسين `source=api` و`trace_id=abc-123`.
## دعم تنسيقات البيانات
يدعم محرك Kafka جميع [التنسيقات](/ar/reference/formats/index) التي يدعمها ClickHouse.
يعتمد عدد الصفوف في رسالة Kafka واحدة على ما إذا كان التنسيق يستند إلى الصفوف أو إلى الكتل:
* في التنسيقات المستندة إلى الصفوف، يمكن التحكم في عدد الصفوف في رسالة Kafka واحدة عبر تعيين `kafka_max_rows_per_message`.
* في التنسيقات المستندة إلى الكتل، لا يمكن تقسيم block إلى أجزاء أصغر، لكن يمكن التحكم في عدد الصفوف في block واحد باستخدام الإعداد العام [max\_block\_size](/ar/reference/settings/session-settings#max_block_size).
## محرك لتخزين الإزاحات المعتمدة في ClickHouse Keeper
إذا كان `allow_experimental_kafka_offsets_storage_in_keeper` مفعّلًا، فيمكن عندئذٍ تحديد إعدادين إضافيين لمحرك جدول Kafka:
* يحدّد `kafka_keeper_path` المسار إلى الجدول في ClickHouse Keeper
* يحدّد `kafka_replica_name` اسم النسخة المتماثلة في ClickHouse Keeper
إما أن يُحدَّد الإعدادان معًا أو لا يُحدَّد أيٌّ منهما. وعند تحديدهما معًا، سيُستخدم محرك Kafka جديد وتجريبي. لا يعتمد هذا المحرك الجديد على تخزين الإزاحات المعتمدة في Kafka، بل يخزّنها في ClickHouse Keeper. وما يزال يحاول تثبيت الإزاحات في Kafka، لكنه لا يعتمد على تلك الإزاحات إلا عند إنشاء الجدول. وفي أي حالة أخرى (مثل إعادة تشغيل الجدول أو استعادته بعد حدوث خطأ)، ستُستخدم الإزاحات المخزّنة في ClickHouse Keeper كإزاحة لمواصلة استهلاك الرسائل. وإلى جانب الإزاحة المعتمدة، يخزّن أيضًا عدد الرسائل التي استُهلكت في الدفعة الأخيرة، بحيث إذا فشلت عملية `insert`، فسيُستهلك العدد نفسه من الرسائل، مما يتيح إزالة التكرار عند الحاجة.
مثال:
```sql theme={null}
CREATE TABLE experimental_kafka (key UInt64, value UInt64)
ENGINE = Kafka('localhost:19092', 'my-topic', 'my-consumer', 'JSONEachRow')
SETTINGS
kafka_keeper_path = '/clickhouse/{database}/{uuid}',
kafka_replica_name = '{replica}'
SETTINGS allow_experimental_kafka_offsets_storage_in_keeper=1;
```
### القيود المعروفة
نظرًا لأن المحرك الجديد لا يزال تجريبيًا، فهو غير جاهز للاستخدام في بيئة الإنتاج بعد. هناك بعض القيود المعروفة في هذا التنفيذ:
* قد يؤدي حذف الجدول بسرعة ثم إعادة إنشائه، أو تحديد مسار ClickHouse Keeper نفسه لمحركات مختلفة، إلى حدوث مشكلات. وكأفضل ممارسة، يمكنك استخدام `{uuid}` في `kafka_keeper_path` لتجنب تعارض المسارات.
* لضمان القراءات القابلة للتكرار، لا يمكن استهلاك الرسائل من عدة أقسام على خيط تنفيذ واحد. ومن ناحية أخرى، يجب إجراء `poll` على مستهلكي Kafka بانتظام للحفاظ على نشاطهم. ونتيجةً لهذين الشرطين، قررنا السماح بإنشاء عدة مستهلكين فقط إذا كان `kafka_thread_per_consumer` مُمكّنًا؛ وإلا فسيكون من المعقد جدًا تجنب المشكلات المرتبطة بإجراء `poll` للمستهلكين بانتظام.
**راجع أيضًا**
* [الأعمدة الافتراضية](/ar/reference/engines/table-engines/index#table_engines-virtual_columns)
* [background\_message\_broker\_schedule\_pool\_size](/ar/reference/settings/server-settings/settings#background_message_broker_schedule_pool_size)
* [system.kafka\_consumers](/ar/reference/system-tables/kafka_consumers)