موصل Confluent HTTP Sink - ClickHouse Documentation

إن موصل HTTP Sink لا يعتمد على نوع بيانات محدد، لذا لا يحتاج إلى مخطط Kafka، كما يدعم أنواع بيانات خاصة بـ ClickHouse مثل Maps وArrays. وتأتي هذه المرونة الإضافية مقابل زيادة طفيفة في تعقيد الإعداد. فيما يلي نشرح عملية تثبيت بسيطة، مع سحب الرسائل من topic واحد في Kafka وإدراج صفوف في جدول ClickHouse.

يُوزَّع HTTP Connector بموجب Confluent Enterprise License.

خطوات البدء السريع

اجمع تفاصيل الاتصال الخاصة بك

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

المعلمات	الوصف
`HOST` and `PORT`	عادةً ما يكون المنفذ 8443 عند استخدام TLS، أو 8123 عند عدم استخدام TLS.
`DATABASE NAME`	افتراضيًا، توجد قاعدة بيانات باسم `default`. استخدم اسم قاعدة البيانات التي تريد الاتصال بها.
`USERNAME` and `PASSWORD`	افتراضيًا، يكون اسم المستخدم `default`. استخدم اسم المستخدم المناسب لحالة الاستخدام لديك.

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

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

تفاصيل اتصال HTTPS لخدمة ClickHouse Cloud

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

شغّل Kafka Connect وموصل HTTP sink

لديك خياران:

Self-managed: نزّل حزمة Confluent وثبّتها محليًا. اتبع تعليمات التثبيت الخاصة بتثبيت الموصل كما هو موثق هنا. إذا استخدمت طريقة التثبيت confluent-hub، فسيتم تحديث ملفات التكوين المحلية لديك.
Confluent Cloud: يتوفر إصدار مُدار بالكامل من HTTP Sink لمن يستخدمون Confluent Cloud لاستضافة Kafka لديهم. ويتطلب ذلك أن تكون بيئة ClickHouse الخاصة بك قابلة للوصول من Confluent Cloud.

تستخدم الأمثلة التالية Confluent Cloud.

أنشئ جدول الوجهة في ClickHouse

قبل اختبار الاتصال، لنبدأ بإنشاء جدول اختبار في ClickHouse Cloud. سيتلقى هذا الجدول البيانات من Kafka:

CREATE TABLE default.my_table
(
    `side` String,
    `quantity` Int32,
    `symbol` String,
    `price` Int32,
    `account` String,
    `userid` String
)
ORDER BY tuple()

تكوين HTTP Sink

أنشئ Kafka topic ومثيلاً من موصل HTTP Sink:
كوّن موصل HTTP Sink:

أدخل اسم الـ topic الذي أنشأته
المصادقة
- HTTP Url - عنوان URL الخاص بـ ClickHouse Cloud مع تحديد query من نوع INSERT بالشكل <protocol>://<clickhouse_host>:<clickhouse_port>?query=INSERT%20INTO%20<database>.<table>%20FORMAT%20JSONEachRow. ملاحظة: يجب ترميز الـ query.
- Endpoint Authentication type - BASIC
- Auth username - اسم مستخدم ClickHouse
- Auth password - كلمة مرور ClickHouse

هذا HTTP Url عرضة لوقوع الأخطاء. تأكد من أن التحويلات الهروبية دقيقة لتجنب المشكلات.

الإعدادات
- Input Kafka record value format - يعتمد ذلك على بيانات المصدر لديك، لكن في معظم الحالات يكون JSON أو Avro. نفترض استخدام JSON في الإعدادات التالية.
- في قسم advanced configurations:
  - HTTP Request Method - اضبطه على POST
  - Request Body Format - json
  - Batch batch size - وفقًا لتوصيات ClickHouse، اضبط هذا على 1000 على الأقل.
  - Batch json as array - true
  - Retry on HTTP codes - 400-500، ولكن عدّله حسب الحاجة. على سبيل المثال، قد يتغير ذلك إذا كان لديك HTTP proxy أمام ClickHouse.
  - Maximum Reties - القيمة الافتراضية (10) مناسبة، لكن لا تتردد في تعديلها للحصول على retries أكثر موثوقية.

اختبار الاتصال

أنشئ رسالة في topic الذي أعددته لـ HTTP Sink
وتحقق من أن الرسالة التي أُنشئت قد كُتبت إلى مثيل ClickHouse الخاص بك.

استكشاف الأخطاء وإصلاحها

لا يُجري HTTP Sink تجميعًا للرسائل على دفعات

وفقًا لـ وثائق Sink:

لا يقوم موصل HTTP Sink بتجميع الطلبات على دفعات للرسائل التي تختلف قيم ترويسات Kafka فيها.

تحقّق من أن سجلات Kafka لديك تستخدم المفتاح نفسه.
عند إضافة معلمات إلى عنوان URL الخاص بواجهة برمجة تطبيقات HTTP، قد ينتج عن كل سجل عنوان URL فريد. ولهذا السبب، يُعطَّل التجميع على دفعات عند استخدام معلمات URL إضافية.

400 طلب غير صالح

CANNOT_PARSE_QUOTED_STRING

إذا تعذّر HTTP Sink وظهرت الرسالة التالية عند إدراج كائن JSON في عمود String:

Code: 26. DB::ParsingException: Cannot parse JSON string: expected opening quote: (while reading the value of key key_name): While executing JSONEachRowRowInputFormat: (at row 1). (CANNOT_PARSE_QUOTED_STRING)

اضبط الإعداد input_format_json_read_objects_as_strings=1 في عنوان URL كسلسلة مُرمَّزة SETTINGS%20input_format_json_read_objects_as_strings%3D1

تحميل مجموعة بيانات GitHub (اختياري)

لاحظ أن هذا المثال يحافظ على الحقول من النوع Array في مجموعة بيانات GitHub. نفترض أن لديك topic فارغًا باسم github في الأمثلة، ونستخدم kcat لإدراج الرسائل في Kafka.

تحضير الإعداد

اتبع هذه التعليمات لإعداد Connect بما يناسب نوع التثبيت لديك، مع الانتباه إلى الفروق بين عنقود standalone وعنقود موزّع. وإذا كنت تستخدم Confluent Cloud، فالإعداد الموزّع هو المقصود هنا.أهم معلَمة هي http.api.url. تتطلب واجهة HTTP في ClickHouse ترميز عبارة INSERT كمعلَمة ضمن عنوان URL. ويجب أن يتضمن ذلك التنسيق (JSONEachRow في هذه الحالة) وقاعدة البيانات الهدف. كما يجب أن يكون التنسيق متوافقًا مع بيانات Kafka، التي ستُحوَّل إلى سلسلة نصية ضمن حمولة HTTP. ويجب أيضًا إجراء URL escaping لهذه المعلَمات. يوضّح المثال أدناه هذا التنسيق لمجموعة بيانات Github (على افتراض أنك تشغّل ClickHouse محليًا):

<protocol>://<clickhouse_host>:<clickhouse_port>?query=INSERT%20INTO%20<database>.<table>%20FORMAT%20JSONEachRow

http://localhost:8123?query=INSERT%20INTO%20default.github%20FORMAT%20JSONEachRow

تُعدّ المعلمات الإضافية التالية ذات صلة باستخدام HTTP Sink مع ClickHouse. يمكن العثور على قائمة كاملة بالمعلمات هنا:

request.method - اضبطها على POST
retry.on.status.codes - اضبطها على 400-500 لإعادة المحاولة عند أي رموز خطأ. حسّن هذا الإعداد بناءً على الأخطاء المتوقعة في البيانات.
request.body.format - في معظم الحالات، ستكون هذه القيمة JSON.
auth.type - اضبطها على BASIC إذا كنت تستخدم authentication مع ClickHouse. لا تتوفر حاليًا آليات authentication الأخرى المتوافقة مع ClickHouse.
ssl.enabled - اضبطها على true إذا كنت تستخدم SSL.
connection.user - اسم المستخدم لـ ClickHouse.
connection.password - كلمة المرور لـ ClickHouse.
batch.max.size - عدد الصفوف التي سيتم إرسالها في Batch واحد. تأكد من ضبطه على عدد كبير مناسب. ووفقًا لتوصيات ClickHouse، ينبغي اعتبار القيمة 1000 حدًا أدنى.
tasks.max - يتيح connector الخاص بـ HTTP Sink تشغيل task واحدة أو أكثر. ويمكن استخدام ذلك لتحسين الأداء. وإلى جانب حجم Batch، يُعد هذا الوسيلة الأساسية لتحسين الأداء.
key.converter - اضبطها وفقًا لـ types الخاصة بالمفاتيح لديك.
value.converter - اضبطها بناءً على نوع البيانات الموجودة في topic لديك. لا تحتاج هذه البيانات إلى schema. يجب أن يكون format هنا متسقًا مع FORMAT المحدد في المعلمة http.api.url. أبسط خيار هنا هو استخدام JSON وconverter ‏org.apache.kafka.connect.json.JsonConverter. ويمكن أيضًا التعامل مع value على أنها string عبر converter ‏org.apache.kafka.connect.storage.StringConverter، رغم أن ذلك سيتطلب من المستخدم استخراج value في statement الخاص بـ insert باستخدام functions. كما أن Avro format مدعوم أيضًا في ClickHouse عند استخدام converter ‏io.confluent.connect.avro.AvroConverter.

يمكن العثور على قائمة كاملة بالإعدادات، بما في ذلك كيفية تكوين proxy، وretries، وSSL المتقدم، هنا.يمكن العثور على example configuration files الخاصة ببيانات GitHub النموذجية هنا، على افتراض أن Connect يعمل في وضع standalone وأن Kafka مستضاف في Confluent Cloud.

إنشاء جدول ClickHouse

تأكد من إنشاء الجدول. يُعرض أدناه مثال لمجموعة بيانات GitHub مصغّرة تستخدم محرك MergeTree القياسي.

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)

أضف البيانات إلى Kafka

أرسل رسائل إلى Kafka. نستخدم أدناه kcat لإرسال 10 آلاف رسالة.

head -n 10000 github_all_columns.ndjson | kcat -b <host>:<port> -X security.protocol=sasl_ssl -X sasl.mechanisms=PLAIN -X sasl.username=<username>  -X sasl.password=<password> -t github

يجب أن يؤكد استعلام بسيط على الجدول الهدف “Github” إدراج البيانات.

SELECT count() FROM default.github;

| count\(\) |
| :--- |
| 10000 |

​خطوات البدء السريع

​استكشاف الأخطاء وإصلاحها

​لا يُجري HTTP Sink تجميعًا للرسائل على دفعات

​400 طلب غير صالح

CANNOT_PARSE_QUOTED_STRING

​تحميل مجموعة بيانات GitHub (اختياري)

خطوات البدء السريع

استكشاف الأخطاء وإصلاحها

لا يُجري HTTP Sink تجميعًا للرسائل على دفعات

400 طلب غير صالح

تحميل مجموعة بيانات GitHub (اختياري)