如何创建Kusto函数实现Azure Data Explorer到Dataverse的增量加载

核心思路 #

增量加载的关键是依赖增量标识字段（如最后修改时间戳、自增业务ID），通过对比上次同步的截止值，筛选出新增/更新的记录。以下是具体实现步骤：

1. 确认增量标识字段 #

优先选择能覆盖新增和更新场景的字段：

• 时间戳字段：如LastModifiedAt（记录最后修改时间，插入时也设置为当前时间）
• 自增ID字段：如RecordId（仅适用于只新增不更新的场景，或更新时ID不会变化的业务表）

如果Kusto表没有现成的时间戳字段，可通过更新策略自动维护：

.alter table YourKustoTableName policy update 
@'[{"Source": "Table", "Query": "YourKustoTableName | extend LastModifiedAt = now()", "IsEnabled": true}]'

2. 编写Kusto增量查询函数 #

根据选择的增量标识字段，创建可复用的函数：

场景1：基于时间戳（推荐，支持新增+更新）

.create-or-alter function GetKustoToDataverseIncremental(lastSyncTime: datetime) {
    // 替换为你的Kusto表名
    YourKustoTableName
    // 筛选上次同步后修改/新增的记录
    | where LastModifiedAt > lastSyncTime
    // 选择需要同步到Dataverse的字段，务必保留增量标识字段用于更新下次同步截止值
    | project 
        Id, 
        Name, 
        Description, 
        LastModifiedAt
}

场景2：基于自增ID（仅支持新增，不支持更新）

.create-or-alter function GetKustoToDataverseIncremental(lastSyncMaxId: long) {
    YourKustoTableName
    | where RecordId > lastSyncMaxId
    | project 
        RecordId, 
        Name, 
        Description
}

3. ADF管道配合逻辑 #

• 维护上次同步的截止值：可以用ADF变量、Azure SQL表或Key Vault存储（比如存储上次同步的最大LastModifiedAt）
• 同步流程：
  1. 读取上次同步的截止值
  2. 调用Kusto函数，传入截止值获取增量数据
  3. 将增量数据复制到Dataverse
  4. 查询本次同步的最大增量标识值（如max(LastModifiedAt)），更新存储的截止值

关键注意事项 #

• 避免重复同步：使用>而非>=筛选，确保只有上次同步后的数据被加载
• 性能优化：确保增量标识字段有索引（Kusto的datetime字段默认带索引，自增ID可设置为表的排序键）
• 异常处理：同步失败时不要更新截止值，避免数据丢失

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