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

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

本文概述了如何使用Azure 数据工厂中的复制活动从/向 REST 终结点复制数据。 本文基于 Azure 数据工厂 中的Copy Activity，并概述了复制活动。

该 REST 连接器、 HTTP 连接器和 Web 表连接器 之间的区别有：

• REST connector 连接器专门支持从 RESTful API 复制数据。
• HTTP 连接器 是通用的，用于从任意 HTTP 端点获取数据，例如下载文件。 在这个 REST 连接器之前，你可能会用 HTTP 连接器来复制 RESTful API 的数据，虽然支持但功能不如 REST 连接器。
• Web 表连接器用于从 HTML 网页中提取表内容。

支持的功能

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

（1） Azure集成运行时 （2） 自承载集成运行时

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

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

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

先决条件

如果数据存储位于本地网络、Azure虚拟网络或 Amazon 虚拟私有云中，则需要配置自承载集成运行时以连接到它。

如果数据存储是托管的云数据服务，则可以使用Azure Integration Runtime。 如果访问仅限于防火墙规则中批准的 IP，则可以将 Azure Integration Runtime IP 添加到允许列表。

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

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

开始

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

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

使用 UI 创建 REST 链接服务

使用以下步骤在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 密钥保管库

可以使用两个选项在 Azure 密钥保管库 中保存服务主体证书：

• 选项 1

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

  2. 将 base64 字符串另存为Azure 密钥保管库中的机密。

     

     

• 方法 2

  如果无法从 Azure 密钥保管库 下载证书，则可以使用此 template 将转换的服务主体证书保存为Azure 密钥保管库中的机密。

  

使用 OAuth2 客户端凭证认证

将 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 复制操作仍然支持它们 as-is，尽管你今后应在活动中使用新模型。

示例：

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

复制活动属性

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

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

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 作为接收器

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

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 操作。

示例数据流脚本

注意在汇盘前使用了 alter 行变换，用来指示 Data Factory 对你的 REST 汇执行什么类型的操作。 这个操作可以是插入、更新、更新或删除。

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（无内容）或任何 JSONPath 表达式返回 paginationRules NULL 时，连接器停止迭代。

分页规则中支持的键：

分页规则中支持的值：

分页规则示例

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

示例 1：QueryParameters 中的变量

此示例提供用于发送多个请求（其变量在 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"

示例2：AbsoluteUrl中的变量

此示例提供用于发送多个请求（其变量在 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"。

示例3：头部中的变量

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

多个请求：

RequestUrl: https://example/table
Request 1: Header(id->0)
Request 2: Header(id->10)
......
Request 100: Header(id->100)

步骤 1：在{id}中输入。

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

示例4：变量在AbsoluteUrl/QueryParameters/Headers中，结尾变量未预定义，结尾条件基于响应

此示例提供配置步骤来发送多个请求，其变量位于 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：当响应中的头部键值等于用户定义的const值时，分页结束

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

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

  将结束条件规则设置为 “EndCondition：headers.完成“：”Const：1“ ，当响应中的头部键值等于用户自定义的const值时，结束分页。

  

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

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

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

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

  

示例5：设置终端条件以避免在未定义范围规则时出现无尽请求

此示例提供了在不使用范围规则时发送多个请求的配置步骤。 可以参考示例 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来设定终结条件规则。

示例6：设置最大请求数以避免无止境请求

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

示例7：RFC 5988分页规则默认支持

后端会自动根据头部中的RFC 5988风格链接获得下一个URL。

示例 8a：在映射数据流中使用分页时，下一个请求 URL 位于响应正文中

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

响应架构如下所示：

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

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

但如果最后一个响应主体中的 @odata.nextLink 值等于最后一个请求URL，就会导致无休止的循环。 若要避免这种情况，请定义结束条件规则。

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

  

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

  

示例 8b：在复制活动中使用分页时，下一个请求 URL 位于响应正文中

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

响应架构如下所示：

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

示例 9：在映射数据流中使用分页时，响应格式为 XML 并且下一个请求 URL 来自响应正文

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

响应架构如下所示：

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

按原样导出 JSON 响应

可以使用 REST 连接器将 REST API 的 JSON 响应 as-is 导出到各种基于文件的存储系统（接收器）。 要实现这种模式无关的复制行为，请使用默认模式映射（不要在复制活动的映射标签页中定义任何映射）。

模式映射

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

相关内容

有关 Azure 数据工厂 中复制活动支持作为源和汇聚点的数据存储的列表，请参阅 支持的数据存储和格式。