Python编写时间触发Azure Function实现Kafka数据上传Azure Blob参考代码问询

前置准备 #

• 首先安装Azure Blob存储SDK，将以下内容添加到项目的requirements.txt文件：

azure-functions
azure-storage-blob
confluent-kafka # 可根据你实际使用的Kafka依赖调整

• 在Azure Function应用配置（本地开发时是local.settings.json的Values节点）中添加2个配置项：
  • AZURE_STORAGE_CONNECTION_STRING：Azure存储账户的连接字符串，可在存储账户的「访问密钥」页面获取
  • TARGET_BLOB_CONTAINER：预先在存储账户中创建好的目标Blob容器名称

完整函数代码示例 #

import azure.functions as func
import json
from datetime import datetime
from azure.storage.blob import BlobServiceClient
import os
# 导入你原有项目的Kafka相关依赖
from confluent_kafka import Consumer, KafkaException

app = func.FunctionApp()

# 时间触发配置，每小时触发的cron表达式为 0 0 * * * *
@app.timer_trigger(schedule="0 0 * * * *", arg_name="myTimer", run_on_startup=False,
              use_monitor=False) 
def timer_trigger_kafka_to_blob(myTimer: func.TimerRequest) -> None:
    # 1. 初始化Blob客户端
    conn_str = os.environ["AZURE_STORAGE_CONNECTION_STRING"]
    container_name = os.environ["TARGET_BLOB_CONTAINER"]
    blob_service_client = BlobServiceClient.from_connection_string(conn_str)
    container_client = blob_service_client.get_container_client(container_name)

    # 2. 你原有拉取Kafka Stream数据的逻辑，以下仅为示例，可直接替换为自己的实现
    kafka_consumer = Consumer({
        'bootstrap.servers': '你的Kafka服务地址:9092',
        'group.id': 'azure-function-consumer-group',
        'auto.offset.reset': 'earliest'
    })
    kafka_consumer.subscribe(['你的Kafka Topic名称'])
    all_records = []
    try:
        # 拉取逻辑可根据业务调整时长、批量大小等参数
        msg = kafka_consumer.poll(timeout=10.0)
        while msg is not None:
            if msg.error():
                raise KafkaException(msg.error())
            else:
                record = json.loads(msg.value().decode('utf-8'))
                all_records.append(record)
            msg = kafka_consumer.poll(timeout=1.0)
    finally:
        kafka_consumer.close()

    # 3. 上传JSON数据到Blob Storage
    if all_records:
        # 生成唯一Blob名称，用UTC时间避免重名，可根据业务规则调整命名格式
        blob_name = f"kafka_data_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}.json"
        blob_client = container_client.get_blob_client(blob_name)
        # 转换为JSON字符串后上传
        json_data = json.dumps(all_records, ensure_ascii=False)
        blob_client.upload_blob(json_data, overwrite=True)

    if myTimer.past_due:
        print('本次定时器触发已延迟')

注意事项 #

• Blob命名可按需添加业务标识、分区信息等字段，方便后续数据检索
• 单批次数据量超过100MB时建议使用分块上传接口，降低上传失败概率
• 不需要覆盖历史数据时可去掉upload_blob方法的overwrite=True参数，遇到重名Blob时会抛出异常避免数据误覆盖
• 所有敏感配置不要硬编码在代码中，统一放到应用配置里管理

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