apache-kafka
herramientas de consola kafka

Introducción

kafka-temas

Esta herramienta le permite enumerar, crear, alterar y describir temas.

Lista de temas:

kafka-topics  --zookeeper localhost:2181 --list

Crear un tema:

kafka-topics  --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

crea un tema con una partición y sin replicación.

Describe un tema:

kafka-topics  --zookeeper localhost:2181 --describe --topic test

Alterar un tema:

# change configuration
kafka-topics  --zookeeper localhost:2181 --alter --topic test --config max.message.bytes=128000
# add a partition
kafka-topics  --zookeeper localhost:2181 --alter --topic test --partitions 2

(Cuidado: Kafka no admite la reducción del número de particiones de un tema) (consulte esta lista de propiedades de configuración )

productor de consola kafka

Esta herramienta te permite producir mensajes desde la línea de comandos.

Enviar mensajes simples de cadena a un tema:

kafka-console-producer --broker-list localhost:9092 --topic test
here is a message
here is another message
^D

(cada nueva línea es un mensaje nuevo, escriba ctrl + D o ctrl + C para detener)

Enviar mensajes con claves:

kafka-console-producer --broker-list localhost:9092 --topic test-topic \
        --property parse.key=true \
        --property key.separator=,
key 1, message 1
key 2, message 2
null, message 3
^D

Enviar mensajes desde un archivo:

kafka-console-producer --broker-list localhost:9092 --topic test_topic < file.log

kafka-console-consumer

Esta herramienta te permite consumir mensajes de un tema.

para usar la implementación anterior del consumidor, reemplace --bootstrap-server con --zookeeper .

Mostrar mensajes simples:

kafka-console-consumer --bootstrap-server localhost:9092 --topic test 

Consumir mensajes antiguos:

Para ver los mensajes más antiguos, puede usar la opción --from-beginning .

Mostrar mensajes de valor-clave :

kafka-console-consumer  --bootstrap-server localhost:9092 --topic test-topic \
    --property print.key=true \
    --property key.separator=, 

kafka-simple-consumidor-shell

Este consumidor es una herramienta de bajo nivel que le permite consumir mensajes de particiones, compensaciones y réplicas específicas.

Parámetros útiles:

• parition : la partición específica para consumir a partir de (por defecto a todos)
• offset : el offset inicial. Usa -2 para consumir mensajes desde el principio, -1 para consumir desde el final.
• max-messages : número de mensajes para imprimir
• replica : la réplica, predeterminada para el broker-leader (-1)

Ejemplo:

kafka-simple-consumer-shell  \
    --broker-list localhost:9092 \
    --partition 1 \
    --offset 4 \
    --max-messages 3 \
    --topic test-topic

muestra 3 mensajes de la partición 1 que comienzan en el desplazamiento 4 del tema prueba-tema.

kafka-grupos de consumidores

Esta herramienta le permite enumerar, describir o eliminar grupos de consumidores. Eche un vistazo a este artículo para obtener más información sobre los grupos de consumidores.

si aún usa la implementación anterior del consumidor, reemplace --bootstrap-server con --zookeeper .

Lista de grupos de consumidores:

kafka-consumer-groups  --bootstrap-server localhost:9092 --list
octopus

Describa un grupo de consumidores:

kafka-consumer-groups  --bootstrap-server localhost:9092 --describe --group octopus
GROUP          TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG       OWNER
octopus        test-topic      0          15              15              0         octopus-1/127.0.0.1
octopus        test-topic      1          14              15              1         octopus-2_/127.0.0.1

Observaciones : en la salida anterior,

• current-offset es el último offset comprometido de la instancia del consumidor,
• log-end-offset es el desplazamiento más alto de la partición (por lo tanto, sumar esta columna le da el número total de mensajes para el tema)
• lag es la diferencia entre la compensación del consumidor actual y la compensación más alta, por lo tanto, cuán lejos está el consumidor,
• owner es el client.id del consumidor (si no se especifica, se muestra uno predeterminado).

Eliminar un grupo de consumidores:

la eliminación solo está disponible cuando los metadatos del grupo se almacenan en zookeeper (antigua api del consumidor). Con la nueva API para el consumidor, el agente maneja todo lo que incluye la eliminación de metadatos: el grupo se elimina automáticamente cuando expira el último desplazamiento confirmado para el grupo.

kafka-consumer-groups --bootstrap-server localhost:9092 --delete --group octopus