能否使用Kafka Avro Console Consumer/Producer实现Avro消息与文件读写

该场景可通过Avro控制台工具的原生参数实现，不需要开发自定义代码，仅需组合基础shell命令即可完成，无需额外编写复杂脚本。

• 第一步：导出Avro消息到本地文件
  调整Avro Console Consumer的序列化配置，直接输出原生Avro二进制内容到文件，避免默认的JSON格式化转换，参考命令如下：

  kafka-avro-console-consumer \
    --bootstrap-server 你的Kafka集群地址 \
    --topic 源Topic名称 \
    --property schema.registry.url=你的Schema Registry服务地址 \
    --property value.deserializer=org.apache.kafka.common.serialization.ByteArrayDeserializer \
    --property print.key=false \
    --property print.value=true \
    --from-beginning > avro_raw_messages.bin
  

  如果需要同时导出消息key，可新增--property print.key=true和--property key.separator='|'参数自定义分隔符，后续生产时对应配置相同的分隔符即可。
• 第二步：读取文件写入目标Kafka Topic
  直接使用Avro Console Producer读取导出的二进制文件，配合你已有的Schema配置即可写入目标Topic，参考命令如下：

  kafka-avro-console-producer \
    --bootstrap-server 你的Kafka集群地址 \
    --topic 目标Topic名称 \
    --property schema.registry.url=你的Schema Registry服务地址 \
    --property value.serializer=org.apache.kafka.common.serialization.ByteArraySerializer \
    --property value.schema='你的完整Avro Schema字符串' < avro_raw_messages.bin
  

小提示：如果你使用的是带Confluent Schema Registry标识的标准Avro消息格式，无需显式传入schema字符串，Producer会自动从Schema Registry拉取匹配的schema完成写入。

如果是单次临时迁移需求，直接执行上述两条命令即可完成操作，不需要编写脚本。如果需要定期执行、增加消息过滤、批量分片等逻辑，仅需编写简单的shell脚本封装上述命令即可，无需开发自定义消费/生产逻辑。

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