Show Menu
主题×

通过源连接器和API从第三方数据库收集数据

Flow Service用于在Adobe Experience Platform内收集和集中来自不同来源的客户数据。 该服务提供用户界面和RESTful API,所有支持的源都可从中连接。
本教程介绍从第三方数据库检索数据并通过源连接器和API将其引入平台的步骤。

入门指南

本教程要求您具有与第三方数据库的有效连接,以及要引入平台的文件(包括文件的路径和结构)的相关信息。 如果您没有此信息,请在尝试本教程 之前参阅教程,了解如何使用流服务 API浏览数据库。
本教程还要求您对Adobe Experience Platform的以下组件有充分的了解:
  • 体验数据模型(XDM)系统 : Experience Platform组织客户体验数据的标准化框架。
    • 模式合成基础 : 了解XDM模式的基本构件,包括模式构成的主要原则和最佳做法。
    • 模式注册开发人员指南 : 包括成功执行对模式注册表API的调用时需要了解的重要信息。 这包括您 {TENANT_ID} 的、“容器”的概念以及发出请求所需的标题(特别要注意“接受”标题及其可能的值)。
  • 目录服务 : Catalog是Experience Platform中数据位置和世系的记录系统。
  • 批量摄取 : Batch Ingestion API允许您将数据作为批处理文件导入到Experience Platform中。
  • 沙箱 : Experience Platform提供虚拟沙箱,将单个Platform实例分为单独的虚拟环境,以帮助开发和改进数字体验应用程序。
以下各节提供了使用流服务API成功连接到数据库或NoSQL系统所需了解的其他信息。

读取示例API调用

本教程提供示例API调用,以演示如何设置请求的格式。 这包括路径、必需的标头和格式正确的请求负载。 还提供API响应中返回的示例JSON。 有关示例API调用文档中使用的惯例的信息,请参阅Experience Platform疑 难解答指南中有关如何阅读示例API调 用的部分。

收集所需标题的值

要调用平台API,您必须先完成身份验证 教程 。 完成身份验证教程后,将提供所有Experience Platform API调用中每个所需标头的值,如下所示:
  • 授权: 承载者 {ACCESS_TOKEN}
  • x-api-key: {API_KEY}
  • x-gw-ims-org-id: {IMS_ORG}
Experience Platform中的所有资源(包括属于流服务的资源)都与特定虚拟沙箱隔离。 对平台API的所有请求都需要一个标头,它指定操作将在以下位置进行的沙箱的名称:
  • x-sandbox-name: {SANDBOX_NAME}
所有包含有效负荷(POST、PUT、PATCH)的请求都需要额外的媒体类型标头:
  • 内容类型: application/json

创建点对点XDM类和模式

要通过源连接器将外部数据引入平台,必须为原始源数据创建一个临时XDM类和模式。
要创建点对点类和模式,请按照点对点模式教 程中概述的步骤操作 。 创建点对点类时,必须在请求主体中描述源数据中找到的所有字段。
继续按照开发人员指南中概述的步骤操作,直到您创建临时模式。 获取并存储点对点模式 $id 的唯一标识符(),然后继续执行本教程的下一步。

创建源连接

创建点对点XDM模式后,现在可以使用对流服务API的POST请求创建源连接。 源连接由基本连接、源数据文件和对描述源模式的引用组成。
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: {IMS_ORG}' \
    -H 'x-sandbox-name: {SANDBOX_NAME}' \
    -H 'Content-Type: application/json' \
    -d '{
        "name": "Source Connection for database or NoSQL",
        "baseConnectionId": "54c22133-3a01-4d3b-8221-333a01bd3b03",
        "description": "Source Connection for database or NoSQL to ingest test1.Mytable",
        "data": {
            "format": "parquet_xdm",
            "schema": {
                "id": "https://ns.adobe.com/{TENANT_ID}/schemas/c8c2b32d6f6a53d5ffc7212c37b3a9369282404a7bd551e8",
                "version": "application/vnd.adobe.xed-full-notext+json; version=1"
            }
        },
        "params": {
            "path": "test1.Mytable"
        },
        "connectionSpec": {
            "id": "3c9b37f8-13a6-43d8-bad3-b863b941fedd",
            "version": "1.0"
        }
}'

属性
描述
baseConnectionId
数据库连接的ID。
data.schema.id
$id 时XDM模式。
params.path
源文件的路径。
connectionSpec.id
数据库或NoSQL系统的连接规范ID。
响应
成功的响应会返回新创建的源 id 连接的唯一标识符()。 此ID是后续步骤中创建目标连接时必需的。
{
    "id": "beefc650-f2dc-45e2-afc6-50f2dcc5e2b8",
    "etag": "\"1600f153-0000-0200-0000-5e4710880000\""
}

创建目标XDM模式

在前面的步骤中,创建了一个专门的XDM模式来构造源数据。 要在平台中使用源模式,还必须创建一个目标,以根据您的需求构建源数据。 然后,目标模式用于创建包含源数据的平台数据集。 此目标XDM模式还扩展了XDM单个用户档案类。
通过对目标注册表API执行POST请求,可以创 建模式XDM模式 。 如果您希望使用Experience Platform中的用户界面,模式编 辑器教程提供了在模式编辑器中 ,执行类似操作的分步说明。
API格式
POST /tenant/schemas

请求
以下示例请求创建一个XDM模式,它扩展了XDM单个用户档案类。
curl -X POST \
    'https://platform.adobe.io/data/foundation/schemaregistry/tenant/schemas' \
    -H 'Authorization: Bearer {ACCESS_TOKEN}' \
    -H 'x-api-key: {API_KEY}' \
    -H 'x-gw-ims-org-id: {IMS_ORG}' \
    -H 'x-sandbox-name: {SANDBOX_NAME}' \
    -H 'Content-Type: application/json' \
    -d '{
        "type": "object",
        "title": "Target schema for database or NoSQL",
        "description": "Target schema for database or NoSQL",
        "allOf": [
            {
                "$ref": "https://ns.adobe.com/xdm/context/profile"
            },
            {
                "$ref": "https://ns.adobe.com/xdm/context/profile-person-details"
            },
            {
                "$ref": "https://ns.adobe.com/xdm/context/profile-personal-details"
            }
        ],
    }'

响应
成功的响应会返回新创建模式的详细信息,包括其唯一标识符( $id )。 后续步骤中需要此ID才能创建目标数据集、映射和数据流。
{
    "$id": "https://ns.adobe.com/{TENANT}/schemas/a9d63c5e3fab2687e064577959d0b91e274823f91f2f578e",
    "meta:altId": "_{TENANT}.schemas.a9d63c5e3fab2687e064577959d0b91e274823f91f2f578e",
    "meta:resourceType": "schemas",
    "version": "1.0",
    "title": "Target schema for database or NoSQL",
    "type": "object",
    "description": "Target schema for database or NoSQL",
    "allOf": [
        {
            "$ref": "https://ns.adobe.com/xdm/context/profile",
            "type": "object",
            "meta:xdmType": "object"
        },
        {
            "$ref": "https://ns.adobe.com/xdm/context/profile-person-details",
            "type": "object",
            "meta:xdmType": "object"
        },
        {
            "$ref": "https://ns.adobe.com/xdm/context/profile-personal-details",
            "type": "object",
            "meta:xdmType": "object"
        }
    ],
    "refs": [
        "https://ns.adobe.com/xdm/context/profile-person-details",
        "https://ns.adobe.com/xdm/context/profile-personal-details",
        "https://ns.adobe.com/xdm/context/profile"
    ],
    "imsOrg": "{IMS_ORG}",
    "meta:extensible": false,
    "meta:abstract": false,
    "meta:extends": [
        "https://ns.adobe.com/xdm/context/profile-person-details",
        "https://ns.adobe.com/xdm/context/profile-personal-details",
        "https://ns.adobe.com/xdm/common/auditable",
        "https://ns.adobe.com/xdm/data/record",
        "https://ns.adobe.com/xdm/context/profile",
        "https://ns.adobe.com/xdm/common/extensible"
    ],
    "meta:xdmType": "object",
    "meta:registryMetadata": {
        "repo:createdDate": 1581716110281,
        "repo:lastModifiedDate": 1581716110281,
        "xdm:createdClientId": "{CREATED_CLIENT_ID}",
        "xdm:lastModifiedClientId": "{LAST_MODIFIED_CLIENT_ID}",
        "xdm:createdUserId": "{CREATED_USER_ID}",
        "xdm:lastModifiedUserId": "{LAST_MODIFIED_USER_ID}",
        "eTag": "6360f2175b40462ff25ec8735dc93a7e3af6a8faadd80a1cc500a59721e1a424"
    },
    "meta:class": "https://ns.adobe.com/xdm/context/profile",
    "meta:containerId": "tenant",
    "meta:tenantNamespace": "{TENANT_ID}"
}

创建目标数据集

通过对Catalog Service API执行POST请求,提供有效负 荷内目标模式的ID ,可以创建目标数据集。
API格式
POST /dataSets

请求
curl -X POST \
    'https://platform.adobe.io/data/foundation/catalog/dataSets?requestDataSource=true' \
    -H 'Authorization: Bearer {ACCESS_TOKEN}' \
    -H 'x-api-key: {API_KEY}' \
    -H 'x-gw-ims-org-id: {IMS_ORG}' \
    -H 'x-sandbox-name: {SANDBOX_NAME}' \
    -H 'Content-Type: application/json' \
    -d '{
        "name": "Target Dataset for database or NoSQL",
        "schemaRef": {
            "id": "https://ns.adobe.com/{TENANT_ID}/schemas/a9d63c5e3fab2687e064577959d0b91e274823f91f2f578e",
            "contentType": "application/vnd.adobe.xed-full-notext+json; version=1"
        }
    }'

属性
描述
schemaRef.id
目标XDM模式的ID。
响应
成功的响应会返回一个数组,其中包含格式为新创建数据集的ID "@/datasets/{DATASET_ID}" 。 数据集ID是由系统生成的只读字符串,用于在API调用中引用数据集。 按照后续步骤中的要求存储目标数据集ID,以创建目标连接和数据流。
[
    "@/dataSets/5e47161fa49bb818ad7f47bd"
]

创建数据集基础连接

为了将外部数据引入Platform,必须先获取Experience Platform数据集基础连接。
要创建数据集基础连接,请按照数据集基础连接教 程中概述的步骤操作
继续按照开发人员指南中所述的步骤操作,直到您创建了数据集基础连接。 获取并存储唯一标识符( $id ),然后在下一步中继续将它用作基本连接ID以创建目标连接。

创建目标连接

您现在具有数据集基础连接、目标模式和目标数据集的唯一标识符。 使用这些标识符,您可以使用流服务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: {IMS_ORG}' \
    -H 'x-sandbox-name: {SANDBOX_NAME}' \
    -H 'Content-Type: application/json' \
    -d '{
        "baseConnectionId": "d6c3988d-14ef-4000-8398-8d14ef000021",
        "name": "Target Connection for database or NoSQL",
        "description": "Target Connection for database or NoSQL",
        "data": {
            "format": "parquet_xdm",
            "schema": {
                "id": "https://ns.adobe.com/{TENANT_ID}/schemas/a9d63c5e3fab2687e064577959d0b91e274823f91f2f578e",
                "version": "application/vnd.adobe.xed-full+json;version=1.0"
            }
        },
        "params": {
            "dataSetId": "5e47161fa49bb818ad7f47bd"
        },
        "connectionSpec": {
            "id": "3c9b37f8-13a6-43d8-bad3-b863b941fedd",
            "version": "1.0"
        }
    }'

属性
描述
baseConnectionId
数据集基础连接的ID。
data.schema.id
目标 $id XDM模式。
params.dataSetId
目标数据集的ID。
connectionSpec.id
第三方数据库的连接规范ID。
创建目标连接时,请确保将数据集基础连接值用于基础连接,而 id 非第三方源连接器的基础连接。
响应
成功的响应会返回新目标连接的唯一标识符( id )。 此值是后续步骤中创建数据流的必需值。
{
    "id": "4f3845b6-87d9-4702-b845-b687d9270297",
    "etag": "\"2a007aa8-0000-0200-0000-5e597aaf0000\""
}

创建映射

为了将源数据引入目标数据集,必须首先将其映射到目标数据集所附加的目标模式。 这是通过对转换服务API执行POST请求而实现的,该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: {IMS_ORG}' \
    -H 'x-sandbox-name: {SANDBOX_NAME}' \
    -H 'Content-Type: application/json' \
    -d '{
        "version": 0,
        "xdmSchema": "https://ns.adobe.com/{TENANT_ID}/schemas/a9d63c5e3fab2687e064577959d0b91e274823f91f2f578e",
        "xdmVersion": "1.0",
        "id": null,
        "mappings": [
            {
                "destinationXdmPath": "person.name",
                "sourceAttribute": "Name",
                "identity": false,
                "identityGroup": null,
                "namespaceCode": null,
                "version": 0
            },
            {
                "destinationXdmPath": "mobilePhone.number",
                "sourceAttribute": "Phone",
                "identity": false,
                "identityGroup": null,
                "namespaceCode": null,
                "version": 0
            },
            {
                "destinationXdmPath": "personalEmail.address",
                "sourceAttribute": "email",
                "identity": false,
                "identityGroup": null,
                "namespaceCode": null,
                "version": 0
            }
        ]
    }'

属性
描述
xdmSchema
目标 $id XDM模式。
响应
成功的响应会返回新创建的映射的详细信息,包括其唯一标识符( id )。 此ID是后续步骤中创建数据流的必需ID。
{
    "id": "ab91c736-1f3d-4b09-8424-311d3d3e3cea",
    "version": 1,
    "createdDate": 1568047685000,
    "modifiedDate": 1568047703000,
    "inputSchemaRef": {
        "id": null,
        "contentType": null
    },
    "outputSchemaRef": {
        "id": "https://ns.adobe.com/{TENANT_ID}/schemas/efea012ad5deefcdf51afd23ceb3583f",
        "contentType": "1.0"
    },
    "mappings": [
        {
            "id": "7bbea5c0f0ef498aa20aa2e2e5c22290",
            "version": 0,
            "createdDate": 1568047685000,
            "modifiedDate": 1568047685000,
            "sourceType": "text/x.schema-path",
            "source": "Id",
            "destination": "_id",
            "identity": false,
            "primaryIdentity": false,
            "matchScore": 0.0,
            "sourceAttribute": "Id",
            "destinationXdmPath": "_id"
        },
        {
            "id": "def7fd7db2244f618d072e8315f59c05",
            "version": 0,
            "createdDate": 1568047685000,
            "modifiedDate": 1568047685000,
            "sourceType": "text/x.schema-path",
            "source": "FirstName",
            "destination": "person.name.firstName",
            "identity": false,
            "primaryIdentity": false,
            "matchScore": 0.0,
            "sourceAttribute": "FirstName",
            "destinationXdmPath": "person.name.firstName"
        },
        {
            "id": "e974986b28c74ed8837570f421d0b2f4",
            "version": 0,
            "createdDate": 1568047685000,
            "modifiedDate": 1568047685000,
            "sourceType": "text/x.schema-path",
            "source": "LastName",
            "destination": "person.name.lastName",
            "identity": false,
            "primaryIdentity": false,
            "matchScore": 0.0,
            "sourceAttribute": "LastName",
            "destinationXdmPath": "person.name.lastName"
        }
    ],
    "status": "PUBLISHED",
    "xdmVersion": "1.0",
    "schemaRef": {
        "id": "https://ns.adobe.com/{TENANT_ID}/schemas/2574494fdb01fa14c25b52d717ccb828",
        "contentType": "1.0"
    },
    "xdmSchema": "https://ns.adobe.com/{TENANT_ID}/schemas/2574494fdb01fa14c25b52d717ccb828"
}

检索数据流规范

数据流负责从源收集数据并将其引入平台。 要创建数据流,必须首先通过对Flow Service API执行GET请求来获取数据流规范。 数据流规范负责从外部数据库或NoSQL系统收集数据。
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: {IMS_ORG}' \
    -H 'x-sandbox-name: {SANDBOX_NAME}'

响应
成功的响应会返回数据流规范的详细信息,该规范负责将数据从数据库或NoSQL系统引入平台。 下一步中需要此ID才能创建新数据流。
{
    "items": [
        {
            "id": "14518937-270c-4525-bdec-c2ba7cce3860",
            "name": "CRMToAEP",
            "providerId": "0ed90a81-07f4-4586-8190-b40eccef1c5a",
            "version": "1.0",
            "transformationSpecs": [
                {
                    "name": "Copy",
                    "spec": {
                        "$schema": "http://json-schema.org/draft-07/schema#",
                        "type": "object",
                        "properties": {
                            "deltaColumn": {
                                "type": "object",
                                "properties": {
                                    "name": {
                                        "type": "string"
                                    },
                                    "dateFormat": {
                                        "type": "string"
                                    },
                                    "timezone": {
                                        "type": "string"
                                    }
                                },
                                "required": [
                                    "name"
                                ]
                            }
                        },
                        "required": [
                            "deltaColumn"
                        ]
                    }
                },
                {
                    "name": "Mapping",
                    "spec": {
                        "$schema": "http://json-schema.org/draft-07/schema#",
                        "type": "object",
                        "description": "defines various params required for different mapping from source to target",
                        "properties": {
                            "mappingId": {
                                "type": "string"
                            },
                            "mappingVersion": {
                                "type": "string"
                            }
                        }
                    }
                }
            ],
            "scheduleSpec": {
                "name": "PeriodicSchedule",
                "type": "Periodic",
                "spec": {
                    "$schema": "http://json-schema.org/draft-07/schema#",
                    "type": "object",
                    "properties": {
                        "startTime": {
                            "description": "epoch time",
                            "type": "integer"
                        },
                        "endTime": {
                            "description": "epoch time",
                            "type": "integer"
                        },
                        "interval": {
                            "type": "integer"
                        },
                        "frequency": {
                            "type": "string",
                            "enum": [
                                "minute",
                                "hour",
                                "day",
                                "week"
                            ]
                        },
                        "backfill": {
                            "type": "boolean",
                            "default": true
                        }
                    },
                    "required": [
                        "startTime",
                        "frequency",
                        "interval"
                    ],
                    "if": {
                        "properties": {
                            "frequency": {
                                "const": "minute"
                            }
                        }
                    },
                    "then": {
                        "properties": {
                            "interval": {
                                "minimum": 15
                            }
                        }
                    },
                    "else": {
                        "properties": {
                            "interval": {
                                "minimum": 1
                            }
                        }
                    }
                }
            }
        }
    ]
}

创建数据流

收集数据的最后一步是创建数据流。 此时,您应准备以下必需值:
数据流负责从源调度和收集数据。 您可以通过执行POST请求来创建数据流,同时在有效负荷中提供以前提到的值。
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: {IMS_ORG}' \
    -H 'x-sandbox-name: {SANDBOX_NAME}' \
    -H 'Content-Type: application/json' \
    -d '{
        "name": "Dataflow between database or NoSQL Platform",
        "description": "Inbound data to Platform",
        "flowSpec": {
            "id": "14518937-270c-4525-bdec-c2ba7cce3860",
            "version": "1.0"
        },
        "sourceConnectionIds": [
            "beefc650-f2dc-45e2-afc6-50f2dcc5e2b8"
        ],
        "targetConnectionIds": [
            "4f3845b6-87d9-4702-b845-b687d9270297"
        ],
        "transformations": [
            {
                "name": "Mapping",
                "params": {
                    "mappingId": "ab91c736-1f3d-4b09-8424-311d3d3e3cea",
                    "mappingVersion": "0"
                }
            }
        ],
        "scheduleParams": {
            "startTime": "1567411548",
            "frequency":"minute",
            "interval":"30"
        }
    }'

属性
描述
flowSpec.id
与数据库关联的数据流规范ID。
sourceConnectionIds
与数据库关联的源连接ID。
targetConnectionIds
与目标库关联的数据连接ID。
transformations.params.mappingId
与数据库关联的映射ID。
响应
成功的响应会返回新创 id 建的数据流的ID()。
{
    "id": "8256cfb4-17e6-432c-a469-6aedafb16cd5"
}

后续步骤

按照本教程,您已创建了一个源连接器,以按计划从第三方数据库收集数据。 现在,下游平台服务(如实时客户用户档案和数据科学工作区)可以使用传入数据。 有关更多详细信息,请参阅以下文档:

附录

以下部分列表不同的云存储源连接器及其连接规范。

连接规范

连接器名称
连接规范ID
Amazon Redshift
3416976c-a9ca-4bba-901a-1f08f66978ff
Azure HDInsights上的Apache Hive
aac9bbd4-6c01-46ce-b47e-51c6f0f6db3f
Azure HDInsights上的Apache Spark
6a8d82bc-1caf-45d1-908d-cadabc9d63a6
Azure Data Explorer
0479cc14-7651-4354-b233-7480606c2ac3
Azure突触分析
a49bcc7d-8038-43af-b1e4-5a7a089a7d79
Azure表存储
ecde33f2-c56f-46cc-bdea-ad151c16cd69
CouchBase
1fe283f6-9bec-11ea-bb37-0242ac130002
Google BigQuery
3c9b37f8-13a6-43d8-bad3-b863b941fedd
IBM DB2
09182899-b429-40c9-a15a-bf3ddbc8ced7
MariaDB
000eb99-cd47-43f3-827c-43caf170f015
Microsoft SQL Server
1f372ff9-38a4-4492-96f5-b9a4e4bd00ec
MySQL
26d738e0-8963-47ea-aadf-c60de735468a
Oracle
d6b52d86-f0f8-475f-89d4-ce54c8527328
凤凰城
102706fb-a5cd-42ee-afe0-bc42f017ff43
PostgreSQL
74a1c565-4e59-48d7-9d67-7c03b8a13137