如何在Azure Data Factory复制活动中实现OpenSearch Scroll分页

核心实现逻辑 #

ADF复制活动的原生分页规则无法直接适配Scroll API这种“首次请求生成scroll_id、后续请求依赖该ID”的特殊分页模式，需要通过管道变量+多复制活动+循环控制的组合方式来实现，具体步骤如下：

步骤1：初始化管道变量 #

创建一个字符串类型的管道变量（例如scrollId），初始值设为空字符串，用于存储每次请求返回的scroll_id。

步骤2：首次请求获取初始数据与scroll_id #

添加第一个复制活动，配置细节：

• 源数据集：选择REST类型，API地址配置为https://<你的OpenSearch实例地址>/products/_search?scroll=1m，请求方法设为POST，请求体填写：

  {"size": 10}
  

• 捕获scroll_id：在复制活动执行完成后，添加一个Set Variable活动，通过表达式@activity('<首次复制活动名称>').output.responseBody._scroll_id，将返回的scroll_id赋值给scrollId变量。
• 数据写入：将首次请求返回的文档数据直接写入目标存储。

步骤3：循环执行后续Scroll请求 #

添加Until循环活动，设置循环终止条件为@empty(variables('scrollId'))（当scroll_id为空或无更多数据时停止循环）：

1. 循环内部添加第二个复制活动：
   • 源数据集：REST类型，API地址设为https://<你的OpenSearch实例地址>/_search/scroll，请求方法POST，请求体通过变量动态填充scroll_id：

     {"scroll_id": "@variables('scrollId')", "scroll": "1m"}
     

   • 数据写入：将返回的文档数据以追加模式写入目标存储。
2. 添加Set Variable活动更新scrollId：
   • 如果响应中仍有有效scroll_id，用表达式@activity('<后续复制活动名称>').output.responseBody._scroll_id更新变量；如果响应无更多数据（hits为空或无scroll_id），则将变量设为空字符串以触发循环终止。

关键注意事项 #

• scroll有效期续期：每次后续请求必须携带scroll参数，避免scroll_id提前过期。
• 数据一致性：确保目标存储的写入模式为追加，防止重复写入数据。
• 错误处理：添加错误分支捕获请求失败或scroll_id无效的情况，避免循环死锁。

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