如何实现Azure Data Explorer与MongoDB版Cosmos DB的无重复定时同步？

方案1：增量拉取+ADX Upsert模式（基础易实现） #

增量拉取配置 #

• 确保MongoDB集合每条记录带更新时间戳字段（如lastModified），或利用_id的ObjectId自带创建时间特性。
• 在ADF复制活动中，数据源选MongoDB版Cosmos DB，设置增量读取规则：
  • 用时间戳：每次读取lastModified大于上一次同步时间的记录，同步时间可存在ADX的水位表或ADF变量中。
  • 用_id：每次读取_id大于上次同步的最大_id的记录，同样持久化存储水位值。

ADX去重实现 #

• 复制活动接收器选Azure Data Explorer，写入行为设置为Upsert，指定ADX表的主键（如_id）。相同主键的记录会自动覆盖，新增记录直接插入，彻底避免重复。

删除记录同步 #

• 采用软删除方案：在MongoDB集合添加isDeleted字段（默认false，删除时设为true）。
• 增量同步时同步该字段，在ADX中通过定时Kusto查询或更新策略，删除isDeleted=true的记录，或移至归档表。
• 数据量较小时，可每次同步后对比ADX与MongoDB的主键列表，删除ADX中存在但MongoDB已移除的记录。

方案2：MongoDB Change Streams捕获全量变更（准实时同步） #

核心逻辑 #

MongoDB版Cosmos DB原生支持Change Streams，可实时捕获集合的插入、更新、删除等所有变更事件，通过Azure Function消费后推送到ADX。

操作步骤 #

1. 无需额外配置，MongoDB版Cosmos DB默认支持Change Streams（依赖其副本集/分片架构）。
2. 编写Azure Function（Node.js/Python均可），连接MongoDB集合监听Change Streams，捕获insert/update/delete事件。
3. 事件处理：
   • insert/update事件：将数据以Upsert模式推送到ADX（基于_id主键）。
   • delete事件：提取_id，调用ADX的.delete命令删除对应记录。
4. 可通过ADF定时触发Function拉取时段内变更，或让Function持续运行实现准实时同步。

方案3：ADF数据流+中间存储（绕过数据源限制） #

解决思路 #

利用中间存储（如Azure Blob Storage）中转，规避ADF数据流不支持MongoDB版Cosmos DB作为数据源的限制。

操作步骤 #

1. 第一阶段：ADF复制活动将MongoDB的增量数据同步到Blob Storage（用增量策略避免全量重复）。
2. 第二阶段：ADF数据流读取Blob Storage中的数据，写入ADX时选择Upsert模式（指定主键）；若结合软删除做全量同步，可选择重建表后插入。
3. 删除同步仍依赖软删除标记，在数据流中过滤或在ADX中执行删除操作。

关键注意事项 #

• 必须指定稳定唯一的主键（如MongoDB的_id），这是Upsert和删除操作的核心依据。
• 增量同步的水位值需持久化存储（ADX水位表、ADF变量或Azure Key Vault），避免重复拉取历史数据。
• 硬删除的高效捕获仅能通过Change Streams实现，软删除是低复杂度场景的最优选择。

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