Azure新手如何用Python/Scala处理事件文件并传入Azure EventHub

针对你当前描述的业务负载（单份10MB、含首尾头尾记录的固定结构flat file、单文件承载5-6万条事件），你提出的「用Python/Scala解析文件后直连EventHub发送」的方案，是匹配当前规模的高性价比落地方案，但并非所有场景下的最优解：

• 适配场景：如果你的文件传输是低到中频率（比如每小时/每天固定批次收文件）、没有上百个文件并行解析的高并发要求，这个方案开发成本极低，不需要依赖重型ETL组件，调试和运维难度远低于Azure原生管线，完全可以覆盖业务需求
• 不适配场景：如果后续业务量涨到单批文件总大小超过几十GB、需要近实时解析刚落地的文件、或者需要内置的解析重试/死信队列/复杂数据校验能力，那自写脚本的方案就会逐渐暴露出可维护性差的问题，这时候切换为Azure原生的Blob存储触发式管线（比如Azure Function、Data Factory映射流）会更省长期运维成本

通用前置准备 #

• 完成EventHub基础配置：创建EventHub命名空间、目标实例，获取连接字符串、发送端权限密钥，记录EventHub的实例名称
• 为解析程序配置flat file的读取权限：如果文件存在本地磁盘，确保程序有对应目录的读权限；如果文件存在Azure Blob存储，为程序分配存储容器的Blob Reader角色
• 明确flat file的固定规则：header的行标识、trailer的行标识、业务记录的分隔符、各字段的位置和数据类型

Python 实现路径 #

• 依赖安装：执行pip install azure-eventhub拉取官方SDK
• 核心实现逻辑：
  • 逐行流式读取文件，不要一次性将全文件加载进内存
  • 识别首行header、末尾trailer行并跳过，对中间的业务记录按照规则拆分字段，映射为结构化字典
  • 采用批量发送模式对接EventHub，注意EventHub单批次发送的消息总大小不能超过1MB，超过阈值时需要拆分批次
  • 配置简单的异常处理：单条记录解析失败时记录日志跳过，发送失败时做3次指数退避重试，不要因为单条错误中断全流程
• 核心参考代码片段：

from azure.eventhub import EventHubProducerClient, EventData
import json

# 初始化EventHub发送客户端
producer = EventHubProducerClient.from_connection_string(
    conn_str="替换为你的EventHub连接字符串",
    eventhub_name="替换为你的EventHub实例名称"
)

valid_events = []
# 逐行解析flat file
with open("替换为你的flat file本地/挂载路径", "r", encoding="utf-8") as f:
    for line_num, line in enumerate(f):
        line = line.strip()
        # 跳过首行header记录
        if line_num == 0:
            continue
        # 识别并跳过末尾trailer记录
        if line.startswith("替换为你的trailer行固定前缀"):
            break
        # 按照文件实际分隔符拆分字段
        split_cols = line.split("替换为你的文件分隔符，比如|、,")
        event_obj = {
            "event_id": split_cols[0],
            "event_time": split_cols[1],
            "event_content": split_cols[2]
        }
        valid_events.append(event_obj)

# 批量发送事件
batch = producer.create_batch()
for event in valid_events:
    try:
        # 序列化后加入当前批次
        batch.add(EventData(json.dumps(event)))
    except ValueError:
        # 批次达到大小上限时先发送当前批次，再新建批次
        producer.send_batch(batch)
        batch = producer.create_batch()
        batch.add(EventData(json.dumps(event)))
# 发送最后剩余的未填满批次
producer.send_batch(batch)
producer.close()

Scala 实现路径 #

• 依赖引入：通过Maven/Gradle引入官方azure-messaging-eventhubs SDK，无需额外集群依赖
• 核心实现逻辑：
  • 采用JVM流式API逐行读取文件，通过行标识过滤header、trailer记录
  • 自定义强类型Schema映射业务记录字段，校验必填字段和数据类型，非法记录写入死信日志
  • 调用SDK批量发送接口，按照1MB的单批大小阈值拆分消息，配置临时网络波动下的自动重试
  • 相比Python实现，Scala在并行处理多个大文件、高吞吐发送场景下的资源利用率更高，适合团队已经有Scala技术栈积累的场景

• 不要将EventHub连接字符串硬编码在代码中，通过环境变量或者Azure Key Vault注入配置，避免密钥泄露
• 当前10MB单文件的负载，哪怕用最低规格的Azure虚拟机、Consumption计划的Azure Function都可以承载，不需要额外采购高规格计算资源
• 为每条业务事件关联全局唯一的业务主键，发送前做简单去重校验，避免重复推送造成下游消费重复
• 给解析和发送流程加基础日志：单文件解析总条数、成功发送条数、失败的记录明细，日志接入Azure Monitor，方便快速定位异常

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