如何将Azure Blob Storage中的JSON数据加载到Apache Kafka中

首先纠正一个常见误解：你用到的Azure Blob Storage Source Connector本身就支持读取任意第三方直接上传到Blob存储的JSON文件，并不要求文件必须先通过同系列Sink Connector写入，只要调整对应配置即可实现自动监听增量文件的需求，这也是当前场景下的首选无代码方案。

方案1：Kafka Connect Azure Blob Storage Source Connector（优先推荐） #

无需编写业务代码，只需要调整连接器配置即可实现全自动化的文件监听、导入逻辑：

• 基础连接配置：指定Blob存储的账号密钥、要监听的容器名称，以及导入的目标Kafka Topic
• 格式适配配置：将记录处理器参数设置为JSON类型 io.confluent.connect.azure.blob.storage.source.format.JsonRecordProcessor，即可自动解析Blob中的JSON内容生成Kafka消息
• 增量监听配置：设置扫描间隔参数 azure.blob.source.scan.interval.ms，比如设置为30000即代表每30秒扫描一次容器内的新增文件，连接器会自动记录已处理文件的偏移量，不会出现重复导入的问题
• 性能配置：根据容器内文件数量调整 tasks.max 并发数，提升大流量下的导入效率

方案2：Azure Functions Blob触发器（轻量场景可选） #

如果不想自行维护Kafka Connect集群，适合流量规模较小的场景：

• 给目标Blob存储容器配置Blob触发器，平台会自动监听容器内的新增/更新文件，触发函数执行
• 函数内仅需要编写少量逻辑：读取Blob文件的JSON内容，调用Kafka生产者API写入目标Topic即可
• 支持按调用量付费，无需长期持有服务器资源，成本更低

通用注意事项 #

• 建议配置Schema Registry做JSON格式校验，避免不符合要求的脏数据写入Kafka Topic
• 超过Kafka单条消息大小限制的大文件建议提前做分片处理，避免导入失败
• 配置死信队列承接格式异常、解析失败的文件内容，方便后续排查问题

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