如何将Airflow读取Kafka的二进制消息转为可读文本？

情况1：消息为普通字符串编码（如UTF-8） #

如果生产者发送的是字符串类消息，只是以二进制形式存储，直接对key和value调用.decode()方法即可，常用编码为utf-8：

# 假设从Kafka获取到的消息对象为msg
msg_key = msg.key().decode('utf-8') if msg.key() else None
msg_value = msg.value().decode('utf-8') if msg.value() else None

print(f"message key: {msg_key} || message value: {msg_value}")

若遇到解码错误（存在非UTF-8字符），可添加错误处理：

# 用?替换无法解码的字符
msg_value = msg.value().decode('utf-8', errors='replace')
# 或忽略无法解码的字符
msg_value = msg.value().decode('utf-8', errors='ignore')

情况2：消息使用了序列化框架（如Avro/Protobuf） #

从你提供的二进制内容来看，包含部分可读字符串但存在大量不可见控制字符，大概率是生产者用了Avro、Protobuf这类序列化工具。这种情况需要匹配生产者的序列化规则和Schema来反序列化：

以Avro为例，使用fastavro库处理：

1. 安装依赖：

pip install fastavro

2. 反序列化代码（需提前获取生产者使用的Avro Schema）：

import fastavro
from io import BytesIO

# 填入生产者对应的Avro Schema（JSON格式）
schema = {
    # Schema内容
}

binary_value = msg.value()
# 部分生产者会在前5字节添加magic byte和schema ID，需跳过
if binary_value.startswith(b'\x00'):
    binary_value = binary_value[5:]

# 反序列化得到可读字典/对象
bytes_io = BytesIO(binary_value)
msg_value = fastavro.schemaless_reader(bytes_io, schema)

print(f"message value: {msg_value}")

如果是Protobuf格式，需要先通过.proto文件生成Python类，再用protobuf库完成反序列化。

额外提示 #

• 优先确认生产者的消息序列化方式，这是解决问题的核心；
• 若使用Airflow的KafkaConsumerOperator，可直接在配置中指定反序列化函数：

from airflow.providers.apache.kafka.operators.consumer import KafkaConsumerOperator

def process_msg(message, context):
    print(f"message key: {message.key} || message value: {message.value}")

consumer_task = KafkaConsumerOperator(
    task_id='kafka_consumer',
    topics=['test_topic'],
    kafka_config={'bootstrap.servers': '你的Kafka地址'},
    value_deserializer=lambda x: x.decode('utf-8') if x else None,
    process_function=process_msg
)

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