使用 Azure 数据工厂从/向 REST 终结点复制和转换数据

适用于：Azure 数据工厂 Azure Synapse Analytics

本文概述了如何使用 Azure 数据工厂中的复制活动从 REST 终结点复制数据以及向其中复制数据。 本文是根据总体概述复制活动的 Azure 数据工厂中的复制活动编写的。

此 REST 连接器、HTTP 连接器和 Web 表连接器之间的区别如下：

• REST connector 连接器专门支持从 RESTful API 复制数据。
• HTTP 连接器是通用的，可从任何 HTTP 终结点检索数据，以执行文件下载等操作。 在使用此 REST 连接器之前，你可能会偶尔使用 HTTP 连接器从 RESTful API 复制数据，虽然 HTTP 连接器受支持，但与 REST 连接器相比功能较少。
• Web 表连接器用于从 HTML 网页中提取表内容。

此 REST 连接器支持以下功能：

① Azure 集成运行时 ② 自承载集成运行时

如需可以用作源/接收器的数据存储的列表，请参阅支持的数据存储。

具体而言，此泛型 REST 连接器支持：

• 使用 GET 或 POST 方法从 REST 终结点复制数据，以及使用 POST、PUT 或 PATCH 方法将数据复制到 REST 终结点。
• 使用以下身份验证之一复制数据：“匿名”、“基本”、“服务主体”、“OAuth2 客户端凭据”、“系统分配的托管标识”和“用户分配的托管标识”。
• REST API 中的分页。
• 对于作为源的 REST，按原样复制 REST JSON 响应，或使用架构映射对其进行分析。 仅支持 JSON 格式的响应有效负载。

如果数据存储位于本地网络、Azure 虚拟网络或 Amazon Virtual Private Cloud 内部，则需要配置自承载集成运行时才能连接到该数据存储。

如果数据存储是托管的云数据服务，则可以使用 Azure Integration Runtime。 如果访问范围限制为防火墙规则中允许的 IP，你可以选择将 Azure Integration Runtime IP 添加到允许列表。

此外，还可以使用 Azure 数据工厂中的托管虚拟网络集成运行时功能访问本地网络，而无需安装和配置自承载集成运行时。

要详细了解网络安全机制和数据工厂支持的选项，请参阅数据访问策略。

若要使用管道执行复制活动，可以使用以下工具或 SDK 之一：

• 复制数据工具
• Azure 门户
• .NET SDK
• Python SDK
• Azure PowerShell
• REST API
• Azure 资源管理器模板

使用以下步骤在 Azure 门户 UI 中创建 REST 链接服务。

1. 浏览到 Azure 数据工厂或 Synapse 工作区中的“管理”选项卡，并选择“链接服务”，然后单击“新建”：

   • Azure 数据工厂
   • Azure Synapse

   

2. 搜索“REST”并选择 REST 连接器。

   

3. 配置服务详细信息、测试连接并创建新的链接服务。

   

对于特定于 REST 连接器的数据工厂实体，以下部分提供了有关用于定义这些实体的属性的详细信息。

REST 链接服务支持以下属性：

有关不同的身份验证类型，请参阅相应的部分了解详细信息。

• 基本身份验证
• 服务主体身份验证
• OAuth2 客户端凭据身份验证
• 系统分配的托管标识身份验证
• 用户分配的托管标识身份验证
• 匿名身份验证

将 authenticationType 属性设置为 Basic。 除了前面部分所述的通用属性，还指定以下属性：

示例

{
    "name": "RESTLinkedService",
    "properties": {
        "type": "RestService",
        "typeProperties": {
            "authenticationType": "Basic",
            "url" : "<REST endpoint>",
            "userName": "<user name>",
            "password": {
                "type": "SecureString",
                "value": "<password>"
            }
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

将 authenticationType 属性设置为 AadServicePrincipal。 除了前面部分所述的通用属性，还指定以下属性：

示例 1：使用服务主体密钥身份验证

{
    "name": "RESTLinkedService",
    "properties": {
        "type": "RestService",
        "typeProperties": {
            "url": "<REST endpoint e.g. https://www.example.com/>",
            "authenticationType": "AadServicePrincipal",
            "servicePrincipalId": "<service principal id>",
            "servicePrincipalCredentialType": "ServicePrincipalKey",
            "servicePrincipalKey": {
                "value": "<service principal key>",
                "type": "SecureString"
            },
            "tenant": "<tenant info, e.g. microsoft.partner.onmschina.cn>",
            "aadResourceId": "<Azure AD resource URL e.g. https://management.core.chinacloudapi.cn>"
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

示例 2：使用服务主体证书身份验证

{
    "name": "RESTLinkedService",
    "properties": {
        "type": "RestService",
        "typeProperties": {
            "url": "<REST endpoint e.g. https://www.example.com/>",
            "authenticationType": "AadServicePrincipal",
            "servicePrincipalId": "<service principal id>",
            "servicePrincipalCredentialType": "ServicePrincipalCert",
            "servicePrincipalEmbeddedCert": {
                "type": "SecureString",
                "value": "<the base64 encoded certificate of your application registered in Microsoft Entra ID>"
            },
            "servicePrincipalEmbeddedCertPassword": {
                "type": "SecureString",
                "value": "<password of your certificate>"
            },
            "tenant": "<tenant info, e.g. microsoft.partner.onmschina.cn>",
            "aadResourceId": "<Azure AD resource URL e.g. https://management.core.chinacloudapi.cn>"
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

可以通过两个选项将服务主体证书保存到 Azure Key Vault 中：

• 选项 1

  1. 将服务主体证书转换为 base64 字符串。 有关详细信息，请参阅本文。

  2. 将 base64 字符串作为机密保存到 Azure Key Vault 中。

     

     

• 方法 2

  如果无法从 Azure Key Vault 下载证书，则可使用此模板将转换后的服务主体证书作为机密保存到 Azure Key Vault 中。

  

将 authenticationType 属性设置为 OAuth2ClientCredential。 除了前面部分所述的通用属性，还指定以下属性：

示例

{
    "name": "RESTLinkedService",
    "properties": {
        "type": "RestService",
        "typeProperties": {
            "url": "<REST endpoint e.g. https://www.example.com/>",
            "enableServerCertificateValidation": true,
            "authenticationType": "OAuth2ClientCredential",
            "clientId": "<client ID>",
            "clientSecret": {
                "type": "SecureString",
                "value": "<client secret>"
            },
            "tokenEndpoint": "<token endpoint>",
            "scope": "<scope>",
            "resource": "<resource>"
        }
    }
}

将 authenticationType 属性设置为 ManagedServiceIdentity。 除了前面部分所述的通用属性，还指定以下属性：

示例

{
    "name": "RESTLinkedService",
    "properties": {
        "type": "RestService",
        "typeProperties": {
            "url": "<REST endpoint e.g. https://www.example.com/>",
            "authenticationType": "ManagedServiceIdentity",
            "aadResourceId": "<AAD resource URL e.g. https://management.core.windows.net>"
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

将 authenticationType 属性设置为 ManagedServiceIdentity。 除了前面部分所述的通用属性，还指定以下属性：

示例

{
    "name": "RESTLinkedService",
    "properties": {
        "type": "RestService",
        "typeProperties": {
            "url": "<REST endpoint e.g. https://www.example.com/>",
            "authenticationType": "ManagedServiceIdentity",
            "aadResourceId": "<Azure AD resource URL e.g. https://management.core.chinacloudapi.cn>",
            "credential": {
                "referenceName": "credential1",
                "type": "CredentialReference"
            }    
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

此外，还可以配置身份验证请求标头，以及内置的身份验证类型。

示例：使用 API 密钥身份验证

{
    "name": "RESTLinkedService",
    "properties": {
        "type": "RestService",
        "typeProperties": {
            "url": "<REST endpoint>",
            "authenticationType": "Anonymous",
            "authHeaders": {
                "x-api-key": {
                    "type": "SecureString",
                    "value": "<API key>"
                }
            }
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

本部分提供 REST 数据集支持的属性列表。

有关可用于定义数据集的各部分和属性的完整列表，请参阅数据集和链接服务。

若要从 REST 复制数据，支持以下属性：

如果你在数据集中设置了 requestMethod、additionalHeaders、requestBody 和 paginationRules，我们仍按原样支持该数据集，但建议你以后在活动中使用新模型。

示例：

{
    "name": "RESTDataset",
    "properties": {
        "type": "RestResource",
        "typeProperties": {
            "relativeUrl": "<relative url>"
        },
        "schema": [],
        "linkedServiceName": {
            "referenceName": "<REST linked service name>",
            "type": "LinkedServiceReference"
        }
    }
}

本部分提供了 REST 源和接收器支持的属性列表。

有关可用于定义活动的各个部分和属性的完整列表，请参阅管道。

复制活动source部分支持以下属性：

示例 1：对分页使用 Get 方法

"activities":[
    {
        "name": "CopyFromREST",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<REST input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "RestSource",
                "additionalHeaders": {
                    "x-user-defined": "helloworld"
                },
                "paginationRules": {
                    "AbsoluteUrl": "$.paging.next"
                },
                "httpRequestTimeout": "00:01:00"
            },
            "sink": {
                "type": "<sink type>"
            }
        }
    }
]

示例 2：使用 Post 方法

"activities":[
    {
        "name": "CopyFromREST",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<REST input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "RestSource",
                "requestMethod": "Post",
                "requestBody": "<body for POST REST request>",
                "httpRequestTimeout": "00:01:00"
            },
            "sink": {
                "type": "<sink type>"
            }
        }
    }
]

复制活动接收器部分中支持以下属性：

REST 连接器作为接收器时适用于接受 JSON 的 REST API。 数据将采用 JSON 以下列模式发送。 根据需要，可以使用复制活动架构映射来重新调整源数据的形式，使之符合 REST API 预期的有效负载。

[
    { <data object> },
    { <data object> },
    ...
]

示例：

"activities":[
    {
        "name": "CopyToREST",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<REST output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "<source type>"
            },
            "sink": {
                "type": "RestSink",
                "requestMethod": "POST",
                "httpRequestTimeout": "00:01:40",
                "requestInterval": 10,
                "writeBatchSize": 10000,
                "httpCompressionType": "none",
            },
        }
    }
]

集成数据集和内联数据集的数据流中支持 REST。

可以设置 delete、insert、update 和 upsert 方法，以及要发送到 REST 接收器以进行 CRUD 操作的相对行数据。

请注意，先于接收器使用更改行转换，指示 ADF 要对 REST 接收器执行哪种类型的操作。 例如 insert、update、upsert、delete。

AlterRow1 sink(allowSchemaDrift: true,
	validateSchema: false,
	deletable:true,
	insertable:true,
	updateable:true,
	upsertable:true,
	rowRelativeUrl: 'periods',
	insertHttpMethod: 'PUT',
	deleteHttpMethod: 'DELETE',
	upsertHttpMethod: 'PUT',
	updateHttpMethod: 'PATCH',
	timeout: 30,
	requestFormat: ['type' -> 'json'],
	skipDuplicateMapInputs: true,
	skipDuplicateMapOutputs: true) ~> sink1

从 REST API 复制数据时，REST API 通常会将单个请求的响应有效负载大小限制在合理的数字以下；若要返回大量的数据，它会将结果拆分成多个页面，并要求调用方发送连续的请求来获取下一页结果。 一般情况下，一个页面的请求是动态的，由上一页响应中返回的信息构成。

此泛型 REST 连接器支持以下分页模式：

• 下一个请求的绝对或相对 URL = 当前响应正文中的属性值
• 下一个请求的绝对或相对 URL = 当前响应标头中的标头值
• 下一个请求的查询参数 = 当前响应正文中的属性值
• 下一个请求的查询参数 = 当前响应标头中的标头值
• 下一个请求的标头 = 当前响应正文中的属性值
• 下一个请求的标头 = 当前响应标头中的标头值

“分页规则”定义为包含一个或多个区分大小写的键值对的数据集中的字典。 该配置将用于从第二页开始生成请求。 当连接器收到 HTTP 状态代码 204（无内容），或者“paginationRules”中的任意 JSONPath 表达式返回 null 时，连接器将停止迭代。

分页规则中支持的键：

分页规则中支持的值：

本部分提供了一系列的分页规则设置示例。

此示例提供用于发送多个请求（其变量在 QueryParameters 中）的配置步骤。

多个请求：

baseUrl/api/now/table/incident?sysparm_limit=1000&sysparm_offset=0,
baseUrl/api/now/table/incident?sysparm_limit=1000&sysparm_offset=1000,
...... 
baseUrl/api/now/table/incident?sysparm_limit=1000&sysparm_offset=10000

步骤 1：在“基本 URL”或“相对 URL”中输入 sysparm_offset={offset}，如以下屏幕截图所示：

或

步骤 2：将“分页规则”设置为选项 1 或选项 2：

• 选项 1："QueryParameters.{offset}" : "RANGE:0:10000:1000"

• 选项 2："AbsoluteUrl.{offset}" : "RANGE:0:10000:1000"

此示例提供用于发送多个请求（其变量在 AbsoluteUrl 中）的配置步骤。

多个请求：

BaseUrl/api/now/table/t1
BaseUrl/api/now/table/t2
...... 
BaseUrl/api/now/table/t100

步骤 1：在“链接服务配置”页的“基本 URL”中或在“数据集连接”窗格的“相对 URL”中输入 {id}。

或

步骤 2：将“分页规则”设置为 "AbsoluteUrl.{id}" :"RANGE:1:100:1"。

此示例提供用于发送多个请求（其变量在 Headers 中）的配置步骤。

多个请求：
RequestUrl：https://example/table
请求 1：Header(id->0)
请求 2：Header(id->10)
......
请求 100：Header(id->100)

步骤 1：在“其他标头”中输入 {id}。

步骤 2：将“分页规则”设置为“Headers.{id}”:“RARNGE:0:100:10”。

此示例提供用于发送多个请求（其变量位于 AbsoluteUrl/QueryParameters/Headers 中，但结束变量未定义）的配置步骤。 对于不同的响应，示例 4.1-4.6 中显示了不同的结束条件规则设置。

多个请求：

Request 1: baseUrl/api/now/table/incident?sysparm_limit=1000&sysparm_offset=0, 
Request 2: baseUrl/api/now/table/incident?sysparm_limit=1000&sysparm_offset=1000,
Request 3: baseUrl/api/now/table/incident?sysparm_limit=1000&sysparm_offset=2000,
...... 

在此示例中遇到了两个响应：

响应 1：

{
    Data: [
        {key1: val1, key2: val2
        },
        {key1: val3, key2: val4
        }
    ]
}

响应 2：

{
    Data: [
        {key1: val5, key2: val6
        },
        {key1: val7, key2: val8
        }
    ]
}

步骤 1：按示例 1 设置“分页规则”的范围，并将范围末尾留空，其形式为 "AbsoluteUrl.{offset}": "RANGE:0::1000"。

步骤 2：根据不同的上一个响应设置不同的结束条件规则。 请参阅以下示例：

• 示例 4.1：当响应中的特定节点的值为空时，分页结束

  REST API 返回采用以下结构的上一个响应：

  {
      Data: []
  }
  

  将结束条件规则设置为 "EndCondition:$.data": "Empty"，以便在响应中的特定节点的值为空时结束分页。

  

• 示例 4.2：当响应中特定节点的值不存在时，分页结束

  REST API 返回采用以下结构的上一个响应：

  {}
  

  将结束条件规则设置为 "EndCondition:$.data": "NonExist"，以便在响应中特定节点的值不存在时结束分页。

  

• 示例 4.3：当响应中的特定节点的值存在时，分页结束

  REST API 返回采用以下结构的上一个响应：

  {
      Data: [
          {key1: val991, key2: val992
          },
          {key1: val993, key2: val994
          }
      ],
              Complete: true
  }
  

  将结束条件规则设置为 "EndCondition:$.Complete": "Exist"，以便在响应中的特定节点的值存在时结束分页。

  

• 示例 4.4：当响应中的特定节点的值是用户定义的常量值时，分页结束

  REST API 返回采用以下结构的响应：

  {
      Data: [
          {key1: val1, key2: val2
          },
          {key1: val3, key2: val4
          }
      ],
              Complete: false
  }
  

  ......

  上一个响应采用以下结构：

  {
      Data: [
          {key1: val991, key2: val992
          },
          {key1: val993, key2: val994
          }
      ],
              Complete: true
  }
  

  将结束条件规则设置为 "EndCondition:$.Complete": "Const:true"，以便在响应中的特定节点的值是用户定义的常量值时结束分页。

  

• 示例 4.5：当响应中的标头键的值是用户定义的常量值时，分页结束

  REST API 响应中的标头键按照下面的结构显示：

  响应头 1：header(Complete->0)
  ......
  上一个响应头：header(Complete->1)
  

  将结束条件规则设置为 "EndCondition:headers.Complete": "Const:1"，以便在响应中的标头键的值等于用户定义的常量值时结束分页。

  

• 示例 4.6：当响应头中存在该键时，分页结束

  REST API 响应中的标头键按照下面的结构显示：

  响应头 1：header()
  ......
  上一个响应头：header(CompleteTime->20220920)
  

  将结束条件规则设置为 "EndCondition:headers.CompleteTime": "Exist"，以便在响应头中存在该键时结束分页。

  

此示例提供了在未使用范围规则时发送多个请求的配置步骤。 可以参考示例 4.1-4.6 来设置结束条件，以避免无限请求。 REST API 返回采用以下结构的响应，在此情况下，下一页的 URL 将在 paging.next 中表示。

{
    "data": [
        {
            "created_time": "2017-12-12T14:12:20+0000",
            "name": "album1",
            "id": "1809938745705498_1809939942372045"
        },
        {
            "created_time": "2017-12-12T14:14:03+0000",
            "name": "album2",
            "id": "1809938745705498_1809941802371859"
        },
        {
            "created_time": "2017-12-12T14:14:11+0000",
            "name": "album3",
            "id": "1809938745705498_1809941879038518"
        }
    ],
    "paging": {
        "cursors": {
            "after": "MTAxNTExOTQ1MjAwNzI5NDE=",
            "before": "NDMyNzQyODI3OTQw"
        },
        "previous": "https://graph.facebook.com/me/albums?limit=25&before=NDMyNzQyODI3OTQw",
        "next": "https://graph.facebook.com/me/albums?limit=25&after=MTAxNTExOTQ1MjAwNzI5NDE="
    }
}
...

上一个响应是：

{
    "data": [],
    "paging": {
        "cursors": {
            "after": "MTAxNTExOTQ1MjAwNzI5NDE=",
            "before": "NDMyNzQyODI3OTQw"
        },
        "previous": "https://graph.facebook.com/me/albums?limit=25&before=NDMyNzQyODI3OTQw",
        "next": "Same with Last Request URL"
    }
}

步骤 1：将“分页规则”设置为 "AbsoluteUrl": "$.paging.next"。

步骤 2：如果上一个响应中的 next 始终与上一个请求 URL 相同并且不为空，则会发送无限请求。 结束条件可用于避免无限请求。 因此，请参阅示例 4.1-4.6 来设置结束条件规则。

设置 MaxRequestNumber 以避免无限请求，如以下屏幕截图所示：

后端将基于标头中的 RFC 5988 样式链接自动获取下一个 URL。

此示例说明当下一个请求 URL 来自响应正文时，如何在映射数据流中设置分页规则与结束条件规则。

响应架构如下所示：

应按以下屏幕截图所示设置分页规则：

默认情况下，当 body.{@odata.nextLink}** 为 null 或为空时，将停止分页。

但如果上一个响应正文中 @odata.nextLink 的值等于上一个请求 URL，则会导致无限循环。 若要避免这种情况，请定义结束条件规则。

• 如果上一个响应中的“值”为空，则可以按下面所示设置结束条件规则：

  

• 如果响应头中完成键的值等于 true 指示分页结束，则可以按下面所示设置结束条件规则：

  

此示例演示当下一个请求 URL 包含在响应正文中时，如何在复制活动中设置分页规则。

响应架构如下所示：

应按以下屏幕截图所示设置分页规则：

此示例说明当响应格式为 XML 并且下一个请求 URL 来自响应正文时，如何在映射数据流中设置分页规则。 如以下屏幕截图所示，第一个 URL 是 https://<user>.dfs.core.chinacloudapi.cn/bugfix/test/movie_1.xml

响应架构如下所示：

分页规则语法与示例 8 中的语法相同，应按以下示例所示进行设置：

可以使用此 REST 连接器将 REST API JSON 响应按原样导出到各种基于文件的存储。 若要实现这种架构不可知的复制，请跳过数据集中的“结构”（也称为“架构”）节和复制活动中的架构映射。

若要将数据从 REST 终结点复制到表格接收器，请参阅架构映射。

有关 Azure 数据工厂中复制活动支持用作源和接收器的数据存储的列表，请参阅支持的数据存储和格式。