> ## Documentation Index > Fetch the complete documentation index at: https://clickhouse.com/docs/llms.txt > Use this file to discover all available pages before exploring further. # Comment ingérer des données depuis Kafka dans ClickHouse > Apprenez à ingérer des données depuis un topic Kafka dans ClickHouse à l’aide du moteur de table Kafka, de vues matérialisées et de tables MergeTree. {frontMatter.description}
## Vue d’ensemble
Cet article décrit le processus d’envoi de données d’un topic Kafka vers une table ClickHouse. Nous utiliserons le flux des modifications récentes de Wikimedia, qui fournit un [flux d’événements](https://stream.wikimedia.org/v2/stream/recentchange) représentant les changements apportés à différents projets Wikimedia. Les étapes sont les suivantes : 1. Comment configurer Kafka sur Ubuntu 2. Ingérer un flux de données dans un topic Kafka 3. Créer une table ClickHouse qui s’abonne au topic
# 1. Configurer Kafka sur Ubuntu
1. Créez une instance **ec2** sous Ubuntu et connectez-vous-y en SSH : ```bash theme={null} ssh -i ~/training.pem ubuntu@ec2.compute.amazonaws.com ``` 2. Installez Kafka (en suivant les instructions ici : [https://www.linode.com/docs/guides/how-to-install-apache-kafka-on-ubuntu/](https://www.linode.com/docs/guides/how-to-install-apache-kafka-on-ubuntu/)) : ```bash theme={null} 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. Démarrez ZooKeeper : ```bash theme={null} cd kafka_2.13-3.7.0 bin/zookeeper-server-start.sh config/zookeeper.properties ``` 4. Ouvrez un nouveau terminal et lancez Kafka : ```bash theme={null} ssh -i ~/training.pem ubuntu@ec2.compute.amazonaws.com cd kafka/kafka_2.13-3.7.0/ bin/kafka-server-start.sh config/server.properties ``` 5. Ouvrez un troisième terminal et créez un topic nommé wikimedia : ```bash theme={null} ssh -i ~/training.pem ubuntu@ec2.compute.amazonaws.com cd kafka/kafka_2.13-3.7.0/ bin/kafka-topics.sh --create --topic wikimedia --bootstrap-server localhost:9092 ``` 6. Vous pouvez vérifier qu’il a bien été créé en : ```bash theme={null} bin/kafka-topics.sh --list --bootstrap-server localhost:9092 ```
# 2. Ingérer le flux Wikimedia dans Kafka
1. Nous avons d’abord besoin de quelques utilitaires : ```bash theme={null} sudo apt-get install librdkafka-dev libyajl-dev sudo apt-get install kafkacat ``` 2. Les données sont envoyées à Kafka à l’aide d’une commande **curl** astucieuse qui récupère les derniers événements Wikimedia, en extrait les données JSON, puis les envoie au topic Kafka : ```bash theme={null} curl -N https://stream.wikimedia.org/v2/stream/recentchange | awk '/^data: /{gsub(/^data: /, ""); print}' | kafkacat -P -b localhost:9092 -t wikimedia ``` 3. Vous pouvez « décrire » le topic : ```bash theme={null} bin/kafka-topics.sh --describe --topic wikimedia --bootstrap-server localhost:9092 ``` 4. Vérifions que tout fonctionne en consommant quelques événements : ```bash theme={null} bin/kafka-console-consumer.sh --topic wikimedia --from-beginning --bootstrap-server localhost:9092 ``` 5. Appuyez sur **Ctrl+c** pour interrompre la commande précédente.
# 3. Ingérez les données dans ClickHouse
1. Voici à quoi ressemblent les données reçues : ```json theme={null} { "$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": "Added qualifier: series ordinal (P1545): 20, Modify PubMed ID: 7292984 citation data from NCBI, Europe PMC and CrossRef" } ``` 2. Nous aurons besoin du **moteur de table** Kafka pour récupérer les données du topic Kafka : ```sql theme={null} 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. Pour une raison quelconque, le **moteur de table** **Kafka** semble prendre l’URL **ec2** publique et la convertir en nom DNS privé. J’ai donc dû l’ajouter à mon fichier local `/etc/hosts` : ```bash theme={null} 52.14.154.92 ip.us-east-2.compute.internal ``` 4. Vous pouvez lire depuis une table Kafka ; il vous suffit d’activer un paramètre : ```sql theme={null} SELECT * FROM wikiQueue LIMIT 20 FORMAT Vertical SETTINGS stream_like_engine_allow_direct_select = 1; ``` Les lignes devraient s’afficher correctement, selon les colonnes définies dans la table **wikiQueue** : ```response theme={null} 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. Il nous faut une table **MergeTree** pour stocker ces événements entrants : ```sql theme={null} 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. Définissons une vue matérialisée qui se déclenche lorsqu’une insertion a lieu sur la table **Kafka** et envoie les données vers notre table **rawEvents** : ```sql theme={null} 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. Vous devriez commencer à voir des données arriver dans **rawEvents** presque immédiatement : ```sql theme={null} SELECT count() FROM rawEvents; ``` 8. Affichons quelques lignes : ```sql theme={null} SELECT * FROM rawEvents LIMIT 5 FORMAT Vertical ``` ```response theme={null} 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. Voyons quels types d’événements arrivent : ``` SELECT type, count() FROM rawEvents GROUP BY type ``` ```response theme={null} ┌─type───────┬─count()─┐ 1. │ 142 │ 1 │ 2. │ new │ 1003 │ 3. │ categorize │ 12228 │ 4. │ log │ 1799 │ 5. │ edit │ 17142 │ └────────────┴─────────┘ ``` Définissons une vue matérialisée enchaînée à notre vue matérialisée actuelle. Nous y suivrons quelques statistiques agrégées par minute : ```sql theme={null} 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. Il nous faudra des fonctions **-Merge** pour afficher les résultats : ```sql theme={null} 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; ```