> ## Documentation Index
> Fetch the complete documentation index at: https://private-7c7dfe99-mintlify-67bc7bf8.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

> استخدام موصل HTTP Sink مع Kafka Connect وClickHouse

# موصل Confluent HTTP Sink

export const Image = ({img, alt, size}) => {
  return <Frame>
      <img src={img} alt={alt} />
    </Frame>;
};

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

فيما يلي نشرح عملية تثبيت بسيطة، مع سحب الرسائل من topic واحد في Kafka وإدراج صفوف في جدول ClickHouse.

<Note>
  يُوزَّع HTTP Connector بموجب [Confluent Enterprise License](https://docs.confluent.io/kafka-connect-http/current/overview.html#license).
</Note>

<div id="quick-start-steps">
  ### خطوات البدء السريع
</div>

<div id="1-gather-your-connection-details">
  #### 1. اجمع بيانات الاتصال الخاصة بك
</div>

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

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

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

<Image img="https://mintcdn.com/private-7c7dfe99-mintlify-67bc7bf8/bY1iopO46snwcr5S/images/_snippets/cloud-connect-button.png?fit=max&auto=format&n=bY1iopO46snwcr5S&q=85&s=fda686b9b9454ad21df18bb7bac60ff5" size="md" alt="زر Connect لخدمة ClickHouse Cloud" border width="998" height="932" data-path="images/_snippets/cloud-connect-button.png" />

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

<Image img="https://mintcdn.com/private-7c7dfe99-mintlify-67bc7bf8/bY1iopO46snwcr5S/images/_snippets/connection-details-https.png?fit=max&auto=format&n=bY1iopO46snwcr5S&q=85&s=a0ea22430cbcef486eb6f9f596b18a46" size="md" alt="تفاصيل اتصال HTTPS لخدمة ClickHouse Cloud" border width="1320" height="1184" data-path="images/_snippets/connection-details-https.png" />

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

<div id="2-run-kafka-connect-and-the-http-sink-connector">
  #### 2. شغّل Kafka Connect وموصل HTTP Sink
</div>

لديك خياران:

* **مُدار ذاتيًا:** نزّل حزمة Confluent وثبّتها محليًا. اتبع تعليمات التثبيت الخاصة بالموصل كما هو موثّق [هنا](https://docs.confluent.io/kafka-connect-http/current/overview.html).
  إذا استخدمت طريقة التثبيت `confluent-hub`، فسيتم تحديث ملفات الإعداد المحلية لديك.

* **Confluent Cloud:** يتوفر إصدار مُدار بالكامل من HTTP Sink لمن يستخدمون Confluent Cloud لاستضافة Kafka. يتطلب ذلك أن تكون بيئة ClickHouse لديك قابلة للوصول من Confluent Cloud.

<Note>
  تستخدم الأمثلة التالية Confluent Cloud.
</Note>

<div id="3-create-destination-table-in-clickhouse">
  #### 3. إنشاء جدول الوجهة في ClickHouse
</div>

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

```sql theme={null}
CREATE TABLE default.my_table
(
    `side` String,
    `quantity` Int32,
    `symbol` String,
    `price` Int32,
    `account` String,
    `userid` String
)
ORDER BY tuple()
```

<div id="4-configure-http-sink">
  #### 4. تكوين HTTP Sink
</div>

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

<Image img="https://mintcdn.com/private-7c7dfe99-mintlify-67bc7bf8/ms4E-Ni592HZjtjR/images/integrations/data-ingestion/kafka/confluent/create_http_sink.png?fit=max&auto=format&n=ms4E-Ni592HZjtjR&q=85&s=3c8a99e84af06fb1067780e11de74904" size="sm" alt="واجهة Confluent Cloud تعرض كيفية إنشاء موصل HTTP Sink" border width="718" height="736" data-path="images/integrations/data-ingestion/kafka/confluent/create_http_sink.png" />

<br />

كوّن موصل 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

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

<Image img="https://mintcdn.com/private-7c7dfe99-mintlify-67bc7bf8/ms4E-Ni592HZjtjR/images/integrations/data-ingestion/kafka/confluent/http_auth.png?fit=max&auto=format&n=ms4E-Ni592HZjtjR&q=85&s=5269df8ce2fd188238ba1eafff7be798" size="lg" alt="واجهة Confluent Cloud تعرض إعدادات المصادقة لموصل HTTP Sink" border width="1944" height="878" data-path="images/integrations/data-ingestion/kafka/confluent/http_auth.png" />

<br />

* الإعدادات
  * `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 أكثر موثوقية.

<Image img="https://mintcdn.com/private-7c7dfe99-mintlify-67bc7bf8/ms4E-Ni592HZjtjR/images/integrations/data-ingestion/kafka/confluent/http_advanced.png?fit=max&auto=format&n=ms4E-Ni592HZjtjR&q=85&s=d569c5a1141227c703c439fb9ac727d4" size="sm" alt="واجهة Confluent Cloud تعرض خيارات التكوين المتقدمة لموصل HTTP Sink" border width="786" height="796" data-path="images/integrations/data-ingestion/kafka/confluent/http_advanced.png" />

<div id="5-testing-the-connectivity">
  #### 5. اختبار الاتصال
</div>

أنشئ رسالة في topic تم ضبطه في HTTP Sink لديك

<Image img="https://mintcdn.com/private-7c7dfe99-mintlify-67bc7bf8/ms4E-Ni592HZjtjR/images/integrations/data-ingestion/kafka/confluent/create_message_in_topic.png?fit=max&auto=format&n=ms4E-Ni592HZjtjR&q=85&s=bf6264ef50b3c4d668a8749509e663e9" size="md" alt="واجهة Confluent Cloud تعرض كيفية إنشاء رسالة اختبار في topic Kafka" border width="2138" height="846" data-path="images/integrations/data-ingestion/kafka/confluent/create_message_in_topic.png" />

<br />

وتحقق من أن الرسالة التي أنشأتها قد تمت كتابتها في مثيل ClickHouse لديك.

<div id="troubleshooting">
  ### استكشاف الأخطاء وإصلاحها
</div>

<div id="http-sink-doesnt-batch-messages">
  #### لا يُجري HTTP Sink تجميعًا للرسائل على دفعات
</div>

وفقًا لـ [وثائق 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 إضافية.

<div id="400-bad-request">
  #### 400 طلب غير صالح
</div>

<div id="cannot_parse_quoted_string">
  ##### CANNOT\_PARSE\_QUOTED\_STRING
</div>

إذا تعذّر 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`

<div id="load-the-github-dataset-optional">
  ### تحميل مجموعة بيانات GitHub (اختياري)
</div>

لاحظ أن هذا المثال يحافظ على الحقول من النوع Array في مجموعة بيانات GitHub. نفترض أن لديك topic فارغًا باسم `github` في الأمثلة، ونستخدم [kcat](https://github.com/edenhill/kcat) لإدراج الرسائل في Kafka.

<div id="1-prepare-configuration">
  ##### 1. تحضير الإعداد
</div>

اتبع [هذه التعليمات](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](/ar/concepts/features/interfaces/http) في ClickHouse ترميز عبارة INSERT كمعلَمة ضمن عنوان URL. ويجب أن يتضمن ذلك التنسيق (`JSONEachRow` في هذه الحالة) وقاعدة البيانات الهدف. كما يجب أن يكون التنسيق متوافقًا مع بيانات Kafka، التي ستُحوَّل إلى سلسلة نصية ضمن حمولة HTTP. ويجب أيضًا إجراء URL escaping لهذه المعلَمات. يوضّح المثال أدناه هذا التنسيق لمجموعة بيانات Github (على افتراض أنك تشغّل ClickHouse محليًا):

```response theme={null}
<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. يمكن العثور على قائمة كاملة بالمعلمات [هنا](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 واحد. تأكد من ضبطه على عدد كبير مناسب. ووفقًا [لتوصيات](/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](/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.

<div id="2-create-the-clickhouse-table">
  ##### 2. أنشئ جدول ClickHouse
</div>

تأكد من إنشاء الجدول. يرد أدناه مثال على الحد الأدنى من مجموعة بيانات 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)

```

<div id="3-add-data-to-kafka">
  ##### 3. إضافة بيانات إلى Kafka
</div>

أدخِل الرسائل إلى Kafka. نستخدم أدناه [kcat](https://github.com/edenhill/kcat) لإدخال 10 آلاف رسالة.

```bash theme={null}
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" أنه تم إدراج البيانات.

```sql theme={null}
SELECT count() FROM default.github;

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

```