Python Airflow 中处理 Kafka 二进制消息的解码实践

decoded_key = msg_key_bytes.decode('utf-8')
decoded_value = msg_value_bytes.decode('utf-8')

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
from kafka import KafkaConsumer
import json # 用于进一步处理JSON格式的消息

def read_and_decode_kafka_messages():
    """
    从 Kafka 主题读取消息并进行解码。
    假定消息使用 UTF-8 编码。
    """
    topic_name = 'your_kafka_topic' # 替换为你的 Kafka 主题名称
    bootstrap_servers = ['localhost:9092'] # 替换为你的 Kafka Broker 地址列表

    # 初始化 Kafka 消费者
    consumer = KafkaConsumer(
        topic_name,
        bootstrap_servers=bootstrap_servers,
        auto_offset_reset='earliest', # 从主题的起始位置开始消费
        enable_auto_commit=True,      # 自动提交偏移量
        group_id='airflow_consumer_group', # 消费者组ID
        # consumer_timeout_ms=5000 # 如果在5秒内没有新消息，则停止消费
    )

    print(f"尝试从主题: {topic_name} 读取消息...")
    messages_processed = 0

    try:
        for message in consumer:
            try:
                # Kafka 消息的 key 和 value 都是 bytes 对象，需要解码
                # 检查 key 或 value 是否存在，避免对 None 调用 decode()
                decoded_key = message.key.decode('utf-8') if message.key else None
                decoded_value = message.value.decode('utf-8') if message.value else None

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

                # 如果消息值是 JSON 字符串，可以进一步解析
                if decoded_value and decoded_value.strip().startswith('{') and decoded_value.strip().endswith('}'):
                    try:
                        json_data = json.loads(decoded_value)
                        print(f"解析后的 JSON 数据: {json_data}")
                    except json.JSONDecodeError as e:
                        print(f"错误：无法将消息值解析为 JSON: {e}")

                messages_processed += 1
                # 示例限制：处理一定数量的消息后停止，实际应用中可能需要更复杂的停止逻辑
                if messages_processed >= 10: 
                    print("已处理10条消息，示例停止。")
                    break

            except UnicodeDecodeError as e:
                print(f"错误：解码消息失败 (偏移量 {message.offset})：{e}")
                print(f"原始键: {message.key}")
                print(f"原始值: {message.value}")
            except Exception as e:
                print(f"发生未知错误：{e}")

    finally:
        consumer.close() # 确保消费者在任务结束时关闭
        print(f"完成消息读取。总共处理消息数: {messages_processed}")

with DAG(
    dag_id='kafka_message_decoder_dag',
    start_date=datetime(2023, 1, 1),
    schedule_interval=None, # 此 DAG 为手动触发或外部触发
    catchup=False,
    tags=['kafka', 'decoding', 'python', 'airflow'],
) as dag:
    decode_kafka_task = PythonOperator(
        task_id='read_and_decode_kafka_messages_task',
        python_callable=read_and_decode_kafka_messages,
    )