终端读取Kafka Avro消息的kafka-avro-console-consumer轻量替代方案查询

以下方案均为MB级体积，无需下载完整Confluent平台：

方案1：kcat（原kafkacat，最推荐） #

kcat是目前应用最广泛的轻量Kafka CLI工具，单二进制文件仅几MB大小，原生支持Avro格式解析和Schema Registry对接。

安装方式 #

• Debian/Ubuntu 系：sudo apt install kafkacat
• macOS：brew install kcat
• 其他系统可直接下载官方预编译二进制文件，无需额外依赖

使用示例 #

kcat -b bootstrap_server_url \
  -t topic \
  -s value=avro \
  -r schema_registry_url \
  --from-beginning \
  -c 10

参数说明：-s value=avro 指定消息值的序列化格式为Avro，-r 填写Schema Registry地址，-c 对应原工具的--max-messages参数。

方案2：Python极简脚本 #

如果已有Python3环境，仅需安装少量轻量依赖即可实现，总依赖大小不超过10MB。

安装依赖 #

pip install confluent-kafka[avro]

脚本示例 #

from confluent_kafka import Consumer
from confluent_kafka.serialization import SerializationContext, MessageField
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroDeserializer

# 以下配置项按需修改
conf = {
    "bootstrap.servers": "bootstrap_server_url",
    "group.id": "temp-avro-consumer",
    "auto.offset.reset": "earliest"
}
schema_registry_conf = {"url": "schema_registry_url"}
topic = "target_topic_name"
max_read_count = 10

sr_client = SchemaRegistryClient(schema_registry_conf)
avro_deserializer = AvroDeserializer(sr_client)
consumer = Consumer(conf)
consumer.subscribe([topic])

read_count = 0
try:
    while read_count < max_read_count:
        msg = consumer.poll(1.0)
        if msg is None:
            continue
        if msg.error():
            print(f"消费错误：{msg.error()}")
            continue
        deserialized_value = avro_deserializer(
            msg.value(), 
            SerializationContext(topic, MessageField.VALUE)
        )
        print(deserialized_value)
        read_count += 1
finally:
    consumer.close()

修改配置项后直接运行脚本即可输出可读的Avro消息内容。

方案3：kcctl #

Kafka社区推出的现代CLI工具，单二进制文件无额外依赖，支持Avro解析，配置后可通过以下命令消费：

kcctl consume <topic名称> \
  --from-beginning \
  --max-messages 10 \
  --value-format avro \
  --schema-registry-url <Schema Registry地址>

内容的提问来源于stack exchange，提问作者y0j0