如何通过Azure Function App动态同步Azure Blob文件至Cosmos MongoDB集合

前置准备 #

• 已创建Azure Blob存储账户，jsonfiles容器文件夹下存放目标11个JSON文件（a.json至k.json）
• 已创建Azure Cosmos DB MongoDB API账户，且提前创建好11个对应集合（注意i.json对应大写I集合）
• 已创建Azure Function App（本文以Python runtime为例，C#逻辑类似）

配置Function触发与绑定 #

采用Blob触发器作为触发源，具体配置：

• 触发器路径设置为 jsonfiles/{name}.json，确保jsonfiles文件夹下的JSON文件新增/更新时自动触发Function
• 推荐直接通过SDK操作Cosmos DB（比输出绑定更灵活，便于处理集合名映射）

核心代码实现（Python） #

1. 定义依赖：在Function App根目录创建requirements.txt，添加以下依赖：

azure-functions
pymongo>=4.0

2. 编写Function代码（__init__.py）：

import azure.functions as func
from pymongo import MongoClient
import json
import os

def main(myblob: func.InputStream):
    # 提取文件名（去除后缀），映射到目标集合名
    raw_file_name = myblob.name.split('/')[-1].rstrip('.json')
    # 处理i.json对应大写I集合的特殊情况
    target_collection = 'I' if raw_file_name == 'i' else raw_file_name

    # 读取Blob中的JSON数据
    try:
        json_data = json.loads(myblob.read().decode('utf-8'))
    except json.JSONDecodeError as e:
        print(f"文件{myblob.name}格式错误: {str(e)}")
        return

    # 连接Cosmos MongoDB并写入数据
    try:
        conn_str = os.environ["COSMOS_MONGODB_CONN"]
        client = MongoClient(conn_str)
        db = client['your-db-name']  # 替换为你的数据库名称
        collection = db[target_collection]

        # 支持单条或数组批量插入
        if isinstance(json_data, list):
            collection.insert_many(json_data)
        else:
            collection.insert_one(json_data)
        
        print(f"完成：{myblob.name} -> 集合{target_collection}")
    except Exception as e:
        print(f"数据写入失败: {str(e)}")
        return

配置环境变量 #

在Function App的配置 > 应用程序设置中添加：

• COSMOS_MONGODB_CONN：从Cosmos DB账户的「连接字符串」页面复制MongoDB连接字符串

权限配置 #

• 给Function App分配Blob存储的存储Blob数据读取者角色，确保能读取Blob文件
• 确保Cosmos DB连接字符串拥有写入集合的权限（推荐用账户级连接字符串，或配置细粒度权限）

测试与验证 #

1. 手动上传一个JSON文件到jsonfiles文件夹，触发Function执行
2. 查看Function的「监控 > 日志」，确认执行状态与结果
3. 登录Cosmos DB数据资源管理器，检查对应集合是否已写入数据
4. 对于已存在的11个文件，可手动触发Function对每个文件执行一次（或通过批量脚本触发）

注意事项 #

• 异常处理：代码中添加了基础的格式错误、连接错误捕获，可根据需求扩展重试逻辑
• 幂等性：如果需要避免重复写入，可在JSON数据中添加唯一标识，写入时用replace_one替代insert方法
• 性能：对于大文件，可考虑分块读取处理，避免内存溢出

内容的提问来源于stack exchange，提问作者Mohan Kumar G