ClickHouse Kafka Connect Sink

ClickHouse Kafka Connect Sink هو موصل لـ Kafka ينقل البيانات من Kafka موضوع إلى ClickHouse table.

يُوزَّع Kafka Connector Sink بموجب ترخيص Apache 2.0

يجب أن يكون إطار عمل Kafka Connect بالإصدار 2.7 أو أحدث مثبّتًا في البيئة، وأن يعمل على Java 11 أو أحدث.

يرفض الموصل البدء عند استخدام خوادم ClickHouse بإصدار أقدم من 23.3. استخدم دائمًا أحدث إصدار، ما لم يكن لديك سبب لتثبيت إصدار أقدم.

• يأتي مزوّدًا بضمانات exactly-once جاهزة للاستخدام. ويعتمد على ميزة أساسية جديدة في ClickHouse باسم KeeperMap (تُستخدم كمخزن للحالة بواسطة الموصل)، ما يتيح معمارية مبسطة للغاية.
• دعم مخازن الحالة التابعة لجهات خارجية: يستخدم حاليًا In-memory افتراضيًا، ويمكنه أيضًا استخدام KeeperMap (ستتم إضافة Redis قريبًا).
• تكامل أصيل: يطوّره ClickHouse ويتولى صيانته ودعمه.
• يُختبَر باستمرار على ClickHouse Cloud.
• يدعم إدراج البيانات بمخطط مُعلن أو بدون مخطط.
• يدعم جميع أنواع البيانات في ClickHouse.

للاتصال بـ ClickHouse باستخدام HTTP(S)، تحتاج إلى المعلومات التالية:

تتوفر تفاصيل خدمة ClickHouse Cloud الخاصة بك في ClickHouse Cloud console. حدِّد خدمة ثم انقر على Connect:

اختر HTTPS. ستظهر تفاصيل الاتصال في مثال لأمر curl.

إذا كنت تستخدم ClickHouse مُدارًا ذاتيًا، فسيُحدِّد مسؤول ClickHouse لديك تفاصيل الاتصال.

يُوزَّع هذا الموصل على هيئة ملف JAR واحد يحتوي على جميع ملفات الفئات اللازمة لتشغيل المكوّن الإضافي.

لتثبيت المكوّن الإضافي، اتبع الخطوات التالية:

• نزّل أرشيف ZIP يحتوي على ملف JAR الخاص بالموصل من صفحة Releases في مستودع ClickHouse Kafka Connect Sink.
• استخرج محتويات ملف ZIP وانسخها إلى الموقع المطلوب.
• أضف path يتضمن دليل المكوّن الإضافي إلى إعداد plugin.path في ملف خصائص Connect للسماح لـ Confluent Platform بالعثور على المكوّن الإضافي.
• وفّر اسم الموضوع وhostname الخاص بـ ClickHouse instance وpassword في config.

connector.class=com.clickhouse.kafka.connect.ClickHouseSinkConnector
tasks.max=1
topics=<topic_name>
ssl=true
jdbcConnectionProperties=?sslmode=STRICT
security.protocol=SSL
hostname=<hostname>
database=<database_name>
password=<password>
ssl.truststore.location=/tmp/kafka.client.truststore.jks
port=8443
value.converter.schemas.enable=false
value.converter=org.apache.kafka.connect.json.JsonConverter
exactlyOnce=true
username=default
schemas.enable=false

• أعد تشغيل Confluent Platform.
• إذا كنت تستخدم Confluent Platform، فسجّل الدخول إلى واجهة المستخدم الخاصة بـ Confluent Control Center للتحقق من أن ClickHouse Sink متاح في قائمة الموصلات المتاحة.

لربط ClickHouse Sink بخادم ClickHouse، يلزم توفير ما يلي:

• تفاصيل الاتصال: اسم المضيف (مطلوب) والمنفذ (اختياري)
• بيانات اعتماد المستخدم: كلمة المرور (مطلوب) واسم المستخدم (اختياري)
• فئة الموصّل: com.clickhouse.kafka.connect.ClickHouseSinkConnector (مطلوب)
• topics أو topics.regex: مواضيع Kafka التي سيقرأها الموصّل — يجب أن تتطابق أسماء المواضيع مع أسماء الجداول (مطلوب)
• محوّلات المفتاح والقيمة: اضبطها بناءً على نوع البيانات في الموضوع لديك. وهي مطلوبة إذا لم تكن معرّفة مسبقًا في إعدادات العامل.

الجدول الكامل لخيارات التهيئة:

يقرأ ClickHouse Connect Sink الرسائل من موضوعات Kafka ويكتبها إلى الجداول المناسبة. ويكتب ClickHouse Connect Sink البيانات إلى الجداول الموجودة مسبقًا. يُرجى التأكد من إنشاء جدول مستهدف ذي مخطط مناسب في ClickHouse قبل البدء في إدراج البيانات فيه.

يتطلب كل موضوع في Kafka جدولًا مستهدفًا مخصصًا في ClickHouse. يجب أن يتطابق اسم الجدول المستهدف مع اسم موضوع المصدر.

إذا كنت بحاجة إلى تحويل الرسائل الصادرة قبل إرسالها إلى ClickHouse Kafka Connect Sink، فاستخدم تحويلات Kafka Connect.

مع تعريف مخطط:

• (1) - لا يكون JSON مدعومًا إلا عندما تتضمن ClickHouse settings القيمة input_format_binary_read_json_as_string=1. ولا يعمل ذلك إلا مع عائلة تنسيقات RowBinary format، كما يؤثر هذا الإعداد في جميع الأعمدة ضمن طلب insert، لذا يجب أن تكون كلها من نوع سلسلة نصية. في هذه الحالة، سيحوّل موصل قيمة STRUCT إلى JSON string.

• (2) - عندما يحتوي struct على unions مثل oneof، يجب تهيئة converter بحيث لا يضيف prefix/suffix إلى أسماء الحقول. يتوفر الإعداد generate.index.for.unions=false لـ ProtobufConverter.

من دون تعريف مخطط:

يُحوَّل السجل إلى JSON ويُرسَل إلى ClickHouse كقيمة بتنسيق JSONEachRow.

فيما يلي بعض وصفات التهيئة الشائعة لمساعدتك على البدء بسرعة.

هذه أبسط تهيئة للبدء. وهي تفترض أنك تشغّل Kafka Connect في الوضع الموزّع، وأن لديك خادم ClickHouse يعمل على localhost:8443 مع تمكين SSL، وأن البيانات بتنسيق JSON بلا مخطط.

{
  "name": "clickhouse-connect",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    "tasks.max": "1",
    "consumer.override.max.poll.records": "5000",
    "consumer.override.max.partition.fetch.bytes": "5242880",
    "database": "default",
    "errors.retry.timeout": "60000",
    "exactlyOnce": "false",
    "hostname": "localhost",
    "port": "8443",
    "ssl": "true",
    "jdbcConnectionProperties": "?ssl=true&sslmode=strict",
    "username": "default",
    "password": "<PASSWORD>",
    "topics": "<TOPIC_NAME>",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",
    "clickhouseSettings": ""
  }
}

يمكن للموصل استهلاك البيانات من عدة موضوعات

{
  "name": "clickhouse-connect",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    ...
    "topics": "SAMPLE_TOPIC, ANOTHER_TOPIC, YET_ANOTHER_TOPIC",
    ...
  }
}

{
  "name": "clickhouse-connect",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    ...
    "errors.tolerance": "all",
    "errors.deadletterqueue.topic.name": "<DLQ_TOPIC>",
    "errors.deadletterqueue.context.headers.enable": "true",
  }
}

{
  "name": "clickhouse-connect",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    ...
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "<SCHEMA_REGISTRY_HOST>:<PORT>",
    "value.converter.schemas.enable": "true",
  }
}

تُحدَّد تعيين الأنواع أدناه بواسطة io.confluent.connect.avro.AvroConverter، وهو التنفيذ الرسمي لتسلسل/إلغاء تسلسل Avro في Kafka Connect. راجع وثائق Kafka Connect للحصول على معلومات متقدمة حول منطق التحويل.

✅: مدعوم

❌: غير مدعوم

️⚠️: مدعوم جزئيًا

ارجع إلى أنواع البيانات المدعومة للاطلاع على التعيين بين أنواع Kafka Connect وأنواع ClickHouse.

مخططات Avro التالية غير مدعومة في الموصّل:

• النوع المنطقي decimal المبني على fixed

{"name": "decimal_18_4", "type": "fixed", "size": 8, "logicalType": "decimal", "precision": 18, "scale": 4}

• اتحادات تقبل NULL

{"name": "mixed_union", "type": ["null", "string", "int"], "default": null}

• حقول اتحاد في السجلات

{
  "name": "record_union",
  "type": [
    {
      "type": "record",
      "name": "TypeA",
      "fields": [
        {
          "name": "label",
          "type": "string"
        }
      ]
    },
    {
      "type": "record",
      "name": "TypeB",
      "fields": [
        {
          "name": "count",
          "type": "int"
        }
      ]
    }
  ]
}

{
  "name": "clickhouse-connect",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    ...
    "value.converter": "io.confluent.connect.protobuf.ProtobufConverter",
    "value.converter.schema.registry.url": "<SCHEMA_REGISTRY_HOST>:<PORT>",
    "value.converter.schemas.enable": "true",
  }
}

يرجى ملاحظة ما يلي: إذا واجهت مشكلات بسبب فئات مفقودة، فليست كل البيئات تتضمن محوّل Protobuf، وقد تحتاج إلى إصدار بديل من ملف JAR يتضمن التبعيات.

يُحدَّد تعيين الأنواع أدناه بواسطة io.confluent.connect.protobuf.ProtobufConverter، وهو التنفيذ الرسمي لتسلسل/إلغاء تسلسل Protobuf في Kafka Connect. راجع وثائق Kafka Connect للحصول على معلومات متقدمة حول منطق التحويل.

✅: مدعوم

❌: غير مدعوم

️⚠️: مدعوم جزئيًا

ارجع إلى أنواع البيانات المدعومة للاطلاع على التعيين بين أنواع Kafka Connect وأنواع ClickHouse.

لا يدعم الموصّل ترجمة اتحادات في Protobuf (oneof) إلى النوع Variant في ClickHouse. وبدلًا من ذلك، أدرِج حقول oneof كحقول Nullable منفصلة في مخطط جدول ClickHouse لديك.

على سبيل المثال:

syntax = "proto3";

package com.clickhouse.kafka.connect.proto.test;

message StringIntUnion {
  oneof mixed {
    string mixed_string = 2;
    int32 mixed_int = 3;
  }
}

ويُترجم إلى تعريف جدول ClickHouse التالي:

CREATE TABLE IF NOT EXISTS `StringIntUnion`
(
    mixed_string Nullable(String),
    mixed_int Nullable(Int32)
) ENGINE = ...;

مخططات Protobuf التالية غير مدعومة في الموصل:

• بُنى اتحاد متعددة الرسائل (قبل الإصدار 26.1 من CH)

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 (راجع صفحة التوثيق هذه).

{
  "name": "clickhouse-connect",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    ...
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
  }
}

يدعم الموصّل محوّل String في تنسيقات ClickHouse المختلفة: JSON، وCSV، وTSV.

{
  "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.

مثال:

{
  "name": "clickhouse-connect",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    ...
    "exactlyOnce": "false",
    "bufferCount": "5000",
    "bufferFlushTime": "2000"
  }
}

يوفَّر التسجيل تلقائيًا بواسطة Kafka Connect Platform. يمكن تهيئة وجهة التسجيل وتنسيقه عبر ملف التهيئة الخاص بـ Kafka Connect.

إذا كنت تستخدم Confluent Platform، فيمكن عرض السجلات بتشغيل أمر من واجهة سطر الأوامر (CLI):

confluent local services connect log

لمزيد من التفاصيل، راجع الدليل العملي.

يبلّغ ClickHouse Kafka Connect عن مقاييس وقت التشغيل عبر Java Management Extensions (JMX). ويكون JMX مفعّلًا في Kafka Connector بشكل افتراضي.

يوفّر الموصل مقاييس مخصّصة عبر اسم MBean التالي:

com.clickhouse:type=ClickHouseKafkaConnector,name=SinkTask{id}

يعرض الموصّل مقاييس 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 ويعرض مقاييس لدورة حياة المهام وتتبع الأخطاء.

مقاييس حالة المهام:

• 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.

• عمليات الحذف غير مدعومة.
• يُورَّث حجم الدفعة من خصائص مستهلك 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.

يجلب Kafka Connect (الإطار) الرسائل من موضوعات Kafka في الخلفية، بشكل مستقل عن الموصل:

• fetch.min.bytes: الحد الأدنى لكمية البيانات قبل أن يمرّر الإطار القيم إلى الموصل (الافتراضي: 1 بايت)
• fetch.max.bytes: الحد الأقصى لكمية البيانات التي يمكن جلبها في طلب واحد (الافتراضي: 52428800 / 50 ميغابايت)
• fetch.max.wait.ms: الحد الأقصى لمدة الانتظار قبل إعادة البيانات إذا لم يتم استيفاء fetch.min.bytes (الافتراضي: 500 مللي ثانية)

يجلب الموصّل الرسائل من المخزن المؤقت الخاص بإطار العمل عبر عملية استطلاع:

• max.poll.records: الحد الأقصى لعدد السجلات المُعادة في عملية استطلاع واحدة (الافتراضي: 500)
• max.partition.fetch.bytes: الحد الأقصى لكمية البيانات لكل تقسيم (الافتراضي: 1048576 / 1 MB)

للحصول على أفضل أداء مع ClickHouse، احرص على استخدام دفعات أكبر:

# 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

مهم: تمثل إعدادات الجلب في Kafka Connect بيانات مضغوطة، بينما يتلقى ClickHouse بيانات غير مضغوطة. وازن بين هذه الإعدادات وفقًا لنسبة الضغط لديك.

المفاضلات:

• دفعات أكبر = أداء أفضل لإدخال البيانات إلى ClickHouse، وأجزاء أقل، وعبء إضافي أقل
• دفعات أكبر = استخدام أعلى للذاكرة، واحتمال زيادة زمن الانتقال من البداية إلى النهاية
• دفعات كبيرة جدًا = خطر حدوث timeout، أو أخطاء OutOfMemory، أو تجاوز max.poll.interval.ms

مزيد من التفاصيل: وثائق Confluent | وثائق Kafka

تُعد عمليات الإدراج غير المتزامنة ميزة فعّالة عندما يرسل الموصّل دفعات صغيرة نسبيًا، أو عندما تريد تحسين إدخال البيانات بشكل أكبر عبر إسناد مسؤولية التجميع إلى 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 insert إلى معلَمة التهيئة clickhouseSettings:

{
  "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: يرسل الموصل الإقرار فورًا بعد التخزين المؤقت. يوفّر أداءً أفضل، لكن قد تُفقد البيانات إذا تعطّل الخادم قبل تفريغها.

يمكنك ضبط سلوك flush في async insert بدقة:

"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 مع الإدراجات غير المتزامنة:

{
  "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.

زِد التوازي لتحسين معدل النقل:

"tasks.max": "4"

تعالج كل مهمة مجموعة فرعية من أقسام موضوع. المزيد من المهام = مزيد من التوازي، ولكن:

• الحد الأقصى الفعّال لعدد المهام = عدد أقسام موضوع
• تحتفظ كل مهمة باتصالها الخاص بـ ClickHouse
• المزيد من المهام = عبء إضافي أكبر واحتمال أعلى لحدوث تنازع على الموارد

التوصية: ابدأ بتعيين tasks.max ليعادل عدد أقسام موضوع، ثم اضبطه بناءً على مقاييس CPU ومعدل النقل.

افتراضيًا، يجمع الموصل الرسائل على دفعات لكل قسم. ولزيادة معدل النقل، يمكنك التجميع على دفعات عبر الأقسام:

"ignorePartitionsWhenBatching": "true"

** تحذير**: استخدم هذا فقط عندما يكون exactlyOnce=false. يمكن أن يحسّن هذا الإعداد معدل النقل من خلال إنشاء دفعات أكبر، لكنه يفقد ضمانات الترتيب داخل كل قسم.

إذا كان موصل لديك مُعدًّا للاشتراك في عدة موضوعات، وكنت تستخدم topic2TableMap لربط الـ موضوعات بالجداول، وكنت تواجه اختناقًا عند الإدراج يؤدي إلى تأخر المستهلك، ففكّر في إنشاء موصل مستقل لكل موضوع بدلًا من ذلك.

يرجع السبب الرئيسي في ذلك إلى أن الدُفعات تُدرَج حاليًا في كل جدول بشكل تسلسلي.

التوصية: عند التعامل مع عدة موضوعات كبيرة الحجم، قم بنشر instance واحدة من موصل لكل موضوع لتحقيق أقصى معدل نقل متوازٍ لعمليات الإدراج.

اختر محرّك جدول ClickHouse المناسب لحالة الاستخدام لديك:

• MergeTree: الأفضل لمعظم حالات الاستخدام، إذ يوازن بين أداء الاستعلام والإدراج
• ReplicatedMergeTree: مطلوب لتحقيق التوفّر العالي، ويضيف عبئًا إضافيًا بسبب النسخ المتماثل
• *MergeTree مع ORDER BY مضبوط على نحو صحيح: حسّنه بما يلائم أنماط الاستعلام لديك

الإعدادات التي ينبغي مراعاتها:

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

لإعدادات الإدراج على مستوى الموصّل:

"clickhouseSettings": "insert_quorum=2,insert_quorum_timeout=60000"

يحتفظ الموصّل باتصالات HTTP مع ClickHouse. اضبط قيم المهلة للشبكات ذات زمن الاستجابة المرتفع:

"jdbcConnectionProperties": "?socket_timeout=300000,connection_timeout=30000"

• socket_timeout (الافتراضي: 30000 مللي ثانية): الحد الأقصى لمدة عمليات القراءة
• connection_timeout (الافتراضي: 10000 مللي ثانية): الحد الأقصى للوقت اللازم لإنشاء الاتصال

ارفع هذه القيم إذا واجهت أخطاء انتهاء المهلة مع الدفعات الكبيرة.

عند استخدام العميل V2 ("client_version": "V2")، تُنسخ البيانات بين المقبس وذاكرة التطبيق عبر مخزن مؤقت يتحكم فيه client_network_buffer_size (بالبايت). الحجم الافتراضي هو 300000 بايت. يمكن أن يؤدي استخدام مخزن مؤقت أكبر إلى تحسين معدل النقل عبر الوصلات ذات زمن الاستجابة المرتفع أو النطاق الترددي العالي، لكنه يزيد من استخدام الذاكرة لكل مهمة اتصال.

تُضبط جميع خيارات العميل عبر jdbcConnectionProperties بدلاً من clickhouseSettings. على سبيل المثال:

"jdbcConnectionProperties": "?client_network_buffer_size=1000000"

إذا كنت بحاجة إلى ضبط عدة خصائص للعميل، فافصل بينها باستخدام &:

"jdbcConnectionProperties": "?ssl=true&client_network_buffer_size=1000000"

راقب هذه المقاييس الأساسية:

1. Consumer lag: استخدم أدوات مراقبة Kafka لتتبّع التأخر في كل تقسيم
2. مقاييس الموصل: راقب receivedRecords وrecordProcessingTime وtaskProcessingTime عبر JMX (راجع Monitoring)
3. مقاييس ClickHouse:
   • system.asynchronous_inserts: راقب استخدام المخزن المؤقت لعملية الإدراج غير المتزامنة
   • system.parts: راقب عدد الأجزاء لاكتشاف مشكلات الدمج
   • system.merges: راقب عمليات الدمج النشطة
   • system.events: تتبّع InsertedRows وInsertedBytes وFailedInsertQuery

مشكلات الأداء الشائعة:

1. ابدأ بالإعدادات الافتراضية، ثم قِس واضبط استنادًا إلى الأداء الفعلي
2. فضّل الدُفعات الأكبر: استهدف 10,000-100,000 صف لكل insert متى أمكن
3. استخدم عمليات الإدراج غير المتزامنة عند إرسال الكثير من الدُفعات الصغيرة أو عند ارتفاع التزامن
4. استخدم دائمًا wait_for_async_insert=1 مع ضمان المعالجة مرة واحدة فقط
5. وسّع أفقيًا: زِد tasks.max حتى يصل إلى عدد التقسيمات
6. موصل واحد لكل topic مرتفع الحجم لتحقيق أقصى إنتاجية
7. راقب باستمرار: تتبّع تأخر المستهلك، وعدد الأجزاء، ونشاط الدمج
8. اختبر جيدًا: اختبر دائمًا تغييرات التكوين تحت حمل واقعي قبل النشر في بيئة الإنتاج

فيما يلي مثال كامل مُحسَّن لمعدل نقل عالٍ:

{
  "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": "<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"
  }
}

هذا الإعداد:

• يعالج ما يصل إلى 10,000 سجل في كل عملية poll
• ينشئ batches عبر الأقسام للحصول على عمليات insert أكبر
• يستخدم عملية الإدراج غير المتزامنة مع buffer بسعة 16 ميغابايت
• يشغّل 8 tasks متوازية (طابقها مع عدد الأقسام لديك)
• مُحسَّن لتحقيق أعلى معدل النقل بدلًا من الالتزام الصارم بالترتيب

يحدث هذا عندما تكون الإزاحة المخزنة في KeeperMap مختلفة عن الإزاحة المخزنة في Kafka، ويحدث ذلك عادةً عند حذف موضوع أو تعديل الإزاحة يدويًا. لإصلاح ذلك، ستحتاج إلى حذف القيم القديمة المخزنة لذلك الموضوع + التقسيم المحددين:

-- 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 إلى إعدادات الموصل:

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 في حقل القيمة بشكل افتراضي، لكن يمكنك استخدام التحويل KeyToValue لنقل المفتاح إلى حقل القيمة (ضمن حقل جديد باسم _key):

transforms=keyToValue
transforms.keyToValue.type=com.clickhouse.kafka.connect.transforms.KeyToValue
transforms.keyToValue.field=_key