Azure Data Factory复制Azure DB for MySQL数据后如何获取变更记录ID的事件通知？

ADF原生的复制活动完成通知（成功/失败）不会直接携带具体的变更记录ID，但可以通过组合ADF内的组件或集成其他服务实现这个需求，具体方案如下：

核心思路 #

先捕获需要更新的记录ID，再将这些ID作为事件内容发送到EventGrid或EventBus，分两步实现：

1. 捕获变更记录ID #

• 用Lookup活动预查询变更ID：在复制活动前添加Lookup活动，编写SQL查询源表中需要更新到目标表的记录（比如基于时间戳筛选增量数据，或对比目标表的主键），获取这些记录的ID列表。将查询结果存储到ADF的数组变量中。
• 复制活动关联变更范围：把Lookup活动得到的ID列表作为复制活动的筛选条件（比如在复制活动的源数据集查询中加入WHERE id IN (@variables('changeIds'))），确保复制活动只处理这些变更记录。

2. 发送事件通知 #

根据选择的目标服务，选择对应的实现方式：

• 发送到Azure EventGrid：

  1. 在Azure中创建EventGrid自定义主题，生成SAS令牌用于认证。
  2. 在复制活动成功后添加Web活动，调用EventGrid的发布API，请求体中包含变更记录ID数组，示例请求体：

     {
       "topic": "/subscriptions/{订阅ID}/resourceGroups/{资源组}/providers/Microsoft.EventGrid/topics/{主题名}",
       "eventType": "MySQLRecordUpdated",
       "subject": "AzureDBforMySQL/TableUpdates",
       "data": {
         "changedRecordIds": @variables('changeIds')
       },
       "eventTime": "@utcNow()",
       "id": "@guid()"
     }
     

  3. 在Web活动的认证设置中填入SAS令牌，完成事件发送。

• 发送到Azure EventBus：

  1. 在Azure中创建EventBus命名空间和队列/主题。
  2. 在复制活动成功后添加Web活动，调用Service Bus的发送消息API，将变更记录ID数组转为JSON字符串作为消息内容，认证使用Shared Access Signature。

• 集成Logic Apps简化流程：
  如果不想在ADF中编写复杂的Web活动配置，可以将变更记录ID写入Azure Blob存储的JSON文件，然后触发Logic Apps：Logic Apps读取文件中的ID列表，再转发到EventGrid/EventBus，这种方式更适合复杂的事件格式需求。

注意事项 #

• 若使用复制活动的Upsert写入行为，需确保目标表已指定主键，ADF会自动处理新增/更新，但仍需单独捕获变更ID，因为复制活动的输出仅包含行数统计，不包含具体ID。
• 对于大数据量的变更，建议将ID列表分批处理，避免变量或请求体过大导致失败。

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