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

# 示例二进制数据
binary_data = b'Hello Kafka!'
# 使用utf-8编码进行解码
decoded_string = binary_data.decode('utf-8')
print(f"原始二进制数据: {binary_data}")
print(f"解码后的字符串: {decoded_string}")
# Output:
# 原始二进制数据: b'Hello Kafka!'
# 解码后的字符串: Hello Kafka!

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
from kafka import KafkaConsumer # 假设已安装kafka-python库

def consume_and_decode_kafka_messages(topic_name, bootstrap_servers):
    """
    消费指定Kafka主题的消息并解码其键和值。
    """
    consumer = KafkaConsumer(
        topic_name,
        bootstrap_servers=bootstrap_servers,
        auto_offset_reset='earliest', # 从最早的可用消息开始消费
        enable_auto_commit=True,      # 自动提交偏移量
        group_id='airflow_consumer_group',
        consumer_timeout_ms=5000      # 如果在5秒内没有消息，则consumer.poll()会超时
    )

    print(f"开始消费Kafka主题: {topic_name}")
    decoded_messages = []
    try:
        for message in consumer:
            # 消息的键和值都是bytes类型，需要解码
            # 在解码前检查是否为None，因为键和值都可能为空
            msg_key = message.key.decode('utf-8') if message.key is not None else None
            msg_value = message.value.decode('utf-8') if message.value is not None else None

            print(f"Topic: {message.topic}, Partition: {message.partition}, Offset: {message.offset}")
            print(f"Decoded Key: {msg_key} || Decoded Value: {msg_value}")

            decoded_messages.append({
                'key': msg_key,
                'value': msg_value,
                'topic': message.topic,
                'partition': message.partition,
                'offset': message.offset
            })
    except Exception as e:
        print(f"消费Kafka消息时发生错误: {e}")
    finally:
        consumer.close()
        print(f"成功消费并解码 {len(decoded_messages)} 条消息。")
        # 可以在这里对解码后的消息进行进一步处理，例如存储到数据库或传递给下一个任务
        return decoded_messages

with DAG(
    dag_id='kafka_message_decoder_dag',
    start_date=datetime(2023, 1, 1),
    schedule_interval=None,
    catchup=False,
    tags=['kafka', 'decoding'],
) as dag:
    decode_kafka_task = PythonOperator(
        task_id='decode_kafka_messages',
        python_callable=consume_and_decode_kafka_messages,
        op_kwargs={
            'topic_name': 'your_kafka_topic',         # 替换为你的Kafka主题名
            'bootstrap_servers': 'your_kafka_broker_ip:9092' # 替换为你的Kafka Broker地址
        },
    )