创建数据流以将数据从CRM摄取到Experience Platform
阅读本指南，了解如何使用Flow Service API创建数据流并将数据摄取到Adobe Experience Platform。
快速入门
本指南要求您对Experience Platform的以下组件有一定的了解：
使用Experience Platform API
有关如何成功调用Experience Platform API的信息，请阅读Experience Platform API快速入门指南。
创建基本连接 base
要为源创建数据流，您需要一个经过完全身份验证的源帐户及其对应的基本连接ID。 如果没有此ID，请访问源目录查找可为其创建基础连接的源列表。
创建目标XDM架构 target-schema
Experience Data Model (XDM)架构提供了一种标准化的方式，用于在Experience Platform中组织和描述客户体验数据。 要将源数据摄取到Experience Platform，您必须首先创建目标XDM架构，该架构定义要摄取的数据结构和类型。 此架构将用作您摄取的数据将驻留的Experience Platform数据集的蓝图。
通过对架构注册表API执行POST请求，可以创建目标XDM架构。 有关如何创建目标XDM架构的详细步骤，请阅读以下指南：
创建后，稍后需要目标XDM架构$id才能进行目标数据集和映射。
创建目标数据集 target-dataset
数据集是用于数据集合的存储和管理结构，其结构通常类似于具有列（架构）和行（字段）的表。 成功引入Experience Platform的数据将作为数据集存储在数据湖中。 在此步骤中，您可以创建新数据集或使用现有数据集。
您可以创建目标数据集，方法是：向目录服务API发出POST请求，同时在有效负载中提供目标架构的ID。 有关如何创建目标数据集的详细步骤，请阅读有关使用API创建数据集的指南。
TIP
如果要将数据摄取到Real-time Customer Profile，则必须确保已为配置文件启用目标XDM架构和目标数据集。
选择以查看示例
API格式
| code language-http |
|---|
|
请求
以下示例显示如何创建已启用实时客户配置文件提取的目标数据集。 在此请求中，unifiedProfile属性设置为true（在tags对象下），以告知Experience Platform在实时客户配置文件中包含数据集。
| code language-shell |
|---|
|
| table 0-row-2 1-row-2 2-row-2 3-row-2 | |
|---|---|
| 属性 | 描述 |
name |
目标数据集的描述性名称。 使用清晰且唯一的名称，以便在未来操作中更容易识别和管理您的数据集。 |
schemaRef.id |
目标XDM架构的ID。 |
tags.unifiedProfile |
一个布尔值，通知Experience Platform是否要将数据摄取到实时客户档案中。 |
响应
成功的响应将返回目标数据集的ID。 稍后创建目标连接时需要此ID。
| code language-json |
|---|
|
创建源连接 source
源连接定义如何从外部源将数据引入Experience Platform。 它指定源系统和传入数据的格式，并引用包含身份验证详细信息的基本连接。 每个源连接对于您的组织都是唯一的。
- 对于基于文件的源（如云存储），源连接可以包括列分隔符、编码类型、压缩类型、文件选择的正则表达式以及是否递归摄取文件等设置。
- 对于基于表的源（如数据库、CRM和营销自动化提供程序），源连接可以指定详细信息，如表名和列映射。
要创建源连接，请向/sourceConnections API的Flow Service端点发出POST请求，并提供您的基本连接ID、连接规范ID以及源数据文件的路径。
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": "ACME source connection",
"description": "A source connection for ACME contact data",
"baseConnectionId": "6990abad-977d-41b9-a85d-17ea8cf1c0e4",
"data": {
"format": "tabular"
},
"params": {
"tableName": "Contact",
"columns": [
{
"name": "TestID",
"type": "string",
"xdm": {
"type": "string"
}
},
{
"name": "Name",
"type": "string",
"xdm": {
"type": "string"
}
},
{
"name": "Datefield",
"type": "string",
"meta:xdmType": "date-time",
"xdm": {
"type": "string",
"format": "date-time"
}
}
]
},
"connectionSpec": {
"id": "cfc0fee1-7dc0-40ef-b73e-d8b134c436f5",
"version": "1.0"
}
}'
属性
描述
name源连接的描述性名称。 使用清晰且唯一的名称，以便将来操作中更容易识别和管理您的连接。
description可添加的可选描述，用于提供源连接的其他信息。
baseConnectionId基础连接的
id。 您可以使用Flow Service API向Experience Platform验证源以检索此ID。data.format数据的格式。 对于基于表的源（如数据库、CRM和营销自动化提供程序），将此值设置为
tabular。params.tableName您的源帐户中要摄取到Experience Platform的表的名称。
params.columns要纳入Experience Platform的特定数据表列。
connectionSpec.id正在使用的源的连接规范ID。
响应
成功的响应将返回源连接的ID。 创建数据流和摄取数据时需要此ID。
{
"id": "b7581b59-c603-4df1-a689-d23d7ac440f3",
"etag": "\"ef05d265-0000-0200-0000-6019e0080000\""
}
创建目标连接 target
目标连接表示与所摄取数据所登陆的目标之间的连接。 要创建目标连接，必须提供与该数据湖关联的固定连接规范ID。 此连接规范ID为： c604ff05-7f1a-43c0-8e18-33bf874cb11c。
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": "ACME target connection",
"description": "ACME target connection",
"data": {
"schema": {
"id": "https://ns.adobe.com/{TENANT_ID}/schemas/52b59140414aa6a370ef5e21155fd7a686744b8739ecc168",
"version": "application/vnd.adobe.xed-full+json;version=1"
}
},
"params": {
"dataSetId": "6889f4f89b982b2b90bc1207"
},
"connectionSpec": {
"id": "c604ff05-7f1a-43c0-8e18-33bf874cb11c",
"version": "1.0"
}
}'
属性
描述
name目标连接的描述性名称。 使用清晰且唯一的名称，以便将来操作中更容易识别和管理您的连接。
description可添加的可选描述，用于提供目标连接的其他信息。
data.schema.id目标XDM架构的ID。
params.dataSetId目标数据集的ID。
connectionSpec.id数据湖的连接规范ID。 此ID已修复：
c604ff05-7f1a-43c0-8e18-33bf874cb11c。映射 mapping
接下来，将源数据映射到目标数据集所遵循的目标架构。 要创建映射，请向mappingSetsAPIData Prep 的端点发出POST请求。 包含您的目标XDM架构ID以及要创建的映射集的详细信息。
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/52b59140414aa6a370ef5e21155fd7a686744b8739ecc168",
"xdmVersion": "1.0",
"id": null,
"mappings": [
{
"destinationXdmPath": "_id",
"sourceAttribute": "TestID",
"identity": false,
"identityGroup": null,
"namespaceCode": null,
"version": 0
},
{
"destinationXdmPath": "person.name.fullName",
"sourceAttribute": "Name",
"identity": false,
"identityGroup": null,
"namespaceCode": null,
"version": 0
},
{
"destinationXdmPath": "person.birthDate",
"sourceAttribute": "Datefield",
"identity": false,
"identityGroup": null,
"namespaceCode": null,
"version": 0
}
]
}'
属性
描述
xdmSchema目标XDM架构的
$id。响应
成功的响应返回新创建的映射的详细信息，包括其唯一标识符(id)。 此ID是稍后步骤创建数据流所必需的。
{
"id": "93ddfa69c4864d978832b1e5ef6ec3b9",
"version": 0,
"createdDate": 1612309018666,
"modifiedDate": 1612309018666,
"createdBy": "{CREATED_BY}",
"modifiedBy": "{MODIFIED_BY}"
}
检索数据流规范 flow-specs
在创建数据流之前，必须首先检索与源对应的数据流规范。 要检索此信息，请向/flowSpecs API的Flow Service端点发出GET请求。
API格式
GET /flowSpecs?property=name=="{NAME}"
查询参数
描述
property=name=="{NAME}"数据流规范的名称。
- 对于基于文件的源（如云存储），将此值设置为
CloudStorageToAEP。 - 对于基于表的源（如数据库、CRM和营销自动化提供程序），将此值设置为
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}'
响应
成功的响应将返回数据流规范的详细信息，该规范负责将数据从源引入Experience Platform。 响应包括创建新数据流所需的唯一流规范id。
为确保您使用正确的数据流规范，请检查响应中的items.sourceConnectionSpecIds数组。 确认源的连接规范ID已包含在此列表中。
选择以查看
| code language-json |
|---|
|
创建数据流 dataflow
数据流是已配置的管道，可在Experience Platform服务之间传输数据。 它定义如何从外部源（如数据库、云存储或API）摄取数据、如何处理数据并将其路由到目标数据集。 然后，Identity Service、实时客户档案和Destinations等服务使用这些数据集进行激活和分析。
要创建数据流，您需要为以下项提供值：
在此步骤中，您可以在scheduleParams中使用以下参数配置数据流的摄取计划：
计划参数
描述
startTime数据流应启动的时段时间（以秒为单位）。
frequency摄取频率。 配置频率以指示数据流运行的频率。 您可以将频率设置为：
once：将频率设置为once以创建一次性摄取。 间隔和回填设置不适用于一次性摄取作业。 默认情况下，调度频率设置为一次。minute：将频率设置为minute以安排数据流按分钟摄取数据。hour：将频率设置为hour以计划数据流每小时摄取数据。day：将频率设置为day以计划每天摄取数据的数据流。week：将频率设置为week可安排数据流每周摄取数据。
interval连续摄取之间的间隔（除once之外的所有频率均需要此间隔）。 配置间隔设置以建立每次引入之间的时间范围。 例如，如果将频率设置为天并且间隔为15，则数据流将每15天运行一次。 不能将间隔设置为零。 每个频率的最小接受间隔值如下：
once：不适用minute： 15hour: 1day: 1week: 1
backfill指示是否摄取
startTime之前的历史数据。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": "ACME Contact Dataflow",
"description": "A dataflow for ACME contact data",
"flowSpec": {
"id": "14518937-270c-4525-bdec-c2ba7cce3860",
"version": "1.0"
},
"sourceConnectionIds": [
"b7581b59-c603-4df1-a689-d23d7ac440f3"
],
"targetConnectionIds": [
"320f119a-5ac1-4ab1-88ea-eb19e674ea2e"
],
"transformations": [
{
"name": "Copy",
"params": {
"deltaColumn": {
"name": "Datefield",
"dateFormat": "YYYY-MM-DD",
"timezone": "UTC"
}
}
},
{
"name": "Mapping",
"params": {
"mappingId": "93ddfa69c4864d978832b1e5ef6ec3b9",
"mappingVersion": 0
}
}
],
"scheduleParams": {
"startTime": "1612310466",
"frequency":"minute",
"interval":"15",
"backfill": "true"
}
}'
属性
描述
name数据流的描述性名称。 使用清晰且唯一的名称，以便在未来操作中更容易识别和管理您的数据流。
description可添加的可选描述，用于提供数据流的附加信息。
flowSpec.id与源对应的流规范的ID。
sourceConnectionIds在之前的步骤中生成的源连接ID。
targetConnectionIds在之前的步骤中生成的目标连接ID。
transformations.params.deltaColum用于区分新数据和现有数据的指定列。 将根据选定列的时间戳摄取增量数据。
deltaColumn支持的格式为yyyy-MM-dd HH:mm:ss。 对于Microsoft Dynamics，deltaColumn支持的格式为yyyy-MM-ddTHH:mm:ssZ。transformations.params.deltaColumn.dateFormat增量列的日期格式。
transformations.params.deltaColumn.timeZone解释增量列中的值时要使用的时区。
transformations.params.mappingId在之前的步骤中生成的映射ID。
scheduleParams.startTime数据流以纪元时间表示的开始时间（自Unix纪元以来的秒数）。 确定数据流何时开始首次运行。
scheduleParams.frequency数据流运行的频率。 可接受的值包括：
once、minute、hour、day或week。scheduleParams.interval连续数据流运行之间的时间间隔，基于所选频率。 必须为非零整数。 例如，如果将频率设置为分钟，并且间隔为15，则数据流将每15分钟运行一次。
scheduleParams.backfill一个布尔值（
true或false），用于确定在首次创建数据流时是否摄取历史数据（回填）。响应
成功的响应返回新创建的数据流的ID (id)。
{
"id": "ae0a9777-b322-4ac1-b0ed-48ae9e497c7e",
"etag": "\"770029f8-0000-0200-0000-6019e7d40000\""
}
使用用户界面验证API工作流 validate-in-ui
您可以使用Experience Platform用户界面验证数据流的创建。 导航到Experience Platform UI中的 源 目录，然后从标题选项卡中选择 数据流。 接下来，使用数据流名称列并找到您使用Flow Service API创建的数据流。
您可以通过数据流活动接口进一步验证数据流。 使用右边栏查看数据流的API使用情况信息。 此部分显示的数据流ID、数据集ID和映射ID与在Flow Service中的数据流创建过程中生成的数据流ID相同。
后续步骤
本教程将指导您完成使用Flow Service API在Experience Platform中创建数据流的过程。 您已了解如何创建和配置必要的组件，包括目标XDM架构、数据集、源连接、目标连接和数据流本身。 通过执行以下步骤，您可以将数据从外部源摄取到Experience Platform中自动化，从而使下游服务（如实时客户档案和目标服务）能够将摄取的数据用于高级用例。
监测数据流
创建数据流后，可直接在Experience Platform UI中监控其性能。 这包括跟踪摄取率、成功量度和发生的任何错误。 有关如何监视数据流的详细信息，请访问有关监视帐户和数据流的教程。
更新您的数据流
要更新数据流计划、映射或常规信息的配置，请访问有关更新源数据流的教程。
删除您的数据流
您可以删除不再必需的数据流或使用 数据流 工作区中提供的 删除 功能错误地创建的数据流。 有关如何删除数据流的详细信息，请访问有关删除数据流的教程。
recommendation-more-help
337b99bb-92fb-42ae-b6b7-c7042161d089