` | MAP | Map(K, V) | ✅ | |
| oneof | STRUCT | Tuple / Variant | ⚠️ | راجع القسم أدناه حول ترجمة oneof إلى مخطط ClickHouse |
| google.protobuf.DoubleValue | FLOAT64 | Nullable(Float64) | ✅ | |
| google.protobuf.FloatValue | FLOAT32 | Nullable(Float32) | ✅ | |
| google.protobuf.Int64Value | INT64 | Nullable(Int64) | ✅ | |
| google.protobuf.UInt64Value | INT64 | Nullable(UInt64) | ✅ | |
| google.protobuf.UInt32Value | INT64 | Nullable(UInt32) | ✅ | |
| google.protobuf.Int32Value | INT32 | Nullable(Int32) | ✅ | |
| google.protobuf.BoolValue | BOOLEAN | Nullable(Bool) | ✅ | |
| google.protobuf.StringValue | STRING | Nullable(String) | ✅ | |
| google.protobuf.BytesValue | BYTES | Nullable(String) | ✅ | |
| google.protobuf.Timestamp | org.apache.kafka.connect.data.Timestamp | DateTime64(3) | ✅ | |
| google.type.Date | org.apache.kafka.connect.data.Date | Date | ✅ | |
| google.type.TimeOfDay | org.apache.kafka.connect.data.Time | Int32 / Int64 | ✅ | |
| google.protobuf.Duration | STRUCT | Tuple(`seconds` Int64, `nano` Nullable(Int32)) | ✅ | |
| google.protobuf.Any | *N/A* | *N/A* | ❌ | |
| google.protobuf.Empty | *N/A* | *N/A* | ❌ | |
ارجع إلى [أنواع البيانات المدعومة](#supported-data-types) للاطلاع على التعيين بين أنواع Kafka Connect وأنواع ClickHouse.
#### ملاحظة حول ترجمة حقول `oneof` إلى أعمدة ClickHouse
لا يدعم الموصّل ترجمة اتحادات في Protobuf (`oneof`) إلى النوع Variant في ClickHouse. وبدلًا من ذلك، أدرِج حقول `oneof` كحقول Nullable منفصلة في مخطط جدول ClickHouse لديك.
على سبيل المثال:
```protobuf theme={null}
syntax = "proto3";
package com.clickhouse.kafka.connect.proto.test;
message StringIntUnion {
oneof mixed {
string mixed_string = 2;
int32 mixed_int = 3;
}
}
```
ويُترجم إلى تعريف جدول ClickHouse التالي:
```sql theme={null}
CREATE TABLE IF NOT EXISTS `StringIntUnion`
(
mixed_string Nullable(String),
mixed_int Nullable(Int32)
) ENGINE = ...;
```
#### مخططات Protobuf غير المدعومة
مخططات Protobuf التالية غير مدعومة في الموصل:
* بُنى union متعددة الرسائل (**قبل الإصدار 26.1 من CH**)
```protobuf theme={null}
syntax = "proto3";
package com.clickhouse.kafka.connect.proto.test;
message TwoRecords {
oneof payload {
TypeA type_a = 2;
TypeB type_b = 3;
}
// translates to Nullable(Tuple(label String)) in ClickHouse, which is unsupported
message TypeA {
string label = 1;
}
// translates to Nullable(Tuple(count Int32)) in ClickHouse, which is unsupported
message TypeB {
int32 count = 1;
}
}
```
اعتبارًا من الإصدار 26.1 من CH وما بعده، أصبح هذا المخطط مدعومًا عند ضبط `allow_experimental_nullable_tuple_type=1` (راجع [صفحة التوثيق هذه](/docs/ar/reference/settings/session-settings#allow_experimental_nullable_tuple_type)).
### دعم مخطط JSON
```json theme={null}
{
"name": "clickhouse-connect",
"config": {
"connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
...
"value.converter": "org.apache.kafka.connect.json.JsonConverter",
}
}
```
### دعم String
يدعم الموصّل محوّل String في تنسيقات ClickHouse المختلفة: [JSON](/docs/ar/reference/formats/JSON/JSONEachRow)، و[CSV](/docs/ar/reference/formats/CSV/CSV)، و[TSV](/docs/ar/reference/formats/TabSeparated/TabSeparated).
```json theme={null}
{
"name": "clickhouse-connect",
"config": {
"connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
...
"value.converter": "org.apache.kafka.connect.storage.StringConverter",
"customInsertFormat": "true",
"insertFormat": "CSV"
}
}
```
### التخزين المؤقت الداخلي
يتيح التخزين المؤقت الداخلي لمهمة الـ sink تجميع السجلات من عدة استدعاءات `poll()` ثم flush لها إلى ClickHouse على شكل دفعات أكبر. ويمكن أن يحسّن ذلك الإنتاجية في أعباء العمل التي ينتج فيها كل استدعاء `poll()` العديد من الدفعات الصغيرة لكل partition.
السلوك الأساسي:
* يتحكم `bufferCount` في عدد السجلات التي تُخزَّن مؤقتًا قبل الـ flush.
* يحدّد `bufferFlushTime` الحد الأقصى لوقت الانتظار (بالمللي ثانية) قبل عمل flush للسجلات المخزّنة مؤقتًا.
* لا يكون `bufferFlushTime` فعّالًا إلا عندما تكون `bufferCount > 0`.
* يُبقي `bufferCount=0` و `bufferFlushTime=0` التخزين المؤقت معطّلًا (السلوك الافتراضي).
* لا يكون التخزين المؤقت مدعومًا عندما تكون `exactlyOnce=true`.
سبب عدم توافق التخزين المؤقت مع وضع exactly-once:
يؤدي التخزين المؤقت إلى تغيير حدود الدفعات، مما يعطّل إزالة تكرار كتل ClickHouse وآلة حالات الإزاحة الخاصة بالموصل.
ولحل ذلك، يمكنك إما تعطيل وضع exactly-once باستخدام `exactlyOnce=false` في config الخاص بالموصل، أو تعطيل التخزين المؤقت باستخدام `bufferCount=0`.
مثال:
```json theme={null}
{
"name": "clickhouse-connect",
"config": {
"connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
...
"exactlyOnce": "false",
"bufferCount": "5000",
"bufferFlushTime": "2000"
}
}
```
### التسجيل
يوفَّر التسجيل تلقائيًا بواسطة Kafka Connect Platform.
يمكن تهيئة وجهة التسجيل وتنسيقه عبر [ملف التهيئة](https://docs.confluent.io/platform/current/connect/logging.html#log4j-properties-file) الخاص بـ Kafka Connect.
إذا كنت تستخدم Confluent Platform، فيمكن عرض السجلات بتشغيل أمر من واجهة سطر الأوامر (CLI):
```bash theme={null}
confluent local services connect log
```
لمزيد من التفاصيل، راجع [الدليل العملي](https://docs.confluent.io/platform/current/connect/logging.html).
### المراقبة
يبلّغ ClickHouse Kafka Connect عن مقاييس وقت التشغيل عبر [Java Management Extensions (JMX)](https://www.oracle.com/technical-resources/articles/javase/jmx.html). ويكون JMX مفعّلًا في Kafka Connector بشكل افتراضي.
#### المقاييس الخاصة بـ ClickHouse
يوفّر الموصل مقاييس مخصّصة عبر اسم MBean التالي:
```java theme={null}
com.clickhouse:type=ClickHouseKafkaConnector,name=SinkTask{id}
```
| اسم المقياس | النوع | الوصف |
| ---------------------- | ----- | ------------------------------------------------------------------------------- |
| `receivedRecords` | long | العدد الإجمالي للسجلات المستلَمة. |
| `recordProcessingTime` | long | إجمالي الوقت بالنانوثانية المستغرَق في تجميع السجلات وتحويلها إلى بنية موحَّدة. |
| `taskProcessingTime` | long | إجمالي الوقت بالنانوثانية المستغرَق في معالجة البيانات وإدخالها إلى ClickHouse. |
#### مقاييس منتِج/مستهلك Kafka
يعرض الموصّل مقاييس Kafka القياسية الخاصة بالمنتِج والمستهلك، والتي توفّر رؤى حول تدفّق البيانات، ومعدل النقل، والأداء.
**مقاييس على مستوى الموضوع:**
* `records-sent-total`: إجمالي عدد السجلات المُرسلة إلى الموضوع
* `bytes-sent-total`: إجمالي البايتات المُرسلة إلى الموضوع
* `record-send-rate`: متوسط معدل السجلات المُرسلة في الثانية
* `byte-rate`: متوسط عدد البايتات المُرسلة في الثانية
* `compression-rate`: نسبة الضغط المتحققة
**مقاييس على مستوى الـ Partition:**
* `records-sent-total`: إجمالي السجلات المُرسلة إلى الـ partition
* `bytes-sent-total`: إجمالي البايتات المُرسلة إلى الـ partition
* `records-lag`: مقدار التأخر الحالي في الـ partition
* `records-lead`: مقدار التقدّم الحالي في الـ partition
* `replica-fetch-lag`: معلومات التأخر الخاصة بالنسخ المتماثلة
**مقاييس الاتصالات على مستوى العقدة:**
* `connection-creation-total`: إجمالي الاتصالات التي أُنشئت مع عقدة Kafka
* `connection-close-total`: إجمالي الاتصالات التي أُغلقت
* `request-total`: إجمالي الطلبات المُرسلة إلى العقدة
* `response-total`: إجمالي الاستجابات المستلمة من العقدة
* `request-rate`: متوسط معدل الطلبات في الثانية
* `response-rate`: متوسط معدل الاستجابات في الثانية
تساعد هذه المقاييس في مراقبة ما يلي:
* **معدل النقل**: تتبّع معدلات إدخال البيانات
* **التأخر**: تحديد الاختناقات وتأخيرات المعالجة
* **الضغط**: قياس كفاءة ضغط البيانات
* **سلامة الاتصال**: مراقبة اتصال الشبكة واستقراره
#### مقاييس Kafka Connect Framework
يتكامل الموصل مع إطار عمل Kafka Connect ويعرض مقاييس لدورة حياة المهام وتتبع الأخطاء.
**مقاييس حالة المهام:**
* `task-count`: العدد الإجمالي للمهام في الموصل
* `running-task-count`: عدد المهام قيد التشغيل حاليًا
* `paused-task-count`: عدد المهام المتوقفة مؤقتًا حاليًا
* `failed-task-count`: عدد المهام التي فشلت
* `destroyed-task-count`: عدد المهام التي أُزيلت
* `unassigned-task-count`: عدد المهام غير المسندة
تشمل قيم حالة المهام: `running` و`paused` و`failed` و`destroyed` و`unassigned`
**مقاييس الأخطاء:**
* `deadletterqueue-produce-failures`: عدد عمليات الكتابة الفاشلة إلى DLQ
* `deadletterqueue-produce-requests`: إجمالي محاولات الكتابة إلى DLQ
* `last-error-timestamp`: الطابع الزمني لآخر خطأ
* `records-skip-total`: العدد الإجمالي للسجلات التي جرى تخطيها بسبب الأخطاء
* `records-retry-total`: العدد الإجمالي للسجلات التي أُعيدت محاولتها
* `errors-total`: العدد الإجمالي للأخطاء التي تم رصدها
**مقاييس الأداء:**
* `offset-commit-failures`: عدد حالات فشل تثبيت الإزاحة
* `offset-commit-avg-time-ms`: متوسط وقت تثبيت الإزاحة
* `offset-commit-max-time-ms`: أقصى وقت لتثبيت الإزاحة
* `put-batch-avg-time-ms`: متوسط وقت معالجة دفعة
* `put-batch-max-time-ms`: أقصى وقت لمعالجة دفعة
* `source-record-poll-total`: إجمالي السجلات التي جرى سحبها
#### أفضل ممارسات المراقبة
1. **راقب تأخر المستهلك**: تتبّع `records-lag` لكل تقسيم لتحديد مواضع اختناق المعالجة
2. **تتبّع معدلات الأخطاء**: راقب `errors-total` و`records-skip-total` لاكتشاف مشكلات جودة البيانات
3. **راقب سلامة المهام**: راقب مقاييس حالة المهام للتأكد من أنها تعمل على النحو الصحيح
4. **قِس معدل الإنتاجية**: استخدم `records-send-rate` و`byte-rate` لتتبّع أداء الاستيعاب
5. **راقب سلامة الاتصال**: تحقّق من مقاييس الاتصال على مستوى العقدة لاكتشاف مشكلات الشبكة
6. **تتبّع كفاءة الضغط**: استخدم `compression-rate` لتحسين نقل البيانات
للاطلاع على تعريفات مقاييس JMX التفصيلية وتكامل Prometheus، راجع ملف الإعدادات [jmx-export-connector.yml](https://github.com/ClickHouse/clickhouse-kafka-connect/blob/main/jmx-export-connector.yml).
### القيود
* عمليات الحذف غير مدعومة.
* يُورَّث حجم الدفعة من خصائص مستهلك Kafka.
* عند استخدام KeeperMap لتحقيق ضمان المعالجة مرة واحدة فقط وتغيير الإزاحة أو إرجاعها إلى قيمة سابقة، تحتاج إلى حذف المحتوى من KeeperMap لذلك الـموضوع المحدد. (راجع دليل استكشاف الأخطاء وإصلاحها أدناه لمزيد من التفاصيل)
### ضبط الأداء وتحسين معدل النقل
يتناول هذا القسم استراتيجيات ضبط الأداء الخاصة بـ ClickHouse Kafka Connect Sink. ويُعد ضبط الأداء أمرًا أساسيًا عند التعامل مع حالات استخدام ذات معدل نقل مرتفع، أو عندما تحتاج إلى تحسين استخدام الموارد وتقليل زمن التأخر.
#### متى تكون هناك حاجة إلى ضبط الأداء؟
تكون هناك حاجة إلى ضبط الأداء عادةً في السيناريوهات التالية:
* **أحمال العمل ذات معدل النقل المرتفع**: عند معالجة ملايين الأحداث في الثانية من موضوعات Kafka
* **تأخر المستهلك**: عندما لا يتمكن الموصّل من مجاراة معدل إنتاج البيانات، مما يؤدي إلى زيادة التأخر
* **قيود الموارد**: عندما تحتاج إلى تحسين استخدام CPU أو الذاكرة أو الشبكة
* **موضوعات متعددة**: عند الاستهلاك من عدة موضوعات كبيرة الحجم في الوقت نفسه
* **أحجام الرسائل الصغيرة**: عند التعامل مع عدد كبير من الرسائل الصغيرة التي ستستفيد من التجميع على جانب الخادم
**لا تكون هناك حاجة إلى ضبط الأداء عادةً** عندما:
* تعالج أحجامًا منخفضة إلى متوسطة (\< 10,000 رسالة/ثانية)
* يكون تأخر المستهلك مستقرًا ومقبولًا بالنسبة إلى حالة الاستخدام لديك
* تكون إعدادات الموصّل الافتراضية تلبي بالفعل متطلبات معدل النقل لديك
* يمكن لعنقود ClickHouse لديك التعامل بسهولة مع الحمل الوارد
#### فهم تدفق البيانات
قبل إجراء الضبط، من المهم فهم كيفية تدفق البيانات عبر الموصّل:
1. يقوم **Kafka Connect Framework** بجلب الرسائل من موضوعات Kafka في الخلفية
2. **يستطلع الموصّل** الرسائل من المخزن المؤقت الداخلي للإطار
3. **يجمّع الموصّل** الرسائل استنادًا إلى حجم الاستطلاع
4. **يتلقى ClickHouse** عملية الإدراج المجمّعة عبر HTTP/S
5. **يعالج ClickHouse** عملية الإدراج (بشكل متزامن أو غير متزامن)
يمكن تحسين الأداء في كل مرحلة من هذه المراحل.
#### ضبط حجم الدُفعات في Kafka Connect
أول مستويات التحسين يتمثل في التحكم بكمية البيانات التي يتلقاها الموصل في كل دفعة من Kafka.
##### إعدادات الجلب
يجلب Kafka Connect (الإطار) الرسائل من موضوعات Kafka في الخلفية، بشكل مستقل عن الموصل:
* **`fetch.min.bytes`**: الحد الأدنى لكمية البيانات قبل أن يمرّر الإطار القيم إلى الموصل (الافتراضي: 1 بايت)
* **`fetch.max.bytes`**: الحد الأقصى لكمية البيانات التي يمكن جلبها في طلب واحد (الافتراضي: 52428800 / 50 ميغابايت)
* **`fetch.max.wait.ms`**: الحد الأقصى لمدة الانتظار قبل إعادة البيانات إذا لم يتم استيفاء `fetch.min.bytes` (الافتراضي: 500 مللي ثانية)
في Confluent Cloud، يتطلب تعديل هذه الإعدادات فتح تذكرة دعم عبر Confluent Cloud.
##### إعدادات الاستطلاع
يجلب الموصّل الرسائل من المخزن المؤقت الخاص بإطار العمل عبر عملية استطلاع:
* **`max.poll.records`**: الحد الأقصى لعدد السجلات المُعادة في عملية استطلاع واحدة (الافتراضي: 500)
* **`max.partition.fetch.bytes`**: الحد الأقصى لكمية البيانات لكل تقسيم (الافتراضي: 1048576 / 1 MB)
في Confluent Cloud، يتطلّب تعديل هذه الإعدادات فتح تذكرة دعم عبر Confluent Cloud.
##### الإعدادات الموصى بها لمعدل النقل العالي
للحصول على أفضل أداء مع ClickHouse، احرص على استخدام دفعات أكبر:
```properties theme={null}
# Increase the number of records per poll
consumer.override.max.poll.records=5000
# Increase the partition fetch size (5 MB)
consumer.override.max.partition.fetch.bytes=5242880
# Optional: Increase minimum fetch size to wait for more data (1 MB)
consumer.override.fetch.min.bytes=1048576
# Optional: Reduce wait time if latency is critical
consumer.override.fetch.max.wait.ms=300
```
تتطلب الخصائص المذكورة أعلاه تمكين تجاوزات العميل في إعداد العامل عبر `connector.client.config.override.policy=All`. راجع [وثائق Kafka Connect](https://docs.confluent.io/platform/current/connect/references/allconfigs.html#override-the-worker-configuration) لمزيد من المعلومات.
**مهم**: تمثل إعدادات الجلب في Kafka Connect بيانات مضغوطة، بينما يتلقى ClickHouse بيانات غير مضغوطة. وازن بين هذه الإعدادات وفقًا لنسبة الضغط لديك.
**المفاضلات**:
* **دفعات أكبر** = أداء أفضل لإدخال البيانات إلى ClickHouse، وأجزاء أقل، وعبء إضافي أقل
* **دفعات أكبر** = استخدام أعلى للذاكرة، واحتمال زيادة زمن الانتقال من البداية إلى النهاية
* **دفعات كبيرة جدًا** = خطر حدوث timeout، أو أخطاء OutOfMemory، أو تجاوز `max.poll.interval.ms`
مزيد من التفاصيل: [وثائق Confluent](https://docs.confluent.io/platform/current/connect/references/allconfigs.html#override-the-worker-configuration) | [وثائق Kafka](https://kafka.apache.org/documentation/#consumerconfigs)
#### عمليات الإدراج غير المتزامنة
تُعد عمليات الإدراج غير المتزامنة ميزة فعّالة عندما يرسل الموصّل دفعات صغيرة نسبيًا، أو عندما تريد تحسين إدخال البيانات بشكل أكبر عبر إسناد مسؤولية التجميع إلى ClickHouse.
##### متى تستخدم عمليات الإدراج غير المتزامنة
فكّر في تمكين عمليات الإدراج غير المتزامنة عندما:
* **العديد من الدُفعات الصغيرة**: يرسل الموصّل دُفعات صغيرة ومتكررة (\< 1000 صف لكل دفعة)
* **تزامن مرتفع**: تكتب مهام موصّل متعددة إلى الجدول نفسه
* **نشر موزّع**: تشغيل العديد من مثيلات الموصّل عبر مضيفين مختلفين
* **عبء إضافي ناتج عن إنشاء الأجزاء**: تواجه أخطاء "عدد كبير جدًا من الأجزاء"
* **عبء عمل مختلط**: الجمع بين استيعاب البيانات في الوقت الفعلي وأعباء عمل الاستعلامات
لا تستخدم **عمليات الإدراج غير المتزامنة** عندما:
* كنت ترسل بالفعل دُفعات كبيرة (> 10,000 صف لكل دفعة) بوتيرة مضبوطة
* كنت تحتاج إلى ظهور البيانات فورًا (يجب أن ترى الاستعلامات البيانات على الفور)
* كانت دلالات ضمان المعالجة مرة واحدة فقط مع `wait_for_async_insert=0` تتعارض مع متطلباتك
* كان سيناريو الاستخدام لديك سيستفيد بدلًا من ذلك من تحسينات التجميع على جهة العميل
##### كيف تعمل عمليات الإدراج غير المتزامنة
عند تمكين عمليات الإدراج غير المتزامنة، يقوم ClickHouse بما يلي:
1. يستقبل استعلام الإدراج من الموصّل
2. يكتب البيانات إلى مخزن مؤقت في الذاكرة (بدلًا من كتابتها إلى القرص فورًا)
3. يعيد نجاح العملية إلى الموصّل (إذا كانت `wait_for_async_insert=0`)
4. يفرّغ المخزن المؤقت إلى القرص عند تحقق أحد الشروط التالية:
* وصول المخزن المؤقت إلى `async_insert_max_data_size` (القيمة الافتراضية: 100 MB)
* انقضاء `async_insert_busy_timeout_ms` ملّي ثانية منذ أول عملية إدراج (القيمة الافتراضية: 1000 ms)
* بلوغ الحد الأقصى لعدد الاستعلامات المتراكمة (`async_insert_max_query_number`، القيمة الافتراضية: 100)
يقلل هذا بشكل كبير عدد الأجزاء التي يتم إنشاؤها ويحسّن معدل النقل الإجمالي.
##### تمكين async inserts
أضف إعدادات async insert إلى معلَمة التهيئة `clickhouseSettings`:
```json theme={null}
{
"name": "clickhouse-connect",
"config": {
"connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
...
"clickhouseSettings": "async_insert=1,wait_for_async_insert=1"
}
}
```
**الإعدادات الرئيسية**:
* **`async_insert=1`**: تمكين الإدراجات غير المتزامنة
* **`wait_for_async_insert=1`** (موصى به): ينتظر الموصل حتى تُفرَّغ البيانات إلى تخزين ClickHouse قبل إرسال الإقرار. يوفّر ضمانات للتسليم.
* **`wait_for_async_insert=0`**: يرسل الموصل الإقرار فورًا بعد التخزين المؤقت. يوفّر أداءً أفضل، لكن قد تُفقد البيانات إذا تعطّل الخادم قبل تفريغها.
##### ضبط سلوك `async insert`
يمكنك ضبط سلوك `flush` في `async insert` بدقة:
```json theme={null}
"clickhouseSettings": "async_insert=1,wait_for_async_insert=1,async_insert_max_data_size=104857600,async_insert_busy_timeout_ms=1000"
```
معلمات الضبط الشائعة:
* **`async_insert_max_data_size`** (الافتراضي: 104857600 / 100 MB): الحد الأقصى لحجم المخزن المؤقت قبل التفريغ
* **`async_insert_busy_timeout_ms`** (الافتراضي: 1000): الحد الأقصى للوقت (مللي ثانية) قبل التفريغ
* **`async_insert_stale_timeout_ms`** (الافتراضي: 0): الوقت (مللي ثانية) منذ آخر عملية إدراج قبل التفريغ
* **`async_insert_max_query_number`** (الافتراضي: 100): الحد الأقصى لعدد الاستعلامات قبل التفريغ
**المفاضلات**:
* **الفوائد**: أجزاء بيانات أقل، وأداء دمج أفضل، وعبء أقل على CPU، وتحسّن في معدل النقل عند ارتفاع التزامن
* **اعتبارات**: لا تصبح البيانات قابلة للاستعلام فورًا، مع زيادة طفيفة في زمن الوصول من البداية إلى النهاية
* **المخاطر**: فقدان البيانات عند تعطل الخادم إذا كانت قيمة `wait_for_async_insert=0`، واحتمال زيادة الضغط على الذاكرة مع المخازن المؤقتة الكبيرة
##### الإدراجات غير المتزامنة مع ضمان المعالجة مرة واحدة فقط
عند استخدام `exactlyOnce=true` مع الإدراجات غير المتزامنة:
```json theme={null}
{
"config": {
"exactlyOnce": "true",
"clickhouseSettings": "async_insert=1,wait_for_async_insert=1"
}
}
```
**مهم**: استخدم دائمًا `wait_for_async_insert=1` مع exactly-once لضمان عدم تنفيذ عمليات commit للإزاحة إلا بعد حفظ البيانات بشكل دائم.
لمزيد من المعلومات حول async inserts، راجع [وثائق ClickHouse الخاصة بـ async inserts](/docs/ar/concepts/best-practices/selecting-an-insert-strategy#asynchronous-inserts).
#### توازي الموصّل
زِد التوازي لتحسين معدل النقل:
##### المهام لكل موصل
```json theme={null}
"tasks.max": "4"
```
تعالج كل مهمة مجموعة فرعية من أقسام موضوع. المزيد من المهام = مزيد من التوازي، ولكن:
* الحد الأقصى الفعّال لعدد المهام = عدد أقسام موضوع
* تحتفظ كل مهمة باتصالها الخاص بـ ClickHouse
* المزيد من المهام = عبء إضافي أكبر واحتمال أعلى لحدوث تنازع على الموارد
**التوصية**: ابدأ بتعيين `tasks.max` ليعادل عدد أقسام موضوع، ثم اضبطه بناءً على مقاييس CPU ومعدل النقل.
##### تجاهل الأقسام عند التجميع على دفعات
افتراضيًا، يجمع الموصل الرسائل على دفعات لكل قسم. ولزيادة معدل النقل، يمكنك التجميع على دفعات عبر الأقسام:
```json theme={null}
"ignorePartitionsWhenBatching": "true"
```
\*\* تحذير\*\*: استخدم هذا فقط عندما يكون `exactlyOnce=false`. يمكن أن يحسّن هذا الإعداد معدل النقل من خلال إنشاء دفعات أكبر، لكنه يفقد ضمانات الترتيب داخل كل قسم.
#### عدة موضوعات عالية الإنتاجية
إذا كان موصل لديك مُعدًّا للاشتراك في عدة موضوعات، وكنت تستخدم `topic2TableMap` لربط الـ موضوعات بالجداول، وكنت تواجه اختناقًا عند الإدراج يؤدي إلى تأخر المستهلك، ففكّر في إنشاء موصل مستقل لكل موضوع بدلًا من ذلك.
يرجع السبب الرئيسي في ذلك إلى أن الدُفعات تُدرَج حاليًا في كل جدول [بشكل تسلسلي](https://github.com/ClickHouse/clickhouse-kafka-connect/blob/578ac07e8be1a920aaa3b26e49183595c3edd04b/src/main/java/com/clickhouse/kafka/connect/sink/ProxySinkTask.java#L95-L100).
**التوصية**: عند التعامل مع عدة موضوعات كبيرة الحجم، قم بنشر instance واحدة من موصل لكل موضوع لتحقيق أقصى معدل نقل متوازٍ لعمليات الإدراج.
#### اعتبارات محرّك جدول ClickHouse
اختر محرّك جدول ClickHouse المناسب لحالة الاستخدام لديك:
* **`MergeTree`**: الأفضل لمعظم حالات الاستخدام، إذ يوازن بين أداء الاستعلام والإدراج
* **`ReplicatedMergeTree`**: مطلوب لتحقيق التوفّر العالي، ويضيف عبئًا إضافيًا بسبب النسخ المتماثل
* **`*MergeTree` مع `ORDER BY` مضبوط على نحو صحيح**: حسّنه بما يلائم أنماط الاستعلام لديك
**الإعدادات التي ينبغي مراعاتها**:
```sql theme={null}
CREATE TABLE my_table (...)
ENGINE = MergeTree()
ORDER BY (timestamp, id)
SETTINGS
-- Increase max insert threads for parallel part writing
max_insert_threads = 4,
-- Allow inserts with quorum for reliability (ReplicatedMergeTree)
insert_quorum = 2
```
لإعدادات الإدراج على مستوى الموصّل:
```json theme={null}
"clickhouseSettings": "insert_quorum=2,insert_quorum_timeout=60000"
```
#### تجميع الاتصالات والمهل الزمنية
يحتفظ الموصّل باتصالات HTTP مع ClickHouse. اضبط قيم المهلة للشبكات ذات زمن الوصول المرتفع:
```json theme={null}
"clickhouseSettings": "socket_timeout=300000,connection_timeout=30000"
```
* **`socket_timeout`** (الافتراضي: 30000 مللي ثانية): الحد الأقصى لمدة عمليات القراءة
* **`connection_timeout`** (الافتراضي: 10000 مللي ثانية): الحد الأقصى للوقت اللازم لإنشاء الاتصال
ارفع هذه القيم إذا واجهت أخطاء انتهاء المهلة مع الدفعات الكبيرة.
#### مراقبة الأداء واستكشاف المشكلات
راقب هذه المقاييس الأساسية:
1. **Consumer lag**: استخدم أدوات مراقبة Kafka لتتبّع التأخر في كل تقسيم
2. **مقاييس الموصل**: راقب `receivedRecords` و`recordProcessingTime` و`taskProcessingTime` عبر JMX (راجع [Monitoring](#monitoring))
3. **مقاييس ClickHouse**:
* `system.asynchronous_inserts`: راقب استخدام المخزن المؤقت لعملية الإدراج غير المتزامنة
* `system.parts`: راقب عدد الأجزاء لاكتشاف مشكلات الدمج
* `system.merges`: راقب عمليات الدمج النشطة
* `system.events`: تتبّع `InsertedRows` و`InsertedBytes` و`FailedInsertQuery`
**مشكلات الأداء الشائعة**:
| العَرَض | السبب المحتمل | الحل |
| ---------------------- | -------------------------------- | --------------------------------------------------------- |
| ارتفاع Consumer lag | batches صغيرة جدًا | زِد `max.poll.records`، وفعّل عملية الإدراج غير المتزامنة |
| أخطاء "Too many parts" | inserts صغيرة ومتكررة | فعّل عملية الإدراج غير المتزامنة، وزِد حجم batch |
| أخطاء timeout | حجم batch كبير أو بطء الشبكة | قلّل حجم batch، وزِد `socket_timeout`، وتحقّق من الشبكة |
| ارتفاع CPU usage | عدد كبير جدًا من الأجزاء الصغيرة | فعّل عملية الإدراج غير المتزامنة، وزِد إعدادات الدمج |
| أخطاء OutOfMemory | حجم batch كبير جدًا | قلّل `max.poll.records` و`max.partition.fetch.bytes` |
| حمل المهام غير متوازن | توزيع التقسيمات غير متوازن | أعد موازنة التقسيمات أو اضبط `tasks.max` |
#### ملخص أفضل الممارسات
1. **ابدأ بالإعدادات الافتراضية**، ثم قِس واضبط استنادًا إلى الأداء الفعلي
2. **فضّل الدُفعات الأكبر**: استهدف 10,000-100,000 صف لكل insert متى أمكن
3. **استخدم عمليات الإدراج غير المتزامنة** عند إرسال الكثير من الدُفعات الصغيرة أو عند ارتفاع التزامن
4. **استخدم دائمًا `wait_for_async_insert=1`** مع ضمان المعالجة مرة واحدة فقط
5. **وسّع أفقيًا**: زِد `tasks.max` حتى يصل إلى عدد التقسيمات
6. **موصل واحد لكل topic مرتفع الحجم** لتحقيق أقصى إنتاجية
7. **راقب باستمرار**: تتبّع تأخر المستهلك، وعدد الأجزاء، ونشاط الدمج
8. **اختبر جيدًا**: اختبر دائمًا تغييرات التكوين تحت حمل واقعي قبل النشر في بيئة الإنتاج
#### مثال: إعداد لمعدل نقل عالٍ
فيما يلي مثال كامل مُحسَّن لمعدل نقل عالٍ:
```json theme={null}
{
"name": "clickhouse-high-throughput",
"config": {
"connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
"tasks.max": "8",
"topics": "high_volume_topic",
"hostname": "my-clickhouse-host.cloud",
"port": "8443",
"database": "default",
"username": "default",
"password": "",
"ssl": "true",
"value.converter": "org.apache.kafka.connect.json.JsonConverter",
"value.converter.schemas.enable": "false",
"exactlyOnce": "false",
"ignorePartitionsWhenBatching": "true",
"consumer.override.max.poll.records": "10000",
"consumer.override.max.partition.fetch.bytes": "5242880",
"consumer.override.fetch.min.bytes": "1048576",
"consumer.override.fetch.max.wait.ms": "500",
"clickhouseSettings": "async_insert=1,wait_for_async_insert=1,async_insert_max_data_size=16777216,async_insert_busy_timeout_ms=1000,socket_timeout=300000"
}
}
```
تتطلب إعدادات الموصل أعلاه تمكين client overrides في إعدادات العامل عبر `connector.client.config.override.policy=All`. راجع [وثائق Kafka Connect](https://docs.confluent.io/platform/current/connect/references/allconfigs.html#override-the-worker-configuration) لمزيد من المعلومات.
**هذا الإعداد**:
* يعالج ما يصل إلى 10,000 سجل في كل عملية poll
* ينشئ batches عبر الأقسام للحصول على عمليات insert أكبر
* يستخدم عملية الإدراج غير المتزامنة مع buffer بسعة 16 ميغابايت
* يشغّل 8 tasks متوازية (طابقها مع عدد الأقسام لديك)
* مُحسَّن لتحقيق أعلى معدل النقل بدلًا من الالتزام الصارم بالترتيب
### استكشاف الأخطاء وإصلاحها
#### "عدم تطابق الحالة للموضوع `[someTopic]` والتقسيم `[0]`"
يحدث هذا عندما تكون الإزاحة المخزنة في KeeperMap مختلفة عن الإزاحة المخزنة في Kafka، ويحدث ذلك عادةً عند حذف موضوع
أو تعديل الإزاحة يدويًا.
لإصلاح ذلك، ستحتاج إلى حذف القيم القديمة المخزنة لذلك الموضوع + التقسيم المحددين:
```sql theme={null}
-- First, identify the database used to store the data.
SELECT * FROM [database].connect_state
-- Identify the key that matches the topic and partition.
ALTER TABLE [database].connect_state DELETE WHERE key = [keyname]
```
قد يؤثر هذا التعديل في ضمان المعالجة مرة واحدة فقط.
#### "ما الأخطاء التي سيُعيد الموصل المحاولة معها؟"
ينصبّ التركيز حاليًا على تحديد الأخطاء العابرة التي يمكن إعادة المحاولة معها، بما في ذلك:
* `ClickHouseException` - هذا استثناء عام يمكن أن يطرحه ClickHouse.
ويُطرح عادةً عندما يكون الخادم مثقلًا، وتُعد رموز الخطأ التالية عابرة بشكل خاص:
* 3 - UNEXPECTED\_END\_OF\_FILE
* 107 - FILE\_DOESNT\_EXIST
* 159 - TIMEOUT\_EXCEEDED
* 164 - READONLY
* 202 - TOO\_MANY\_SIMULTANEOUS\_QUERIES
* 203 - NO\_FREE\_CONNECTION
* 209 - SOCKET\_TIMEOUT
* 210 - NETWORK\_ERROR
* 241 - MEMORY\_LIMIT\_EXCEEDED
* 242 - TABLE\_IS\_READ\_ONLY
* 252 - TOO\_MANY\_PARTS
* 285 - TOO\_FEW\_LIVE\_REPLICAS
* 319 - UNKNOWN\_STATUS\_OF\_INSERT
* 425 - SYSTEM\_ERROR
* 999 - KEEPER\_EXCEPTION
* `SocketTimeoutException` - يُطرح هذا عندما تنتهي مهلة المقبس.
* `UnknownHostException` - يُطرح هذا عندما يتعذر تحديد المضيف.
* `IOException` - يُطرح هذا عندما تكون هناك مشكلة في الشبكة.
#### "كل بياناتي فارغة/أصفار"
من المرجّح أن الحقول في بياناتك لا تتطابق مع الحقول في الجدول، وهذا شائع خصوصًا مع CDC (وتنسيق Debezium).
أحد الحلول الشائعة هو إضافة تحويل flatten إلى إعدادات الموصل:
```properties theme={null}
transforms=flatten
transforms.flatten.type=org.apache.kafka.connect.transforms.Flatten$Value
transforms.flatten.delimiter=_
```
سيؤدي ذلك إلى تحويل بياناتك من JSON متداخل إلى JSON مُسطَّح (باستخدام `_` كفاصل). وستتبع الحقول في الجدول بعد ذلك التنسيق "field1\_field2\_field3" (أي "before\_id" و"after\_id" وما إلى ذلك).
#### "أريد استخدام مفاتيح Kafka الخاصة بي في ClickHouse"
لا تُحفَظ مفاتيح Kafka في حقل القيمة بشكل افتراضي، لكن يمكنك استخدام التحويل `KeyToValue` لنقل المفتاح إلى حقل القيمة (ضمن حقل جديد باسم `_key`):
```properties theme={null}
transforms=keyToValue
transforms.keyToValue.type=com.clickhouse.kafka.connect.transforms.KeyToValue
transforms.keyToValue.field=_key
```
