موصل Confluent HTTP Sink

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

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

• أدخل اسم الـ 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

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

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

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

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)

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

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

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

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)

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

SELECT count() FROM default.github;

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