كيفية إدخال البيانات من Kafka إلى ClickHouse

تعرّف على كيفية إدخال البيانات من موضوع في Kafka إلى ClickHouse باستخدام محرك الجدول Kafka وعروض مادية وMergeTree tables.

​

نظرة عامة

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

1. كيفية إعداد Kafka على Ubuntu
2. إدخال دفق من البيانات إلى موضوع في Kafka
3. إنشاء جدول ClickHouse يشترك في الموضوع

​

1. إعداد Kafka على Ubuntu

1. أنشئ مثيل Ubuntu على ec2 واتصل به عبر SSH:

ssh -i ~/training.pem [e-mail]

2. ثبّت Kafka (باتباع الإرشادات هنا: https://www.linode.com/docs/guides/how-to-install-apache-kafka-on-ubuntu/):

sudo apt update
sudo apt install openjdk-11-jdk

mkdir /home/ubuntu/kafka
cd /home/ubuntu/kafka/

wget https://downloads.apache.org/kafka/3.7.0/kafka_2.13-3.7.0.tgz

tar -zxvf kafka_2.13-3.7.0.tgz

3. شغّل ZooKeeper:

cd kafka_2.13-3.7.0
bin/zookeeper-server-start.sh config/zookeeper.properties

4. افتح نافذة طرفية جديدة وشغّل Kafka:

ssh -i ~/training.pem [e-mail]
cd kafka/kafka_2.13-3.7.0/
bin/kafka-server-start.sh config/server.properties

5. افتح نافذة طرفية ثالثة وأنشئ موضوع باسم wikimedia:

ssh -i ~/training.pem [e-mail]
cd kafka/kafka_2.13-3.7.0/

bin/kafka-topics.sh --create --topic wikimedia --bootstrap-server localhost:9092

6. يمكنك التأكد من إنشائه بنجاح من خلال:

bin/kafka-topics.sh --list --bootstrap-server localhost:9092

​

2. إدخال تدفّق Wikimedia إلى Kafka

1. نحتاج أولًا إلى بعض الأدوات:

sudo apt-get install librdkafka-dev libyajl-dev
sudo apt-get install kafkacat

2. تُرسَل البيانات إلى Kafka باستخدام أمر curl ذكي يجلب أحدث أحداث Wikimedia، ويستخلص منها بيانات JSON ثم يرسلها إلى موضوع Kafka:

curl -N https://stream.wikimedia.org/v2/stream/recentchange  | awk '/^data: /{gsub(/^data: /, ""); print}' | kafkacat -P -b localhost:9092 -t wikimedia

3. يمكنك “وصف” الـ موضوع:

bin/kafka-topics.sh --describe --topic wikimedia --bootstrap-server localhost:9092

4. لنتحقق من أن كل شيء يعمل كما ينبغي من خلال استهلاك بعض الأحداث:

bin/kafka-console-consumer.sh --topic wikimedia --from-beginning --bootstrap-server localhost:9092

5. اضغط على Ctrl+c لإيقاف الأمر السابق.

​

3. إدخال البيانات إلى ClickHouse

1. إليك شكل البيانات الواردة:

{
	"$schema": "/mediawiki/recentchange/1.0.0",
	"meta": {
		"uri": "https://www.wikidata.org/wiki/Q45791749",
		"request_id": "f64cfb17-04ba-4d09-8935-38ec6f0001c2",
		"id": "9d7d2b5a-b79b-45ea-b72c-69c3b69ae931",
		"dt": "2024-04-18T13:21:21Z",
		"domain": "www.wikidata.org",
		"stream": "mediawiki.recentchange",
		"topic": "eqiad.mediawiki.recentchange",
		"partition": 0,
		"offset": 5032636513
	},
	"id": 2196113017,
	"type": "edit",
	"namespace": 0,
	"title": "Q45791749",
	"title_url": "https://www.wikidata.org/wiki/Q45791749",
	"comment": "/* wbsetqualifier-add:1| */ [[Property:P1545]]: 20, Modify PubMed ID: 7292984 citation data from NCBI, Europe PMC and CrossRef",
	"timestamp": 1713446481,
	"user": "Cewbot",
	"bot": true,
	"notify_url": "https://www.wikidata.org/w/index.php?diff=2131981357&oldid=2131981341&rcid=2196113017",
	"minor": false,
	"patrolled": true,
	"length": {
		"old": 75618,
		"new": 75896
	},
	"revision": {
		"old": 2131981341,
		"new": 2131981357
	},
	"server_url": "https://www.wikidata.org",
	"server_name": "www.wikidata.org",
	"server_script_path": "/w",
	"wiki": "wikidatawiki",
	"parsedcomment": "<span dir=\"auto\"><span class=\"autocomment\">Added qualifier: </span></span> <a href=\"/wiki/Property:P1545\" title=\"series ordinal | position of an item in its parent series (most frequently a 1-based index), generally to be used as a qualifier (different from &quot;rank&quot; defined as a class, and from &quot;ranking&quot; defined as a property for evaluating a quality).\"><span class=\"wb-itemlink\"><span class=\"wb-itemlink-label\" lang=\"en\" dir=\"ltr\">series ordinal</span> <span class=\"wb-itemlink-id\">(P1545)</span></span></a>: 20, Modify PubMed ID: 7292984 citation data from NCBI, Europe PMC and CrossRef"
}

2. سنحتاج إلى محرك الجدول Kafka لسحب البيانات من موضوع Kafka:

CREATE OR REPLACE TABLE wikiQueue
(
    `id` UInt32,
    `type` String,
    `title` String,
    `title_url` String,
    `comment` String,
    `timestamp` UInt64,
    `user` String,
    `bot` Bool,
    `server_url` String,
    `server_name` String,
    `wiki` String,
    `meta` Tuple(uri String, id String, stream String, topic String, domain String)
)
ENGINE = Kafka(
   'ec2.compute.amazonaws.com:9092',
   'wikimedia',
   'consumer-group-wiki',
   'JSONEachRow'
);

3. لسببٍ ما، يبدو أن محرك الجدول Kafka يأخذ عنوان URL العام لـ ec2 ويحوّله إلى اسم DNS الخاص، لذا اضطررت إلى إضافته إلى ملف /etc/hosts المحلي:

52.14.154.92  ip.us-east-2.compute.internal

4. يمكنك القراءة من جدول Kafka، ما عليك سوى تمكين إعداد:

SELECT *
FROM wikiQueue
LIMIT 20
FORMAT Vertical
SETTINGS stream_like_engine_allow_direct_select = 1;

يُفترض أن تظهر الصفوف مُحلَّلةً بشكل صحيح وفقًا للأعمدة المعرّفة في جدول wikiQueue:

id:          2473996741
type:        edit
title:       File:Père-Lachaise - Division 6 - Cassereau 05.jpg
title_url:   https://commons.wikimedia.org/wiki/File:P%C3%A8re-Lachaise_-_Division_6_-_Cassereau_05.jpg
comment:     /* wbcreateclaim-create:1| */ [[d:Special:EntityPage/P921]]: [[d:Special:EntityPage/Q112327116]], [[:toollabs:quickstatements/#/batch/228454|batch #228454]]
timestamp:   1713457283
user:        Ameisenigel
bot:         false
server_url:  https://commons.wikimedia.org
server_name: commons.wikimedia.org
wiki:        commonswiki
meta:        ('https://commons.wikimedia.org/wiki/File:P%C3%A8re-Lachaise_-_Division_6_-_Cassereau_05.jpg','01a832e2-24c5-4ccb-bd93-8e2c0e429418','mediawiki.recentchange','eqiad.mediawiki.recentchange','commons.wikimedia.org')

5. نحتاج إلى جدول MergeTree لتخزين هذه الأحداث الواردة:

CREATE TABLE rawEvents (
    id UInt64,
    type LowCardinality(String),
    comment String,
    timestamp DateTime64(3, 'UTC'),
    title_url String,
    topic LowCardinality(String),
    user String
)
ENGINE = MergeTree
ORDER BY (type, timestamp);

6. لنعرّف عرضًا ماديًا يُفعَّل عند تنفيذ عملية insert على جدول Kafka ويرسل البيانات إلى جدول rawEvents:

CREATE MATERIALIZED VIEW rawEvents_mv TO rawEvents
AS
   SELECT
       id,
       type,
       comment,
       toDateTime(timestamp) AS timestamp,
       title_url,
       tupleElement(meta, 'topic') AS topic,
       user
FROM wikiQueue
WHERE title_url <> '';

7. ينبغي أن تبدأ بمشاهدة البيانات تتدفق إلى rawEvents على الفور تقريبًا:

SELECT count()
FROM rawEvents;

8. لنطّلع على بعض الصفوف:

SELECT *
FROM rawEvents
LIMIT 5
FORMAT Vertical

Row 1:
──────
id:        124842852
type:      142
comment:   Pere prlpz commented on "Plantilles Enciclopèdia Catalana" (Diria que no cal fer res als articles. Es pot actualitzar els enllaços que es facin servir a les referències (tot i que l'antic encara ha...)
timestamp: 2024-04-18 16:22:29.000
title_url: https://ca.wikipedia.org/wiki/Tema:Wu36d6vfsiuu4jsi
topic:     eqiad.mediawiki.recentchange
user:      Pere prlpz

Row 2:
──────
id:        2473996748
type:      categorize
comment:   [[:File:Ruïne van een poortgebouw, RP-T-1976-29-6(R).jpg]] removed from category
timestamp: 2024-04-18 16:21:20.000
title_url: https://commons.wikimedia.org/wiki/Category:Pieter_Moninckx
topic:     eqiad.mediawiki.recentchange
user:      Warburg1866

Row 3:
──────
id:        311828596
type:      categorize
comment:   [[:Cujo (película)]] añadida a la categoría
timestamp: 2024-04-18 16:21:21.000
title_url: https://es.wikipedia.org/wiki/Categor%C3%ADa:Pel%C3%ADculas_basadas_en_obras_de_Stephen_King
topic:     eqiad.mediawiki.recentchange
user:      Beta15

Row 4:
──────
id:        311828597
type:      categorize
comment:   [[:Cujo (película)]] eliminada de la categoría
timestamp: 2024-04-18 16:21:21.000
title_url: https://es.wikipedia.org/wiki/Categor%C3%ADa:Trabajos_basados_en_obras_de_Stephen_King
topic:     eqiad.mediawiki.recentchange
user:      Beta15

Row 5:
──────
id:        48494536
type:      categorize
comment:   [[:braiteremmo]] ajoutée à la catégorie
timestamp: 2024-04-18 16:21:21.000
title_url: https://fr.wiktionary.org/wiki/Cat%C3%A9gorie:Wiktionnaire:Exemples_manquants_en_italien
topic:     eqiad.mediawiki.recentchange
user:      Àncilu bot

9. لنرَ أنواع الأحداث التي تصل:

SELECT
    type,
    count()
FROM rawEvents
GROUP BY type

   ┌─type───────┬─count()─┐
1. │ 142        │       1 │
2. │ new        │    1003 │
3. │ categorize │   12228 │
4. │ log        │    1799 │
5. │ edit       │   17142 │
   └────────────┴─────────┘

لنعرّف عرضًا ماديًا آخر متسلسلًا مع عرضنا المادي الحالي. سنحتفظ ببعض الإحصاءات المجمّعة لكل دقيقة:

CREATE TABLE byMinute
(
    `dateTime` DateTime64(3, 'UTC') NOT NULL,
    `users` AggregateFunction(uniq, String),
    `pages` AggregateFunction(uniq, String),
    `updates` AggregateFunction(sum, UInt32)
)
ENGINE = AggregatingMergeTree
ORDER BY dateTime;

CREATE MATERIALIZED VIEW byMinute_mv TO byMinute
AS SELECT
    toStartOfMinute(timestamp) AS dateTime,
    uniqState(user) AS users,
    uniqState(title_url) AS pages,
    sumState(toUInt32(1)) AS updates
FROM rawEvents
GROUP BY dateTime;

9. سنحتاج إلى دوال -Merge لعرض النتائج:

SELECT
    dateTime AS dateTime,
    uniqMerge(users) AS users,
    uniqMerge(pages) AS pages,
    sumMerge(updates) AS updates
FROM byMinute
GROUP BY dateTime
ORDER BY dateTime DESC
LIMIT 10;