结构数据工厂提供了一组可靠的 API,使用户能够高效地自动执行和管理其数据流。 这些 API 允许与各种数据源和服务无缝集成,使用户能够以编程方式创建、更新和监视其数据工作流。 API 支持各种作,包括数据流 CRUD(创建、读取、更新和删除)、计划和监视,使用户能够更轻松地管理其数据集成过程。
数据流的 API 用例
Fabric 数据工厂中的数据流 API 可用于多种场景:
- 自动部署:使用 CI/CD 做法自动跨不同环境(开发、测试、生产)部署数据流。
- 监视和警报:设置自动监视和警报系统,以跟踪数据流的状态,并在出现故障或性能问题时接收通知。
- 数据集成:将来自多个源(例如数据库、数据湖和云服务)的数据集成到统一的数据流中,以便进行处理和分析。
- 错误处理:实现自定义错误处理和重试机制,以确保数据流顺利运行,并从故障中恢复。
了解 API
若要有效地使用 Fabric Data Factory 中用于数据流的 API,必须了解关键概念和组件:
- 终结点:API 终结点提供对各种数据流作的访问权限,例如创建、更新和删除数据流。
- 身份验证:使用 OAuth 或 API 密钥等身份验证机制保护对 API 的访问。
- 请求和响应:了解 API 请求和响应的结构,包括所需的参数和预期输出。
- 速率限制:请注意对 API 使用情况施加的速率限制,以避免超过允许的请求数。
CRUD 支持
CRUD 代表“创建”、“读取”、“更新”和“删除”,这些是可对数据执行的四项基本操作。 在 Fabric 数据工厂中,通过 Fabric API 支持 CRUD 操作。 这些 API 允许用户以编程方式管理其数据流。 以下是有关 CRUD 支持的一些要点:
- 创建:使用 API 创建新的数据流。 这涉及到定义数据流结构、指定数据源、转换和目标。
- 读取:检索有关现有数据流的信息。 这包括有关其配置、状态和执行历史记录的详细信息。
- 更新:更新现有数据流。 这可能涉及修改数据流结构、更改数据源或更新转换逻辑。
- 删除:删除不再需要的数据流。 这有助于管理和清理资源。
可在 Microsoft Fabric REST API 文档中找到 Microsoft Fabric REST API 的主要联机参考文档。
数据流生命周期
使用数据流时,请务必了解数据流的生命周期,以确保顺利且高效的数据集成过程。 数据流生命周期的关键阶段包括
- 定义: 这是创建或修改数据流定义的初始阶段。 这涉及到指定数据源、转换和定义。
- 发布: 修改数据流的定义后,需要调用 发布 作。 这是一个关键步骤,因为它保存对数据流定义所做的更改并使其可供执行。
- 刷新: 发布数据流后,可以调用 刷新 作。 这会触发数据流运行、从指定源拉取数据、应用定义的转换并将数据加载到目标。
开始使用数据流的公共 API
在本部分中,我们将介绍以下所有主题:
- 获取授权令牌
- 创建数据流
- 使用定义创建数据流
- 获取数据流
- 使用定义获取数据流
- 更新数据流
- 使用定义更新数据流
- 删除数据流
- 运行按需数据流作业(刷新)
- 运行按需数据流发布作业
- 获取数据流作业实例
- 取消数据流作业实例
获取授权令牌
你需要拥有所有 REST API 调用的持有者令牌,并且可以使用以下选项之一获取它:
MSAL.Net
Fabric API 快速入门 - Microsoft Fabric REST API
使用 MSAL.Net 获取具有以下作用域的 Fabric 服务的 Microsoft Entra ID 令牌:Workspace.ReadWrite.All、Item.ReadWrite.All。 有关使用 MSAL.Net 获取令牌的详细信息,请参阅令牌获取 - 适用于 .NET 的 Microsoft 身份验证库。
粘贴之前复制的应用程序(客户端)ID,并将其粘贴到 ClientId 变量。
Fabric 门户
登录到要测试的租户的 Fabric 门户,然后按 F12 进入浏览器的开发人员模式。 在控制台中运行以下内容:
powerBIAccessToken
复制令牌并将其粘贴到 ClientId 变量中。
创建数据流
在指定的工作区中创建数据流。
示例请求
URI: POST [https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items](https://api.fabric.microsoft.com/v1/workspaces/%7bworkspaceId%7d/items)
Headers:
{
"Authorization": "\<bearer-token>",
"Content-Type": "application/json"
}
Payload:
{
"displayName": "My dataflow",
"description": "My dataflow description",
"type": "Dataflow"
}
示例响应
{
"id": "\<itemId>",
"type": "Dataflow",
"displayName": "My dataflow",
"description": "My dataflow description",
"workspaceId": "\<workspaceId>"
}
使用定义创建数据流
在指定的工作区中创建具有 base64 定义的数据流。
示例 Mashup 文档
{
"editingSessionMashup": {
"mashupName": "",
"mashupDocument": "section Section1;\r\nshared Query = let\n Source = Table.FromRows(\n {\n {1, \"Bob\", \"123-4567\"},\n {2, \"Jim\", \"987-6543\"}\n },\n {\"CustomerID\", \"Name\", \"Phone\"})\nin\n Source;\r\n",
"queryGroups": [],
"documentLocale": "en-US",
"gatewayObjectId": null,
"queriesMetadata": null,
"connectionOverrides": [],
"trustedConnections": null,
"useHostConnectionProvider": false,
"fastCombine": false,
"allowNativeQueries": true,
"allowedModules": null,
"skipAutomaticTypeAndHeaderDetection": false,
"disableAutoAnonymousConnectionUpsert": null,
"hostProperties": {
"DataflowRefreshOutputFileFormat": "Parquet",
"EnableDateTimeFieldsForStaging": "true",
"EnablePublishWithoutLoadedQueries": "true"
},
"defaultOutputDestinationConfiguration": null,
"stagingDefinition": null
}
}
使用 Base64 编码和解码对 JSON 进行编码。
确保未选中“执行 URL 安全编码”复选框。
将您的 Base64 编码的请求负载响应粘贴到负载定义中
示例请求
URI: POST [https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items](https://api.fabric.microsoft.com/v1/workspaces/%7bworkspaceId%7d/items)
Headers:
{
"Authorization": "\<bearer-token>",
"Content-Type": "application/json"
}
Payload:
{
"displayName": " My dataflow",
"description": "My dataflow description",
"type": "Dataflow",
"definition": {
"parts": [
{
"path": "dataflow-content.json",
"payload": "ewogICAgImVkaXRpbmdTZXNzaW9uTWFzaHVwIjogewogICAgICAgICJtYXNodXBOYW1lIjogIiIsCiAgICAgICAgIm1hc2h1cERvY3VtZW50IjogInNlY3Rpb24gU2VjdGlvbjE7XHJcbnNoYXJlZCBRdWVyeSA9IGxldFxuIFNvdXJjZSA9IFRhYmxlLkZyb21Sb3dzKFxuIHtcbiB7MSwgXCJCb2JcIiwgXCIxMjMtNDU2N1wifSxcbiB7MiwgXCJKaW1cIiwgXCI5ODctNjU0M1wifVxuIH0sXG4ge1wiQ3VzdG9tZXJJRFwiLCBcIk5hbWVcIiwgXCJQaG9uZVwifSlcbmluXG4gU291cmNlO1xyXG4iLAogICAgICAgICJxdWVyeUdyb3VwcyI6IFtdLAogICAgICAgICJkb2N1bWVudExvY2FsZSI6ICJlbi1VUyIsCiAgICAgICAgImdhdGV3YXlPYmplY3RJZCI6IG51bGwsCiAgICAgICAgInF1ZXJpZXNNZXRhZGF0YSI6IG51bGwsCiAgICAgICAgImNvbm5lY3Rpb25PdmVycmlkZXMiOiBbXSwKICAgICAgICAidHJ1c3RlZENvbm5lY3Rpb25zIjogbnVsbCwKICAgICAgICAidXNlSG9zdENvbm5lY3Rpb25Qcm92aWRlciI6IGZhbHNlLAogICAgICAgICJmYXN0Q29tYmluZSI6IGZhbHNlLAogICAgICAgICJhbGxvd05hdGl2ZVF1ZXJpZXMiOiB0cnVlLAogICAgICAgICJhbGxvd2VkTW9kdWxlcyI6IG51bGwsCiAgICAgICAgInNraXBBdXRvbWF0aWNUeXBlQW5kSGVhZGVyRGV0ZWN0aW9uIjogZmFsc2UsCiAgICAgICAgImRpc2FibGVBdXRvQW5vbnltb3VzQ29ubmVjdGlvblVwc2VydCI6IG51bGwsCiAgICAgICAgImhvc3RQcm9wZXJ0aWVzIjogewogICAgICAgICAgICAiRGF0YWZsb3dSZWZyZXNoT3V0cHV0RmlsZUZvcm1hdCI6ICJQYXJxdWV0IiwKICAgICAgICAgICAgIkVuYWJsZURhdGVUaW1lRmllbGRzRm9yU3RhZ2luZyI6ICJ0cnVlIiwKICAgICAgICAgICAgIkVuYWJsZVB1Ymxpc2hXaXRob3V0TG9hZGVkUXVlcmllcyI6ICJ0cnVlIgogICAgICAgIH0sCiAgICAgICAgImRlZmF1bHRPdXRwdXREZXN0aW5hdGlvbkNvbmZpZ3VyYXRpb24iOiBudWxsLAogICAgICAgICJzdGFnaW5nRGVmaW5pdGlvbiI6IG51bGwKICAgIH0KfQ==",
"payloadType": "InlineBase64"
}
]
}
}
示例响应
{
"id": "\<itemId>",
"type": "Dataflow",
"displayName": "My dataflow",
"description": "My dataflow description",
"workspaceId": "\<workspaceId>"
}
获取数据流
返回指定数据流的属性。
示例请求
URI: GET [https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items/{itemId}](https://api.fabric.microsoft.com/v1/workspaces/%7bworkspaceId%7d/items/%7bitemId%7d)
Headers:
{
"Authorization": "\<bearer-token>"
}
示例响应
{
"id": "\<itemId>",
"type": "Dataflow",
"displayName": "My dataflow",
"description": "My dataflow description",
"workspaceId": "\<workspaceId>"
}
通过定义获取数据流
返回数据流项定义。
示例请求
URI: POST [https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items/{itemId}/getDefinition](https://api.fabric.microsoft.com/v1/workspaces/%7bworkspaceId%7d/items/%7bitemId%7d/getDefinition)
Headers:
{
"Authorization": "\<bearer-token>"
}
示例响应
{
"definition": {
"parts": [
{
"path": "dataflow-content.json",
"payload": " ewogICAgImVkaXRpbmdTZXNzaW9uTWFzaHVwIjogewogICAgICAgICJtYXNodXBOYW1lIjogIiIsCiAgICAgICAgIm1hc2h1cERvY3VtZW50IjogInNlY3Rpb24gU2VjdGlvbjE7XHJcbnNoYXJlZCBRdWVyeSA9IGxldFxuIFNvdXJjZSA9IFRhYmxlLkZyb21Sb3dzKFxuIHtcbiB7MSwgXCJCb2JcIiwgXCIxMjMtNDU2N1wifSxcbiB7MiwgXCJKaW1cIiwgXCI5ODctNjU0M1wifVxuIH0sXG4ge1wiQ3VzdG9tZXJJRFwiLCBcIk5hbWVcIiwgXCJQaG9uZVwifSlcbmluXG4gU291cmNlO1xyXG4iLAogICAgICAgICJxdWVyeUdyb3VwcyI6IFtdLAogICAgICAgICJkb2N1bWVudExvY2FsZSI6ICJlbi1VUyIsCiAgICAgICAgImdhdGV3YXlPYmplY3RJZCI6IG51bGwsCiAgICAgICAgInF1ZXJpZXNNZXRhZGF0YSI6IG51bGwsCiAgICAgICAgImNvbm5lY3Rpb25PdmVycmlkZXMiOiBbXSwKICAgICAgICAidHJ1c3RlZENvbm5lY3Rpb25zIjogbnVsbCwKICAgICAgICAidXNlSG9zdENvbm5lY3Rpb25Qcm92aWRlciI6IGZhbHNlLAogICAgICAgICJmYXN0Q29tYmluZSI6IGZhbHNlLAogICAgICAgICJhbGxvd05hdGl2ZVF1ZXJpZXMiOiB0cnVlLAogICAgICAgICJhbGxvd2VkTW9kdWxlcyI6IG51bGwsCiAgICAgICAgInNraXBBdXRvbWF0aWNUeXBlQW5kSGVhZGVyRGV0ZWN0aW9uIjogZmFsc2UsCiAgICAgICAgImRpc2FibGVBdXRvQW5vbnltb3VzQ29ubmVjdGlvblVwc2VydCI6IG51bGwsCiAgICAgICAgImhvc3RQcm9wZXJ0aWVzIjogewogICAgICAgICAgICAiRGF0YWZsb3dSZWZyZXNoT3V0cHV0RmlsZUZvcm1hdCI6ICJQYXJxdWV0IiwKICAgICAgICAgICAgIkVuYWJsZURhdGVUaW1lRmllbGRzRm9yU3RhZ2luZyI6ICJ0cnVlIiwKICAgICAgICAgICAgIkVuYWJsZVB1Ymxpc2hXaXRob3V0TG9hZGVkUXVlcmllcyI6ICJ0cnVlIgogICAgICAgIH0sCiAgICAgICAgImRlZmF1bHRPdXRwdXREZXN0aW5hdGlvbkNvbmZpZ3VyYXRpb24iOiBudWxsLAogICAgICAgICJzdGFnaW5nRGVmaW5pdGlvbiI6IG51bGwKICAgIH0KfQ=="
"payloadType": "InlineBase64"
},
{
"path": ".platform",
"payload": "ewogICIkc2NoZW1hIjogImh0dHBzOi8vZGV2ZWxvcGVyLm1pY3Jvc29mdC5jb20vanNvbi1zY2hlbWFzL2ZhYnJpYy9naXRJbnRlZ3JhdGlvbi9wbGF0Zm9ybVByb3BlcnRpZXMvMi4wLjAvc2NoZW1hLmpzb24iLAogICJtZXRhZGF0YSI6IHsKICAgICJ0eXBlIjogIkRhdGFmbG93IiwKICAgICJkaXNwbGF5TmFtZSI6ICJEYXRhZmxvdyAzIgogIH0sCiAgImNvbmZpZyI6IHsKICAgICJ2ZXJzaW9uIjogIjIuMCIsCiAgICAibG9naWNhbElkIjogIjAwMDAwMDAwLTAwMDAtMDAwMC0wMDAwLTAwMDAwMDAwMDAwMCIKICB9Cn0=",
"payloadType": "InlineBase64"
}
]
}
}
来自 API 的响应
{
'definition':
{
'parts': [
{
'path': 'queryMetadata.json', 'payload': '<super long entry>', 'payloadType': 'InlineBase64'
},
{
'path': 'mashup.pq', 'payload': '<super long entry>', 'payloadType': 'InlineBase64'
},
{
'path': '.platform', 'payload': '<super long entry>', 'payloadType': 'InlineBase64'
}
]
}
}
更新数据流
更新数据流的属性。
示例请求
URI: PATCH [https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items/{itemId}](https://api.fabric.microsoft.com/v1/workspaces/%7bworkspaceId%7d/items/%7bitemId%7d)
Headers:
{
"Authorization": "\<bearer-token>",
"Content-Type": "application/json"
}
Payload:
{
"displayName": "My dataflow updated",
"description": "My dataflow description updated",
"type": "Dataflow"
}
示例响应
{
"id": "\<itemId>",
"type": "Dataflow",
"displayName": "My dataflow updated",
"description": "My dataflow description updated",
"workspaceId": "\<workspaceId>"
}
用定义更新数据流
更新数据流项定义。
示例请求
URI: POST [https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items/{itemId}/updateDefinition](https://api.fabric.microsoft.com/v1/workspaces/%7bworkspaceId%7d/items/%7bitemId%7d/updateDefinition)
Headers:
{
"Authorization": "\<bearer-token>",
"Content-Type": "application/json"
}
Payload:
{
"displayName": " My dataflow",
"type": "Dataflow",
"definition": {
"parts": [
{
"path": "dataflow-content.json",
"payload": " ewogICAgImVkaXRpbmdTZXNzaW9uTWFzaHVwIjogewogICAgICAgICJtYXNodXBOYW1lIjogIiIsCiAgICAgICAgIm1hc2h1cERvY3VtZW50IjogInNlY3Rpb24gU2VjdGlvbjE7XHJcbnNoYXJlZCBRdWVyeSA9IGxldFxuIFNvdXJjZSA9IFRhYmxlLkZyb21Sb3dzKFxuIHtcbiB7MSwgXCJCb2JcIiwgXCIxMjMtNDU2N1wifSxcbiB7MiwgXCJKaW1cIiwgXCI5ODctNjU0M1wifVxuIH0sXG4ge1wiQ3VzdG9tZXJJRFwiLCBcIk5hbWVcIiwgXCJQaG9uZVwifSlcbmluXG4gU291cmNlO1xyXG4iLAogICAgICAgICJxdWVyeUdyb3VwcyI6IFtdLAogICAgICAgICJkb2N1bWVudExvY2FsZSI6ICJlbi1VUyIsCiAgICAgICAgImdhdGV3YXlPYmplY3RJZCI6IG51bGwsCiAgICAgICAgInF1ZXJpZXNNZXRhZGF0YSI6IG51bGwsCiAgICAgICAgImNvbm5lY3Rpb25PdmVycmlkZXMiOiBbXSwKICAgICAgICAidHJ1c3RlZENvbm5lY3Rpb25zIjogbnVsbCwKICAgICAgICAidXNlSG9zdENvbm5lY3Rpb25Qcm92aWRlciI6IGZhbHNlLAogICAgICAgICJmYXN0Q29tYmluZSI6IGZhbHNlLAogICAgICAgICJhbGxvd05hdGl2ZVF1ZXJpZXMiOiB0cnVlLAogICAgICAgICJhbGxvd2VkTW9kdWxlcyI6IG51bGwsCiAgICAgICAgInNraXBBdXRvbWF0aWNUeXBlQW5kSGVhZGVyRGV0ZWN0aW9uIjogZmFsc2UsCiAgICAgICAgImRpc2FibGVBdXRvQW5vbnltb3VzQ29ubmVjdGlvblVwc2VydCI6IG51bGwsCiAgICAgICAgImhvc3RQcm9wZXJ0aWVzIjogewogICAgICAgICAgICAiRGF0YWZsb3dSZWZyZXNoT3V0cHV0RmlsZUZvcm1hdCI6ICJQYXJxdWV0IiwKICAgICAgICAgICAgIkVuYWJsZURhdGVUaW1lRmllbGRzRm9yU3RhZ2luZyI6ICJ0cnVlIiwKICAgICAgICAgICAgIkVuYWJsZVB1Ymxpc2hXaXRob3V0TG9hZGVkUXVlcmllcyI6ICJ0cnVlIgogICAgICAgIH0sCiAgICAgICAgImRlZmF1bHRPdXRwdXREZXN0aW5hdGlvbkNvbmZpZ3VyYXRpb24iOiBudWxsLAogICAgICAgICJzdGFnaW5nRGVmaW5pdGlvbiI6IG51bGwKICAgIH0KfQ==",
"payloadType": "InlineBase64"
}
]
}
}
示例响应
200 OK
删除数据流
删除指定的数据流。
示例请求
URI: DELETE [https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items/{itemId}](https://api.fabric.microsoft.com/v1/workspaces/%7bworkspaceId%7d/items/%7bitemId%7d)
Headers:
{
"Authorization": "\<bearer-token>"
}
示例响应
200 OK
运行按需数据流作业(刷新)
为刷新实例运行按需数据流作业。
示例请求
URI: POST [https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items/{itemId}/jobs/instances?jobType=Refresh](https://api.fabric.microsoft.com/v1/workspaces/%7bworkspaceId%7d/items/%7bitemId%7d/jobs/instances?jobType=Refresh)
Headers:
{
"Authorization": "\<bearer-token>"
}
Payload:
{
"executionData": {
"DataflowName": "Dataflow",
"OwnerUserPrincipalName": "\<name@email.com>",
"OwnerUserObjectId": "\<ObjectId>"
}
}
示例响应
202 Accepted
[
{
"id": "\<jobId>",
"itemId": "\<dataflowItemId>",
"jobType": "Refresh",
"invokeType": "OnDemand",
"status": "Accepted",
"jobInstanceId": "\<uniqueJobInstanceId>",
"rootActivityId": "\<rootActivityId>",
"startTimeUtc": "2025-01-30T11:10:50Z",
"endTimeUtc": null,
"failureReason": null
}
]
运行按需数据流发布作业
为发布实例运行按需数据流作业。
示例请求
URI: POST [https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items/{itemId}/jobs/instances?jobType=Publish](https://api.fabric.microsoft.com/v1/workspaces/%7bworkspaceId%7d/items/%7bitemId%7d/jobs/instances?jobType=Publish)
Headers:
{
"Authorization": "\<bearer-token>"
}
Payload:
{
"executionData": {
"DataflowName": "Dataflow",
"OwnerUserPrincipalName": "\<name@email.com>",
"OwnerUserObjectId": "\<ObjectId>"
}
}
示例响应
202 Accepted
[
{
"id": "\<jobId>",
"itemId": "\<dataflowItemId>",
"jobType": "Publish",
"invokeType": "OnDemand",
"status": "Accepted",
"jobInstanceId": "\<uniqueJobInstanceId>",
"rootActivityId": "\<rootActivityId>",
"startTimeUtc": "2025-01-30T11:10:50Z",
"endTimeUtc": null,
"failureReason": null
}
]
获取数据流作业实例
获取单一数据流的作业实例。
示例请求
URI: GET [https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items/{itemId}/jobs/instances/{jobInstanceId}](https://api.fabric.microsoft.com/v1/workspaces/%7bworkspaceId%7d/items/%7bitemId%7d/jobs/instances/%7bjobInstanceId%7d)
Headers:
{
"Authorization": "\<bearer-token>"
}
示例响应
{
"id": "\<id>",
"itemId": "<itemId?",
"jobType": "Refresh",
"invokeType": "Manual",
"status": "Completed",
"rootActivityId": "\<rootActivityId>",
"startTimeUtc": "2023-08-22T06:35:00.7812154",
"endTimeUtc": "2023-08-22T06:35:00.8033333",
"failureReason": null
}
取消数据流作业实例
取消数据流的作业实例
示例请求
URI: POST [https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items/{itemId}/jobs/instances/{jobInstanceId}/cancel](https://api.fabric.microsoft.com/v1/workspaces/%7bworkspaceId%7d/items/%7bitemId%7d/jobs/instances/%7bjobInstanceId%7d/cancel)
Headers:
{
"Authorization": "\<bearer-token>"
}
示例响应
Location: https://api.fabric.microsoft.com/v1/workspaces/\<worksapceId>/items/\<itemId>/jobs/instances/\<jobInstanceId>
Retry-After: 60
当前限制
- 目前不支持服务主体身份验证。
- 如果筛选数据流项类型,“获取项”和“列表项访问详细信息”不会返回正确的信息。
- 如果未指定类型,它将返回 数据流 Gen2(CI/CD、预览版) -新的数据流 Gen2(CI/CD 和 GIT 支持)。
- 运行 API 是可调用的,但实际运行永远不会成功。