Python Airflow 集成 Kafka：消息解码实践

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
from confluent_kafka import Consumer, KafkaException, KafkaError
import json
import logging

# 配置日志
log = logging.getLogger(__name__)

def read_kafka_messages_task():
    """
    Airflow 任务，用于从 Kafka topic 读取并解码消息。
    """
    # Kafka 消费者配置
    conf = {
        'bootstrap.servers': 'localhost:9092',  # 替换为你的 Kafka 服务器地址
        'group.id': 'airflow_consumer_group',
        'auto.offset.reset': 'earliest',        # 从最早的可用偏移量开始消费
        'enable.auto.commit': False             # 手动控制偏移量提交
    }

    consumer = Consumer(conf)
    topic = 'test_topic'  # 替换为你的 Kafka topic 名称

    try:
        consumer.subscribe([topic])
        log.info(f"开始监听 Kafka topic: {topic}")

        # 尝试在一定时间内消费消息
        messages_processed = 0
        timeout_ms = 5000  # 5秒超时
        max_messages_to_process = 10  # 最多处理10条消息，防止无限循环

        while messages_processed < max_messages_to_process:
            # poll 方法的 timeout 参数是秒
            msg = consumer.poll(timeout=timeout_ms / 1000)

            if msg is None:
                log.info(f"在 {timeout_ms}ms 内未收到消息，停止消费。")
                break
            if msg.error():
                if msg.error().code() == KafkaError._PARTITION_EOF:
                    # 到达分区末尾
                    log.info(f'%% {msg.topic()} [{msg.partition()}] 已达到末尾偏移量 {msg.offset()}')
                elif msg.error():
                    raise KafkaException(msg.error())
            else:
                # 成功收到消息
                msg_key_bytes = msg.key()
                msg_value_bytes = msg.value()

                decoded_key = None
                decoded_value = None

                # 核心：解码二进制消息键和值
                # 假设使用 UTF-8 编码，如果你的数据是其他编码，请替换
                if msg_key_bytes:
                    try:
                        decoded_key = msg_key_bytes.decode('utf-8')
                    except UnicodeDecodeError:
                        log.warning(f"警告：消息键解码失败，原始字节：{msg_key_bytes}")
                        decoded_key = str(msg_key_bytes) # 作为备用，直接转换为字符串表示

                if msg_value_bytes:
                    try:
                        decoded_value = msg_value_bytes.decode('utf-8')
                        # 如果值是 JSON 字符串，可以进一步解析
                        # try:
                        #     decoded_value = json.loads(decoded_value)
                        # except json.JSONDecodeError:
                        #     log.debug(f"消息值不是有效的 JSON 格式，保持为字符串。")
                        #     pass
                    except UnicodeDecodeError:
                        log.warning(f"警告：消息值解码失败，原始字节：{msg_value_bytes}")
                        decoded_value = str(msg_value_bytes) # 作为备用

                log.info(f"成功从 Kafka topic: {msg.topic()}, partition: {msg.partition()}, offset: {msg.offset()} 收到记录。")
                log.info(f"消息键 (解码后): {decoded_key}")
                log.info(f"消息值 (解码后): {decoded_value}")
                messages_processed += 1

                # 手动提交偏移量（如果 enable.auto.commit 为 False）
                consumer.commit(message=msg)

    except Exception as e:
        log.error(f"读取 Kafka 消息时发生错误: {e}")
        raise # 抛出异常，Airflow 会将任务标记为失败
    finally:
        consumer.close()
        log.info("Kafka 消费者已关闭。")

with DAG(
    dag_id='kafka_message_decoder_dag',
    start_date=datetime(2023, 1, 1),
    schedule_interval=None,
    catchup=False,
    tags=['kafka', 'python', 'decoding'],
    doc_md="""
    ### Kafka 消息解码 DAG
    此 DAG 演示了如何在 Airflow 中使用 PythonOperator 从 Kafka topic 读取消息，
    并将其二进制键和值解码为可读的字符串格式。
    """
) as dag:
    read_and_decode_task = PythonOperator(
        task_id='read_and_decode_kafka_messages',
        python_callable=read_kafka_messages_task,
    )