> ## Documentation Index > Fetch the complete documentation index at: https://private-7c7dfe99-home-button.mintlify.site/llms.txt > Use this file to discover all available pages before exploring further. # 1. Configuración de Kafka en Ubuntu > Aprende a ingestar datos de un topic de Kafka en ClickHouse mediante el motor de tabla Kafka, vistas materializadas y tablas MergeTree.
## 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 4. 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; ```