如何用Python编写Azure Event Hub触发器函数写入Blob存储？

前置配置检查 #

首先确保你的local.settings.json或Azure门户应用配置中已正确配置Event Hub连接字符串、Blob存储连接字符串，且对应权限（Event Hub数据接收者、Blob存储数据写入者）已配置完成。

问题1：写入Blob为txt格式解决 #

• 写入时显式指定文件后缀和内容类型即可解决格式异常问题，如果你用Blob输出绑定，可在function.json的output配置中新增contentType字段：

{
  "name": "outputBlob",
  "type": "blob",
  "direction": "out",
  "path": "容器名/xxx/xxx/数据文件.txt",
  "connection": "BLOB存储连接字符串的配置项名",
  "contentType": "text/plain"
}

• 代码层面写入时直接传入UTF-8编码的字符串格式数据，不要转二进制流，避免内容类型被自动覆盖为二进制格式。

问题2：追加写入避免覆盖解决 #

默认Blob输出绑定使用块Blob，每次写入会覆盖原有文件，需要改用追加Blob实现写入不覆盖，推荐使用Blob SDK手动实现，灵活度更高：

import azure.functions as func
from azure.storage.blob import BlobServiceClient
import os
from datetime import datetime

# 重客户端放在全局作用域初始化，复用连接减少性能开销
blob_conn_str = os.environ["BLOB存储连接字符串的配置项名"]
blob_service_client = BlobServiceClient.from_connection_string(blob_conn_str)
container_name = "你的目标容器名"

app = func.FunctionApp()
@app.event_hub_message_trigger(arg_name="events", event_hub_name="你的Event Hub名", connection="EVENTHUB连接字符串的配置项名") 
def eventhub_processor(events: func.EventHubEvent):
    # 生成层级路径，对应问题3的要求
    now = datetime.utcnow()
    blob_path = f"Year={now.year}/month={now.month:02d}/date={now.day:02d}/event_log.txt"

    # 初始化追加Blob客户端
    append_blob_client = blob_service_client.get_blob_client(container=container_name, blob=blob_path)
    # Blob不存在则先创建追加Blob
    if not append_blob_client.exists():
        append_blob_client.create_append_blob()
    
    # 批量处理收到的事件，每条事件换行分隔
    write_content = ""
    for event in events:
        write_content += event.get_body().decode("utf-8") + "\n"
    
    # 追加内容到Blob
    append_blob_client.append_block(write_content, length=len(write_content))

小流量场景也可以在输出绑定的路径中加入{rand-guid}、事件偏移量等唯一标识避免覆盖，但会生成大量小文件，不适合高吞吐场景。

问题3：自定义层级路径实现 #

上述代码已经包含路径生成逻辑，按Year=xxxx/month=xx/date=xx格式生成即可，注意月份和日期用两位数字补零，避免后续路径筛选时出现排序异常。如果你需要按事件入队时间而非函数触发时间生成路径，可以从事件的系统属性中读取入队时间：

event_time = event.enqueued_time
blob_path = f"Year={event_time.year}/month={event_time.month:02d}/date={event_time.day:02d}/event_log.txt"

Python Azure函数开发最佳实践 #

• 批量触发配置：开启Event Hub批量触发，在host.json中调整extensions.eventHubs.maxBatchSize、extensions.eventHubs.minBatchSize参数，减少触发次数降低冷启动开销。
• 异常与死信处理：增加异常捕获逻辑，处理Blob写入失败的场景，将处理失败的事件写入死信队列避免数据丢失。
• 监控埋点：开启Application Insights集成，添加自定义指标监控函数执行耗时、事件处理成功率、Blob写入延迟等核心指标。
• 冷启动优化：消费级计划下精简第三方依赖包，移除不需要的引入，减少冷启动时间。
• 幂等性设计：Event Hub为至少一次交付模式，函数逻辑需要支持重复消费同一条事件不产生脏数据，可通过事件ID去重避免重复写入。
• 资源复用：所有重资源客户端（Blob客户端、数据库客户端等）都放在全局作用域初始化，不要在每次函数触发时重复创建。

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