如何用Azure Synapse Analytics筛选MongoDB数据匹配CosmosDB的_id

核心思路 #

先从CosmosDB的Account容器提取所有目标_id，再用动态内容生成MongoDB的$in筛选条件，只复制Purchase集合中AccountId匹配这些_id的记录。

步骤1：配置Lookup活动获取Account ID列表 #

在Synapse管道中添加Lookup活动（命名为Get_Account_IDs）：

• 数据源：关联你的Azure CosmosDB连接，选择Account容器
• 查询语句：SELECT c._id FROM c（可按需添加过滤条件，比如WHERE c.Status = 'Active'）
• 取消勾选「First row only」，确保返回所有匹配的_id数组

步骤2：在复制活动中动态生成MongoDB筛选条件 #

添加复制活动，源端配置MongoDB连接与Purchase集合，然后在「查询/筛选器」中使用以下动态内容：

场景1：MongoDB的AccountId为字符串类型（与CosmosDB的_id字符串完全匹配） #

@concat('{"AccountId": {"$in": ', json(map(activity('Get_Account_IDs').output.value, item()._id)), '}}')

场景2：MongoDB的AccountId为ObjectId类型，CosmosDB的_id为字符串 #

如果MongoDB中AccountId存储为ObjectId，需要把CosmosDB的字符串_id转成MongoDB的ObjectId格式：

@concat('{"AccountId": {"$in": [', string(join(map(activity('Get_Account_IDs').output.value, concat('ObjectId("', item()._id, '")')), ', ')), ']}}')

注意事项与优化建议 #

• 若Account容器数据量过大（>1000条），Lookup活动可能出现性能瓶颈，可改为：
  • 用CosmosDB分页查询分批次提取_id
  • 先将Account的_id导出到ADLS Gen2，再通过Synapse SQL读取生成筛选条件
• 测试时先通过Lookup活动的输出预览，确认_id数组格式正确，再验证复制活动的筛选结果

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