Azure Synapse Pipeline 中的动态 REST 调用答案

【问题标题】：Dynamic REST calls in Azure Synapse PipelineAzure Synapse Pipeline 中的动态 REST 调用
【发布时间】：2021-11-27 01:57:08
【问题描述】：

我正在使用 Azure Synapse 调用 REST API，返回数据集如下所示：

{
"links": [
    {
        "rel": "next",
        "href": "[myRESTendpoint]?limit=1000&offset=1000"
    },
    {
        "rel": "last",
        "href": "[myRESTendpoint]?limit=1000&offset=60000"
    },
    {
        "rel": "self",
        "href": "[myRESTendpoint]"
    }
],
"count": 1000,
"hasMore": true,
"items": [
    {
        "links": [],
        "closedate": "6/16/2014",
        "id": "16917",
        "number": "62000",
        "status": "H",
        "tranid": "0062000"
    },...
],
"offset": 0,
"totalResults": 60316
}

我熟悉对单个端点进行 REST 调用，该端点可以使用 Synapse 管道通过单个调用返回所有数据，但是这个特定的 REST 端点对仅返回 1000 条记录有硬性限制，但它确实提供了一个名为“hasMore”的属性。

有没有办法在 Synapse 管道中递归地进行休息调用，直到“hasMore”属性等于 false？

这样做的最终目标是将数据接收到专用 SQL 池或 ADLS2 并从那里进行转换。

【问题讨论】：

标签： azure rest azure-data-factory azure-synapse

【解决方案1】：

我尝试使用 Azure 数据工厂来实现相同的场景，这似乎更合适且更容易实现目标“这样做的最终目标是将数据下沉到专用 SQL 池或 ADLS2 并从那里转换”。

由于您必须递归地访问页面以获取 1000 条记录，如果响应标头/响应正文包含下一页的 URL，您可以按以下方式设置它。

如果下一页链接或查询参数未包含在响应标头/正文中，您将不太可能使用该功能。

或者，您可以利用循环逻辑并执行复制活动。

在 Rest Connector 中创建两个参数：

填写 RestConnector 的相对 URL 参数。

使用设置变量操作，该变量的值将在循环中增加。对于每个循环，Copy Activity 的 URL 都是动态设置的。如果要循环或迭代，可以使用 Until 活动。

替代方案：

根据我的经验，REST 连接分页非常严格。通常将动作放在一个循环中。结果，有更多的控制权。 FOREACH 循环，here

【讨论】：

我在周末玩了这个并做了一些额外的搜索。您建议的内容适用于 REST api 的第 1 页，但我仍然遇到分页规则问题。当我尝试在动态内容窗口中使用 $ 运算符时出现错误，因为我需要一个 if 语句来查看“链接”数组并找到显示“下一个”的“rel”属性，然后返回“ href" 属性在同一个对象中获取下一页 url。我可以像你提到的那样使用 foreach 循环，我只是好奇如果没有它也可以工作。谢谢！
感谢@mfarinella14 的更新，能否请您分享错误，是的，请使用 ForEach 循环尝试一下。
ForEach 循环运行良好，除了在这种情况下我想使用嵌套的 ForEach 循环并且 Synapse 中不允许这样做。对于 AbsoluteUrl 中的动态值，我可以使用：@if(equals($['links'][0]['rel'],'next'),$['links'][0]['href' ],if()) 第二个 if() 只是表示嵌套 if 语句，并且显示的错误是：Unrecognized expression: $['links'][0]['rel']

【解决方案2】：

对于那些关注线程的人，我使用了 IpsitaDash-MT 的建议，使用 ForEach 循环。在此 API 的情况下，当进行调用时，我会在调用结束时返回一个名为“totalResults”的属性。以下是我用来实现我想要做的事情的步骤：

对 API 进行虚拟调用以获取“totalResults”参数。这只是一个返回我希望获得的结果数量的调用。在此 API 的情况下，请求的主体是一条 SQL 语句，因此在发出虚拟请求时，我只要求提供我想要获得的结果的 ID。

SQL statement example

然后我从该请求中获取属性“totalResults”，在 ForEach 循环的“项目”中设置一个动态值，如下所示：

@range(0,add(div(sub(int(activity('Get Pages Customers').output.totalResults),mod(int(activity('Get Pages Customers').output.totalResults),1000) ),1000),1))

注意： API 只允许包含 1000 个结果的页面，我做了一些数学运算来获得一系列页码。我还必须在最终结果中加 1 以包含最后一页。

ForEach Loop Settings

在 API 中，我有两个参数可以传递“limit”和“offset”。由于我想要所有数据，因此没有理由将限制设置为 1000（允许的最大数量）以外的任何值。偏移量参数可以设置为任何小于或等于“totalResults”-“限制”且大于等于0的数字。所以我使用步骤2中建立的范围并将其乘以1000来设置偏移量参数网址。

Setting the offset parameter in the copy data activity

Dynamic value of the Relative URL in the REST connector

注意：由于查找功能，我发现最好先将数据作为 JSON 放入 ADLS2，而不是放入专用 SQL 池。

由于突触不允许嵌套 ForEach 循环，我通过数据流运行数据以格式化数据并检查重复和更新。
当数据流完成时，它会启动查找活动以获取刚刚处理的数据并将其传递到新管道中，以使用另一个 ForEach 循环来获取父数据的每个 ID 的子数据。

Data Flow and Lookup for child data pipeline

【讨论】：