Python Airflow中解码Kafka二进制消息的实践指南

decoded_key = binary_key.decode('utf-8')
decoded_value = binary_value.decode('utf-8')

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
from kafka import KafkaConsumer
import json

def read_and_decode_kafka_messages(
    topic_name: str,
    bootstrap_servers: str,
    group_id: str,
    max_records: int = 10
):
    """
    从Kafka主题读取消息，并将其键和值从二进制解码为字符串。
    """
    consumer = KafkaConsumer(
        topic_name,
        bootstrap_servers=bootstrap_servers.split(','),
        group_id=group_id,
        auto_offset_reset='earliest', # 从最早的可用偏移量开始
        enable_auto_commit=True,
        value_deserializer=None, # 不使用内置的反序列化器，手动处理
        key_deserializer=None    # 不使用内置的反序列化器，手动处理
    )

    print(f"开始从Kafka主题 '{topic_name}' 消费消息...")
    processed_count = 0
    for message in consumer:
        try:
            # 消息的键和值都是bytes类型，需要解码
            message_key_decoded = message.key.decode('utf-8') if message.key else None
            message_value_decoded = message.value.decode('utf-8') if message.value else None

            print(f"主题: {message.topic}, 分区: {message.partition}, 偏移量: {message.offset}")
            print(f"解码后的键: {message_key_decoded}")
            print(f"解码后的值: {message_value_decoded}")

            # 进一步处理解码后的消息，例如解析JSON
            if message_value_decoded:
                try:
                    json_data = json.loads(message_value_decoded)
                    print(f"解析后的JSON数据: {json_data}")
                    # 在此处添加您的业务逻辑，例如写入数据库或进行进一步处理
                except json.JSONDecodeError:
                    print(f"警告: 消息值不是有效的JSON格式: {message_value_decoded}")

            processed_count += 1
            if processed_count >= max_records:
                print(f"已处理 {max_records} 条消息，停止消费。")
                break

        except UnicodeDecodeError as e:
            print(f"解码消息时发生错误: {e}")
            print(f"原始消息键: {message.key}, 原始消息值: {message.value}")
        except Exception as e:
            print(f"处理消息时发生未知错误: {e}")

    consumer.close()
    print("Kafka消费者已关闭。")

with DAG(
    dag_id='kafka_message_decoder_dag',
    start_date=datetime(2023, 1, 1),
    schedule_interval=None,
    catchup=False,
    tags=['kafka', 'data_pipeline'],
) as dag:
    decode_kafka_task = PythonOperator(
        task_id='read_and_decode_kafka_messages_task',
        python_callable=read_and_decode_kafka_messages,
        op_kwargs={
            'topic_name': 'your_kafka_topic', # 替换为您的Kafka主题名
            'bootstrap_servers': 'localhost:9092', # 替换为您的Kafka服务器地址
            'group_id': 'airflow_consumer_group',
            'max_records': 5 # 示例中只读取5条消息
        },
    )