使用Flow Service API为协议源创建数据流
本教程介绍从协议源检索数据以及使用Flow Service API将这些数据引入平台的步骤。
NOTE
- 要创建数据流，您必须已具有包含协议源的有效基本连接ID。 如果您没有此ID，请查看源概述，了解可创建基本连接的协议源的列表。
快速入门
本教程还要求您实际了解Adobe Experience Platform的以下组件：
-
Experience Data Model (XDM) System：Experience Platform用于组织客户体验数据的标准化框架。
- 架构组合的基础知识：了解XDM架构的基本构建块，包括架构组合中的关键原则和最佳实践。
- 架构注册表开发人员指南：包含成功执行对架构注册表API的调用所需了解的重要信息。 这包括您的
{TENANT_ID}、“容器”的概念以及发出请求所需的标头（请特别注意“接受”标头及其可能的值）。
-
Catalog Service：目录是Experience Platform中数据位置和历程的记录系统。
-
Batch ingestion：批量摄取API允许您将数据作为批处理文件摄取到Experience Platform中。
-
沙盒：Experience Platform提供了将单个Platform实例划分为多个单独的虚拟环境的虚拟沙盒，以帮助开发和改进数字体验应用程序。
使用平台API
有关如何成功调用平台API的信息，请参阅平台API快速入门指南。
创建源连接 source
您可以通过向Flow Service API发出POST请求来创建源连接。 源连接由连接ID、源数据文件的路径以及连接规范ID组成。
要创建源连接，还必须为数据格式属性定义一个枚举值。
为基于文件的连接器使用以下枚举值：
数据格式
枚举值
已分隔
delimitedJSON
jsonParquet
parquet对于所有基于表的连接器，将该值设置为tabular。
API格式
POST /sourceConnections
请求
curl -X POST \
'https://platform.adobe.io/data/foundation/flowservice/sourceConnections' \
-H 'Authorization: Bearer {ACCESS_TOKEN}' \
-H 'x-api-key: {API_KEY}' \
-H 'x-gw-ims-org-id: {ORG_ID}' \
-H 'x-sandbox-name: {SANDBOX_NAME}' \
-H 'Content-Type: application/json' \
-d '{
"name": "Generic OData source connection",
"baseConnectionId": "a5c6b647-e784-4b58-86b6-47e784ab580b",
"description": "Generic OData source connection",
"data": {
"format": "tabular",
},
"params": {
"tableName": "Orders",
"columns": [
{
"name": "OrderID",
"type": "integer",
"xdm": {
"type": "integer",
"minimum": -2147483648,
"maximum": 2147483647
}
},
{
"name": "CustomerID",
"type": "string",
"xdm": {
"type": "string"
}
},
{
"name": "OrderDate",
"type": "string",
"meta:xdmType": "date-time",
"xdm": {
"type": "string",
"format": "date-time"
}
},
{
"name": "ShippedDate",
"type": "string",
"meta:xdmType": "date-time",
"xdm": {
"type": "string",
"format": "date-time"
}
}
]
},
"connectionSpec": {
"id": "8e6b41a8-d998-4545-ad7d-c6a9fff406c3",
"version": "1.0"
}
}'
属性
描述
baseConnectionId协议应用程序的连接ID
params.path源文件的路径。
connectionSpec.id协议应用程序的连接规范ID。
响应
成功的响应返回新创建的源连接的唯一标识符(id)。 在后续步骤中，创建目标连接时需要此ID。
{
"id": "0a768941-ddfb-499d-b689-41ddfbf99db0",
"etag": "\"8f00753e-0000-0200-0000-5e8a547d0000\""
}
创建目标XDM架构 target-schema
为了在Platform中使用源数据，必须创建目标架构，以根据您的需求构建源数据。 然后，使用目标架构创建包含源数据的Platform数据集。
通过向架构注册表API执行POST请求，可以创建目标XDM架构。
有关如何创建目标XDM架构的详细步骤，请参阅有关使用API 创建架构的教程。
创建目标数据集 target-dataset
可以通过向目录服务API执行POST请求，在有效负载中提供目标架构的ID来创建目标数据集。
有关如何创建目标数据集的详细步骤，请参阅有关使用API创建数据集的教程。
创建目标连接 target-connection
目标连接表示与所摄取数据所登陆的目标之间的连接。 要创建目标连接，您必须提供与数据湖关联的固定连接规范ID。 此连接规范ID为： c604ff05-7f1a-43c0-8e18-33bf874cb11c。
现在，您拥有目标架构、目标数据集以及到数据湖的连接规范ID的唯一标识符。 使用Flow Service API，您可以通过指定这些标识符以及将包含入站源数据的数据集来创建目标连接。
API格式
POST /targetConnections
请求
curl -X POST \
'https://platform.adobe.io/data/foundation/flowservice/targetConnections' \
-H 'Authorization: Bearer {ACCESS_TOKEN}' \
-H 'x-api-key: {API_KEY}' \
-H 'x-gw-ims-org-id: {ORG_ID}' \
-H 'x-sandbox-name: {SANDBOX_NAME}' \
-H 'Content-Type: application/json' \
-d '{
"name": "Target Connection for protocols",
"description": "Target Connection for protocols",
"data": {
"format": "parquet_xdm",
"schema": {
"id": "https://ns.adobe.com/{TENANT_ID}/schemas/e669d7aba5a02f294fafb7b269af25f7cd4a66ce59193545",
"version" "application/vnd.adobe.xed-full+json;version=1"
}
},
"params": {
"dataSetId": "5e8a55ca53662c18af37a83a"
},
"connectionSpec": {
"id": "c604ff05-7f1a-43c0-8e18-33bf874cb11c",
"version": "1.0"
}
}'
属性
描述
data.schema.id目标XDM架构的
$id。params.dataSetId上一步中生成的目标数据集的ID。 注意：创建目标连接时，必须提供有效的数据集ID。 无效的数据集ID将导致错误。
connectionSpec.id用于连接到数据湖的连接规范ID。 此ID为：
c604ff05-7f1a-43c0-8e18-33bf874cb11c。响应
成功的响应返回新目标连接的唯一标识符(id)。 在后续步骤中需要使用此值来创建数据流。
{
"id": "576d5ecf-f114-4587-ad5e-cff1144587f4",
"etag": "\"13013506-0000-0200-0000-5e8a56d80000\""
}
创建映射 mapping
要将源数据摄取到目标数据集中，必须首先将其映射到目标数据集所遵循的目标架构。
要创建映射集，请在提供目标XDM架构$id和要创建的映射集的详细信息时，向Data Prep API的mappingSets端点发出POST请求。
API格式
POST /mappingSets
请求
curl -X POST \
'https://platform.adobe.io/data/foundation/conversion/mappingSets' \
-H 'Authorization: Bearer {ACCESS_TOKEN}' \
-H 'x-api-key: {API_KEY}' \
-H 'x-gw-ims-org-id: {ORG_ID}' \
-H 'x-sandbox-name: {SANDBOX_NAME}' \
-H 'Content-Type: application/json' \
-d '{
"version": 0,
"xdmSchema": "https://ns.adobe.com/{TENANT_ID}/schemas/e669d7aba5a02f294fafb7b269af25f7cd4a66ce59193545",
"xdmVersion": "1.0",
"id": null,
"mappings": [
{
"destinationXdmPath": "_id",
"sourceAttribute": "OrderID",
"identity": false,
"identityGroup": null,
"namespaceCode": null,
"version": 0
},
{
"destinationXdmPath": "_id",
"sourceAttribute": "CustomerID",
"identity": false,
"identityGroup": null,
"namespaceCode": null,
"version": 0
},
{
"destinationXdmPath": "_id",
"sourceAttribute": "EmployeeID",
"identity": false,
"identityGroup": null,
"namespaceCode": null,
"version": 0
},
{
"destinationXdmPath": "createdByBatchID",
"sourceAttribute": "OrderDate",
"identity": false,
"identityGroup": null,
"namespaceCode": null,
"version": 0
}
]
}'
属性
描述
xdmSchema目标XDM架构的
$id。响应
成功的响应返回新创建的映射的详细信息，包括其唯一标识符(id)。 此ID是稍后步骤创建数据流所必需的。
{
"id": "37409d3017e24a3eb4a2dc21020f7a5b",
"version": 0,
"createdDate": 1586124873209,
"modifiedDate": 1586124873209,
"createdBy": "{CREATED_BY}",
"modifiedBy": "{MODIFIED_BY}"
}
查找数据流规范 specs
数据流负责从源收集数据并将这些数据导入Platform。 要创建数据流，必须首先获取负责从协议应用程序收集数据的数据流规范。
API格式
GET /flowSpecs?property=name=="CRMToAEP"
请求
curl -X GET \
'https://platform.adobe.io/data/foundation/flowservice/flowSpecs?property=name=="CRMToAEP"' \
-H 'x-api-key: {API_KEY}' \
-H 'x-gw-ims-org-id: {ORG_ID}' \
-H 'x-sandbox-name: {SANDBOX_NAME}'
响应
成功的响应将返回数据流规范的详细信息，该规范负责将数据从源引入Platform。 响应包括创建新数据流所需的唯一流规范id。
NOTE
为简单起见，以下JSON响应有效负载已隐藏。 选择“payload”（有效负载）可查看响应有效负载。
查看有效负荷
| code language-json |
|---|
|
创建数据流
收集数据的最后一步是创建数据流。 此时，您应该准备以下所需的值：
数据流负责从源中计划和收集数据。 您可以通过在有效负载中提供上述值时执行POST请求来创建数据流。
要计划摄取，您必须先将开始时间值设置为纪元时间（以秒为单位）。 然后，必须将频率值设置为五个选项之一： once、minute、hour、day或week。 间隔值用于指定两次连续摄取之间的时间段，创建一次性摄取不需要设置间隔。 对于所有其他频率，间隔值必须设置为等于或大于15。
API格式
POST /flows
请求
curl -X POST \
'https://platform.adobe.io/data/foundation/flowservice/flows' \
-H 'x-api-key: {API_KEY}' \
-H 'x-gw-ims-org-id: {ORG_ID}' \
-H 'x-sandbox-name: {SANDBOX_NAME}' \
-H 'Content-Type: application/json' \
-d '{
"name": "Creating a dataflow for a protocols source",
"description": "Creating a dataflow for a protocols source",
"flowSpec": {
"id": "14518937-270c-4525-bdec-c2ba7cce3860",
"version": "1.0"
},
"sourceConnectionIds": [
"0a768941-ddfb-499d-b689-41ddfbf99db0"
],
"targetConnectionIds": [
"576d5ecf-f114-4587-ad5e-cff1144587f4"
],
"transformations": [
{
"name": "Copy",
"params": {
"deltaColumn": {
"name": "updatedAt",
"dateFormat": "YYYY-MM-DD",
"timezone": "UTC"
}
}
},
{
"name": "Mapping",
"params": {
"mappingId": "7409d3017e24a3eb4a2dc21020f7a5b",
"mappingVersion": 0
}
}
],
"scheduleParams": {
"startTime": "1567411548",
"frequency":"minute",
"interval":"30"
}
}'
属性
描述
flowSpec.id在上一步中检索到流规范ID。
sourceConnectionIds在之前的步骤中检索到源连接ID。
targetConnectionIds在之前的步骤中检索到目标连接ID。
transformations.params.mappingId在之前的步骤中检索到映射ID。
transformations.params.deltaColum用于区分新数据和现有数据的指定列。 将根据选定列的时间戳摄取增量数据。 使用通用OData时
deltaColumn支持的格式为yyyy-MM-ddTHH:mm:ssZ。transformations.params.mappingId与数据库关联的映射ID。
scheduleParams.startTime以纪元时间表示的数据流开始时间。
scheduleParams.frequency数据流收集数据的频率。 可接受的值包括：
once、minute、hour、day或week。scheduleParams.interval间隔指定两次连续流运行之间的周期。 间隔的值应为非零整数。 每个频率的最小接受间隔值如下：
- 一次：不适用
- 分钟： 15
- 小时： 1
- 天： 1
- 周： 1
响应
成功的响应返回新创建的数据流的ID id。
{
"id": "8256cfb4-17e6-432c-a469-6aedafb16cd5",
"etag": "\"04004fe9-0000-0200-0000-5ebc4c8b0000\""
}
监测数据流
创建数据流后，您可以监视通过它摄取的数据，以查看有关流运行、完成状态和错误的信息。 有关如何监视数据流的详细信息，请参阅有关API中监视数据流的教程
后续步骤
在本教程之后，您已创建一个源连接器，以按计划从协议应用程序中收集数据。 传入数据现在可供下游平台服务（如Real-Time Customer Profile和Data Science Workspace）使用。 有关更多详细信息，请参阅以下文档：
recommendation-more-help
337b99bb-92fb-42ae-b6b7-c7042161d089