> ## 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. # Cómo ingestar datos de Kafka en ClickHouse > Aprende a ingestar datos de un topic de Kafka en ClickHouse mediante el motor de tabla Kafka, vistas materializadas y tablas MergeTree. {frontMatter.description}
## Descripción general
Este artículo explica el proceso de enviar datos desde un topic de Kafka a una tabla de ClickHouse. Usaremos el feed de cambios recientes de Wiki, que proporciona un [flujo de eventos](https://stream.wikimedia.org/v2/stream/recentchange) que representa los cambios realizados en varias propiedades de Wikimedia. Los pasos incluyen: 1. Cómo configurar Kafka en Ubuntu 2. Ingestar un flujo de datos en un topic de Kafka 3. Crear una tabla de ClickHouse que se suscriba al topic
# 1. Configurar Kafka en Ubuntu
1. Cree una instancia **ec2** de Ubuntu y conéctese a ella por SSH: ```bash theme={null} ssh -i ~/training.pem ubuntu@ec2.compute.amazonaws.com ``` 2. Instala Kafka (siguiendo las instrucciones de aquí: [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. Inicie ZooKeeper: ```bash theme={null} cd kafka_2.13-3.7.0 bin/zookeeper-server-start.sh config/zookeeper.properties ``` 4. Abre una nueva terminal e inicia 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. Abra una tercera consola y cree un topic llamado 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. Puedes verificar que se haya creado correctamente de la siguiente manera: ```bash theme={null} bin/kafka-topics.sh --list --bootstrap-server localhost:9092 ```
# 2. Ingestar el flujo de Wikimedia a Kafka
1. Primero necesitamos algunas utilidades: ```bash theme={null} sudo apt-get install librdkafka-dev libyajl-dev sudo apt-get install kafkacat ``` 2. Los datos se envían a Kafka mediante un ingenioso comando **curl** que obtiene los eventos más recientes de Wikimedia, extrae los datos JSON y los envía al topic de 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. Puede "describir" el topic: ```bash theme={null} bin/kafka-topics.sh --describe --topic wikimedia --bootstrap-server localhost:9092 ``` 4. Verifiquemos que todo funcione correctamente consumiendo algunos eventos: ```bash theme={null} bin/kafka-console-consumer.sh --topic wikimedia --from-beginning --bootstrap-server localhost:9092 ``` 5. Pulsa **Ctrl+c** para detener el comando anterior.
# 3. Ingestar los datos en ClickHouse
1. Este es el aspecto de los datos entrantes: ```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. Necesitaremos el motor de tabla **Kafka** para extraer los datos del topic de 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. Por alguna razón, el **motor de tabla Kafka** parece tomar la URL pública de **ec2** y convertirla en el nombre DNS privado, así que tuve que añadirlo a mi archivo local `/etc/hosts`: ```bash theme={null} 52.14.154.92 ip.us-east-2.compute.internal ``` 4. Puedes leer de una tabla de Kafka; solo tienes que habilitar una opción de configuración: ```sql theme={null} SELECT * FROM wikiQueue LIMIT 20 FORMAT Vertical SETTINGS stream_like_engine_allow_direct_select = 1; ``` Las filas deberían mostrarse correctamente interpretadas según las columnas definidas en la tabla **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. Necesitamos una tabla **MergeTree** para almacenar estos eventos entrantes: ```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. Definamos una **vista materializada** que se active cuando se realice una inserción en la tabla **Kafka** y envíe los datos a nuestra tabla **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. Deberías empezar a ver datos en **rawEvents** casi de inmediato: ```sql theme={null} SELECT count() FROM rawEvents; ``` 8. Veamos algunas de las filas: ```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. Veamos qué tipos de eventos están llegando: ``` 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 │ └────────────┴─────────┘ ``` Definamos una vista materializada encadenada a la vista materializada actual. Mantendremos algunas estadísticas agregadas por minuto: ```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. Necesitaremos las funciones **-Merge** para ver los resultados: ```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; ```