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 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:

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

2. Instala Kafka (siguiendo las instrucciones de aquí: 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. Inicie ZooKeeper:

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

4. Abre una nueva terminal e inicia Kafka:

ssh -i ~/training.pem [e-mail]
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:

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. Puedes verificar que se haya creado correctamente de la siguiente manera:

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

​

2. Ingestar el flujo de Wikimedia a Kafka

1. Primero necesitamos algunas utilidades:

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:

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:

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

4. Verifiquemos que todo funcione correctamente consumiendo algunos eventos:

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:

{
	"$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. Necesitaremos el motor de tabla Kafka para extraer los datos del topic de 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. 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:

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:

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:

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:

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:

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:

SELECT count()
FROM rawEvents;

8. Veamos algunas de las filas:

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. Veamos qué tipos de eventos están llegando:

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 │
   └────────────┴─────────┘

Definamos una vista materializada encadenada a la vista materializada actual. Mantendremos algunas estadísticas agregadas por minuto:

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:

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;