%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](https://docs.confluent.io/kafka-connectors/http/current/overview.html#http-sink-connector-for-cp):
> لا يقوم موصل HTTP Sink بتجميع الطلبات على دفعات للرسائل التي تختلف قيم ترويسات Kafka فيها.
1. تحقّق من أن سجلات Kafka لديك تستخدم المفتاح نفسه.
2. عند إضافة معلمات إلى عنوان URL الخاص بواجهة برمجة تطبيقات HTTP، قد ينتج عن كل سجل عنوان URL فريد. ولهذا السبب، يُعطَّل التجميع على دفعات عند استخدام معلمات URL إضافية.
#### 400 طلب غير صالح
##### CANNOT\_PARSE\_QUOTED\_STRING
إذا تعذّر HTTP Sink وظهرت الرسالة التالية عند إدراج كائن JSON في عمود `String`:
```response theme={null}
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](https://github.com/edenhill/kcat) لإدراج الرسائل في Kafka.
اتبع [هذه التعليمات](https://docs.confluent.io/cloud/current/cp-component/connect-cloud-config.html#set-up-a-local-connect-worker-with-cp-install) لإعداد Connect بما يناسب نوع التثبيت لديك، مع الانتباه إلى الفروق بين عنقود standalone وعنقود موزّع. وإذا كنت تستخدم Confluent Cloud، فالإعداد الموزّع هو المقصود هنا.
أهم معلَمة هي `http.api.url`. تتطلب [واجهة HTTP](/docs/ar/concepts/features/interfaces/http) في ClickHouse ترميز عبارة INSERT كمعلَمة ضمن عنوان URL. ويجب أن يتضمن ذلك التنسيق (`JSONEachRow` في هذه الحالة) وقاعدة البيانات الهدف. كما يجب أن يكون التنسيق متوافقًا مع بيانات Kafka، التي ستُحوَّل إلى سلسلة نصية ضمن حمولة HTTP. ويجب أيضًا إجراء URL escaping لهذه المعلَمات. يوضّح المثال أدناه هذا التنسيق لمجموعة بيانات Github (على افتراض أنك تشغّل ClickHouse محليًا):
```response theme={null}
://:?query=INSERT%20INTO%20.%20FORMAT%20JSONEachRow
http://localhost:8123?query=INSERT%20INTO%20default.github%20FORMAT%20JSONEachRow
```
تُعدّ المعلمات الإضافية التالية ذات صلة باستخدام HTTP Sink مع ClickHouse. يمكن العثور على قائمة كاملة بالمعلمات [هنا](https://docs.confluent.io/kafka-connect-http/current/connector_config.html):
* `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 واحد. تأكد من ضبطه على عدد كبير مناسب. ووفقًا [لتوصيات](/docs/ar/reference/statements/insert-into#performance-considerations) 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](/docs/ar/reference/formats/Avro/Avro) مدعوم أيضًا في ClickHouse عند استخدام converter io.confluent.connect.avro.AvroConverter.
يمكن العثور على قائمة كاملة بالإعدادات، بما في ذلك كيفية تكوين proxy، وretries، وSSL المتقدم، [هنا](https://docs.confluent.io/kafka-connect-http/current/connector_config.html).
يمكن العثور على example configuration files الخاصة ببيانات GitHub النموذجية [هنا](https://github.com/ClickHouse/clickhouse-docs/tree/main/docs/integrations/data-ingestion/kafka/code/connectors/http_sink)، على افتراض أن Connect يعمل في وضع standalone وأن Kafka مستضاف في Confluent Cloud.
تأكد من إنشاء الجدول. يُعرض أدناه مثال لمجموعة بيانات GitHub مصغّرة تستخدم محرك MergeTree القياسي.
```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)
```
أرسل رسائل إلى Kafka. نستخدم أدناه [kcat](https://github.com/edenhill/kcat) لإرسال 10 آلاف رسالة.
```bash theme={null}
head -n 10000 github_all_columns.ndjson | kcat -b : -X security.protocol=sasl_ssl -X sasl.mechanisms=PLAIN -X sasl.username= -X sasl.password= -t github
```
يجب أن يؤكد استعلام بسيط على الجدول الهدف "Github" إدراج البيانات.
```sql theme={null}
SELECT count() FROM default.github;
| count\(\) |
| :--- |
| 10000 |
```
