如何通过ADF活动向CosmosDB(SQL API)插入数据并自定义id

一、更合适的插入方案（推荐使用存储过程） #

方案1：Stored Procedure Activity（最适配遍历场景） #

针对遍历文件并插入状态的需求，用存储过程可直接控制CosmosDB的插入逻辑，避开Copy Data Activity的源限制：

1. 在CosmosDB中创建存储过程
   编写接收参数的存储过程，负责构造合规文档并执行插入/更新：

   function upsertFileStatus(fileName, timestamp) {
       const context = getContext();
       const collection = context.getCollection();
       // 替换CosmosDB id不允许的特殊字符：/ \ : ? " < > |
       const safeFileName = fileName.replace(/[\/\\:?"<>|]/g, "_");
       // 构造状态文档
       const doc = {
           id: safeFileName,
           fileName: fileName,
           processedTimestamp: timestamp,
           status: "completed"
           // 可按需扩展字段，如处理时长、错误信息等
       };
   
       // 执行upsert（存在则更新，不存在则插入）
       const isAccepted = collection.upsertDocument(collection.getSelfLink(), doc, (err, res) => {
           if (err) throw new Error(`插入失败: ${err.message}`);
       });
       if (!isAccepted) throw new Error("CosmosDB服务忙，请求未被接受");
   }
   

2. ADF管道配置

   • 使用Get Metadata Activity获取Blob存储的文件列表（配置Field List为childItems）。
   • 添加ForEach Activity，遍历Get Metadata输出的childItems集合。
   • 在ForEach内部添加Stored Procedure Activity：
     • 链接到目标CosmosDB服务。
     • 选择创建好的存储过程。
     • 传入参数：
       • fileName：@item().name
       • timestamp：@utcNow()（或文件实际处理完成的时间戳）

方案2：Azure Function Activity（适合复杂业务逻辑） #

如果需要在插入前做复杂处理（如校验文件完整性、解析额外元数据），可使用Azure Function：

1. 编写Azure Function（以C#为例），通过CosmosDB SDK执行插入：

   [FunctionName("InsertFileStatus")]
   public static async Task<IActionResult> Run(
       [HttpTrigger(AuthorizationLevel.Function, "post", Route = null)] HttpRequest req,
       [CosmosDB(
           databaseName: "YourDB",
           collectionName: "FileStatus",
           ConnectionStringSetting = "CosmosDBConnection")] IAsyncCollector<FileStatus> outputCollection,
       ILogger log)
   {
       var requestBody = await new StreamReader(req.Body).ReadToEndAsync();
       var data = JsonConvert.DeserializeObject<dynamic>(requestBody);
       var fileName = data.fileName.ToString();
       var timestamp = data.timestamp.ToString();
   
       await outputCollection.AddAsync(new FileStatus
       {
           id = fileName.Replace("/", "_").Replace("\\", "_"),
           fileName = fileName,
           processedTimestamp = timestamp
       });
   
       return new OkObjectResult("插入成功");
   }
   
   public class FileStatus
   {
       public string id { get; set; }
       public string fileName { get; set; }
       public string processedTimestamp { get; set; }
   }
   

2. 在ADF的ForEach活动中调用该Function，传入文件名和时间戳参数。

二、沿用Copy Data Activity时自定义id的解决方法 #

若坚持使用Copy Data Activity，需解决源数据结构和映射问题：

1. 构造结构化数据源
   不能直接用二进制Blob作为源，需构造包含id、fileName、timestamp的结构化数据：
   • 使用Lookup Activity查询虚拟数据源（如Blob中存储的空JSON文件{"id":"","fileName":"","timestamp":""}），或连接SQL数据库执行SELECT '' AS id, '' AS fileName, '' AS timestamp返回固定结构。
2. 生成目标字段值
   在Lookup之后，用Set Variable构造JSON对象，将id和fileName设为当前遍历的文件名（@item().name），timestamp设为@utcNow()；或在Copy Activity的源转换中添加Derived Columns生成对应字段。
3. CosmosDB Sink配置
   • 在Copy Activity的Sink标签页，开启“允许从源指定id值”（关闭自动生成GUID的逻辑）。
   • 在Mapping标签页，确保源的id字段映射到CosmosDB集合的id字段，而非留空或映射到其他字段。
   • 提前替换文件名中的特殊字符，避免id不符合CosmosDB规则。

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