批量摄取开发人员指南
本文档提供了全面的使用指南 批量摄取API端点 在Adobe Experience Platform中。 有关批量摄取API的概述,包括先决条件和最佳实践,请从阅读 批量摄取API概述.
本文档的附录提供了 设置用于摄取的数据格式,包括示例CSV和JSON数据文件。
快速入门
本指南中使用的API端点是 批量摄取API. 通过RESTful API提供批量摄取,您可以在其中对支持的对象类型执行基本CRUD操作。
摄取JSON文件
创建批次
首先,您需要创建一个批次,以JSON作为输入格式。 创建批次时,您需要提供数据集ID。 您还需要确保作为批次的一部分上传的所有文件符合链接到所提供数据集的XDM架构。
isMultiLineJson
需要设置标志。 欲知更多信息,请阅读 批量摄取疑难解答指南.API格式
POST /batches
请求
curl -X POST https://platform.adobe.io/data/foundation/import/batches \
-H 'Authorization: Bearer {ACCESS_TOKEN}' \
-H 'Content-Type: application/json' \
-H 'x-gw-ims-org-id: {ORG_ID}' \
-H 'x-api-key: {API_KEY}' \
-H 'x-sandbox-name: {SANDBOX_NAME}' \
-d '{
"datasetId": "{DATASET_ID}",
"inputFormat": {
"format": "json"
}
}'
{DATASET_ID}
响应
{
"id": "{BATCH_ID}",
"imsOrg": "{ORG_ID}",
"updated": 0,
"status": "loading",
"created": 0,
"relatedObjects": [
{
"type": "dataSet",
"id": "{DATASET_ID}"
}
],
"version": "1.0.0",
"tags": {},
"createdUser": "{USER_ID}",
"updatedUser": "{USER_ID}"
}
{BATCH_ID}
{DATASET_ID}
上传文件
现在您已经创建了批次,可以使用批次创建响应中的批次ID将文件上传到批次。 您可以将多个文件上载到批处理。
API格式
PUT /batches/{BATCH_ID}/datasets/{DATASET_ID}/files/{FILE_NAME}
{BATCH_ID}
{DATASET_ID}
{FILE_NAME}
请求
curl -X PUT https://platform.adobe.io/data/foundation/import/batches/{BATCH_ID}/datasets/{DATASET_ID}/files/{FILE_NAME}.json \
-H 'Authorization: Bearer {ACCESS_TOKEN}' \
-H 'content-type: application/octet-stream' \
-H 'x-gw-ims-org-id: {ORG_ID}' \
-H 'x-api-key: {API_KEY}' \
-H 'x-sandbox-name: {SANDBOX_NAME}' \
--data-binary "@{FILE_PATH_AND_NAME}.json"
{FILE_PATH_AND_NAME}
acme/customers/campaigns/summer.json
.响应
200 OK
完成批处理
上传完文件的所有不同部分后,您需要表明数据已完全上传,并且批次已准备好进行升级。
API格式
POST /batches/{BATCH_ID}?action=COMPLETE
{BATCH_ID}
请求
curl -X POST "https://platform.adobe.io/data/foundation/import/batches/{BATCH_ID}?action=COMPLETE" \
-H 'Authorization: Bearer {ACCESS_TOKEN}' \
-H 'x-gw-ims-org-id: {ORG_ID}' \
-H 'x-api-key: {API_KEY}' \
-H 'x-sandbox-name: {SANDBOX_NAME}'
响应
200 OK
摄取Parquet文件 ingest-parquet-files
创建批次
首先,需要创建一个批次,以Parquet作为输入格式。 创建批次时,您需要提供数据集ID。 您还需要确保作为批次的一部分上传的所有文件符合链接到所提供数据集的XDM架构。
请求
curl -X POST "https://platform.adobe.io/data/foundation/import/batches" \
-H "Authorization: Bearer {ACCESS_TOKEN}" \
-H "Content-Type: application/json" \
-H "x-gw-ims-org-id: {ORG_ID}" \
-H "x-api-key: {API_KEY}" \
-H "x-sandbox-name: {SANDBOX_NAME}"
-d '{
"datasetId": "{DATASET_ID}",
"inputFormat": {
"format": "parquet"
}
}'
{DATASET_ID}
响应
201 Created
{
"id": "{BATCH_ID}",
"imsOrg": "{ORG_ID}",
"updated": 0,
"status": "loading",
"created": 0,
"relatedObjects": [
{
"type": "dataSet",
"id": "{DATASET_ID}"
}
],
"version": "1.0.0",
"tags": {},
"createdUser": "{USER_ID}",
"updatedUser": "{USER_ID}"
}
{BATCH_ID}
{DATASET_ID}
{USER_ID}
上传文件
现在您已经创建了批次,接下来可以使用 batchId
从之前将文件上传到批处理。 您可以将多个文件上载到批处理。
API格式
PUT /batches/{BATCH_ID}/datasets/{DATASET_ID}/files/{FILE_NAME}
{BATCH_ID}
{DATASET_ID}
{FILE_NAME}
请求
curl -X PUT https://platform.adobe.io/data/foundation/import/batches/{BATCH_ID}/datasets/{DATASET_ID}/files/{FILE_NAME}.parquet \
-H 'Authorization: Bearer {ACCESS_TOKEN}' \
-H 'Content-Type: application/octet-stream' \
-H 'x-gw-ims-org-id: {ORG_ID}' \
-H 'x-api-key: {API_KEY}' \
-H 'x-sandbox-name: {SANDBOX_NAME}' \
--data-binary "@{FILE_PATH_AND_NAME}.parquet"
{FILE_PATH_AND_NAME}
acme/customers/campaigns/summer.parquet
.响应
200 OK
完成批处理
上传完文件的所有不同部分后,您需要表明数据已完全上传,并且批次已准备好进行升级。
API格式
POST /batches/{BATCH_ID}?action=complete
{BATCH_ID}
请求
curl -X POST https://platform.adobe.io/data/foundation/import/batches/{BATCH_ID}?action=COMPLETE \
-H 'Authorization: Bearer {ACCESS_TOKEN}' \
-H 'x-gw-ims-org-id: {ORG_ID}' \
-H 'x-api-key: {API_KEY}' \
-H 'x-sandbox-name: {SANDBOX_NAME}'
响应
200 OK
摄取大型Parquet文件
创建批次
首先,需要创建一个批次,以Parquet作为输入格式。 创建批次时,您需要提供数据集ID。 您还需要确保作为批次的一部分上传的所有文件符合链接到所提供数据集的XDM架构。
API格式
POST /batches
请求
curl -X POST https://platform.adobe.io/data/foundation/import/batches \
-H 'Authorization: Bearer {ACCESS_TOKEN}' \
-H 'Content-Type: application/json' \
-H 'x-gw-ims-org-id: {ORG_ID}' \
-H 'x-api-key: {API_KEY}' \
-H 'x-sandbox-name: {SANDBOX_NAME}'
-d '{
"datasetId": "{DATASET_ID}",
"inputFormat": {
"format": "parquet"
}
}'
{DATASET_ID}
响应
201 Created
{
"id": "{BATCH_ID}",
"imsOrg": "{ORG_ID}",
"updated": 0,
"status": "loading",
"created": 0,
"relatedObjects": [
{
"type": "dataSet",
"id": "{DATASET_ID}"
}
],
"version": "1.0.0",
"tags": {},
"createdUser": "{USER_ID}",
"updatedUser": "{USER_ID}"
}
{BATCH_ID}
{DATASET_ID}
{USER_ID}
初始化大文件
创建批次后,您需要先初始化大文件,然后再将块上传到批次。
API格式
POST /batches/{BATCH_ID}/datasets/{DATASET_ID}/files/{FILE_NAME}
{BATCH_ID}
{DATASET_ID}
{FILE_NAME}
请求
curl -X POST https://platform.adobe.io/data/foundation/import/batches/{BATCH_ID}/datasets/{DATASET_ID}/files/{FILE_NAME}.parquet?action=INITIALIZE \
-H 'Authorization: Bearer {ACCESS_TOKEN}' \
-H 'x-gw-ims-org-id: {ORG_ID}' \
-H 'x-api-key: {API_KEY}' \
-H 'x-sandbox-name: {SANDBOX_NAME}'
响应
201 Created
上载大文件块
现在文件已创建,可以通过重复的PATCH请求(文件的每个部分各一个)上传所有后续块。
API格式
PATCH /batches/{BATCH_ID}/datasets/{DATASET_ID}/files/{FILE_NAME}
{BATCH_ID}
{DATASET_ID}
{FILE_NAME}
请求
curl -X PATCH https://platform.adobe.io/data/foundation/import/batches/{BATCH_ID}/datasets/{DATASET_ID}/files/{FILE_NAME}.parquet \
-H 'Authorization: Bearer {ACCESS_TOKEN}' \
-H 'Content-Type: application/octet-stream' \
-H 'Content-Range: bytes {CONTENT_RANGE}' \
-H 'x-gw-ims-org-id: {ORG_ID}' \
-H 'x-api-key: {API_KEY}' \
-H 'x-sandbox-name: {SANDBOX_NAME}' \
--data-binary "@{FILE_PATH_AND_NAME}.parquet"
{CONTENT_RANGE}
{FILE_PATH_AND_NAME}
acme/customers/campaigns/summer.json
.响应
200 OK
完整的大文件
现在您已经创建了批次,接下来可以使用 batchId
从之前将文件上传到批处理。 您可以将多个文件上载到批处理。
API格式
POST /batches/{BATCH_ID}/datasets/{DATASET_ID}/files/{FILE_NAME}
{BATCH_ID}
{DATASET_ID}
{FILE_NAME}
请求
curl -X POST https://platform.adobe.io/data/foundation/import/batches/{BATCH_ID}/datasets/{DATASET_ID}/files/{FILE_NAME}.parquet?action=COMPLETE \
-H 'Authorization: Bearer {ACCESS_TOKEN}' \
-H 'x-gw-ims-org-id: {ORG_ID}' \
-H 'x-api-key: {API_KEY}' \
-H 'x-sandbox-name: {SANDBOX_NAME}'
响应
201 Created
完成批处理
上传完文件的所有不同部分后,您需要表明数据已完全上传,并且批次已准备好进行升级。
API格式
POST /batches/{BATCH_ID}?action=COMPLETE
{BATCH_ID}
请求
curl -X POST https://platform.adobe.io/data/foundation/import/batches/{BATCH_ID}?action=COMPLETE \
-H 'Authorization: Bearer {ACCESS_TOKEN}' \
-H 'x-gw-ims-org-id: {ORG_ID}' \
-H 'x-api-key: {API_KEY}' \
-H 'x-sandbox-name: {SANDBOX_NAME}'
响应
200 OK
摄取CSV文件
为了摄取CSV文件,您需要创建一个支持CSV的类、架构和数据集。 有关如何创建必要类和架构的详细信息,请按照 临时架构创建教程.
创建数据集
按照上面的说明创建必要的类和架构后,您将需要创建可支持CSV的数据集。
API格式
POST /catalog/dataSets
请求
curl -X POST https://platform.adobe.io/data/foundation/catalog/dataSets \
-H 'Authorization: Bearer {ACCESS_TOKEN}' \
-H 'Content-Type: application/json' \
-H 'x-api-key: {API_KEY}' \
-H 'x-gw-ims-org-id: {ORG_ID}' \
-H 'x-sandbox-name: {SANDBOX_NAME}'
-d '{
"name": "{DATASET_NAME}",
"schemaRef": {
"id": "https://ns.adobe.com/{TENANT_ID}/schemas/{SCHEMA_ID}",
"contentType": "application/vnd.adobe.xed+json;version=1"
}
}'
{TENANT_ID}
{SCHEMA_ID}
创建批次
接下来,您需要创建一个将CSV作为输入格式的批次。 创建批次时,您需要提供数据集ID。 您还需要确保作为批次的一部分上传的所有文件都符合链接到所提供数据集的架构。
API格式
POST /batches
请求
curl -X POST https://platform.adobe.io/data/foundation/import/batches \
-H 'Authorization: Bearer {ACCESS_TOKEN}' \
-H 'Content-Type: application/json' \
-H 'x-gw-ims-org-id: {ORG_ID}' \
-H 'x-api-key: {API_KEY}' \
-H 'x-sandbox-name: {SANDBOX_NAME}'
-d '{
"datasetId": "{DATASET_ID}",
"inputFormat": {
"format": "csv"
}
}'
{DATASET_ID}
响应
201 Created
{
"id": "{BATCH_ID}",
"imsOrg": "{ORG_ID}",
"updated": 0,
"status": "loading",
"created": 0,
"relatedObjects": [
{
"type": "dataSet",
"id": "{DATASET_ID}"
}
],
"version": "1.0.0",
"tags": {},
"createdUser": "{USER_ID}",
"updatedUser": "{USER_ID}"
}
{BATCH_ID}
{DATASET_ID}
{USER_ID}
上传文件
现在您已经创建了批次,接下来可以使用 batchId
从之前将文件上传到批处理。 您可以将多个文件上载到批处理。
API格式
PUT /batches/{BATCH_ID}/datasets/{DATASET_ID}/files/{FILE_NAME}
{BATCH_ID}
{DATASET_ID}
{FILE_NAME}
请求
curl -X PUT https://platform.adobe.io/data/foundation/import/batches/{BATCH_ID}/datasets/{DATASET_ID}/files/{FILE_NAME}.csv \
-H 'Authorization: Bearer {ACCESS_TOKEN}' \
-H 'Content-Type: application/octet-stream' \
-H 'x-gw-ims-org-id: {ORG_ID}' \
-H 'x-api-key: {API_KEY}' \
-H 'x-sandbox-name: {SANDBOX_NAME}' \
--data-binary "@{FILE_PATH_AND_NAME}.csv"
{FILE_PATH_AND_NAME}
acme/customers/campaigns/summer.csv
.响应
200 OK
完成批处理
完成上载文件的所有不同部分后,您需要发出信号表明数据已完全上载,并且批次已准备好进行升级。
API格式
POST /batches/{BATCH_ID}?action=COMPLETE
请求
curl -X POST https://platform.adobe.io/data/foundation/import/batches/{BATCH_ID}?action=COMPLETE \
-H 'Authorization: Bearer {ACCESS_TOKEN}' \
-H 'x-gw-ims-org-id: {ORG_ID}' \
-H 'x-api-key: {API_KEY}' \
-H 'x-sandbox-name: {SANDBOX_NAME}'
响应
200 OK
取消批次
在处理批次时,仍可取消它。 但是,一旦完成批处理(例如成功或失败状态),就无法取消批处理。
API格式
POST /batches/{BATCH_ID}?action=ABORT
{BATCH_ID}
请求
curl -X POST https://platform.adobe.io/data/foundation/import/batches/{BATCH_ID}?action=ABORT \
-H 'Authorization: Bearer {ACCESS_TOKEN}' \
-H 'x-gw-ims-org-id: {ORG_ID}' \
-H 'x-api-key: {API_KEY}' \
-H 'x-sandbox-name: {SANDBOX_NAME}'
响应
200 OK
删除批次 delete-a-batch
通过使用执行以下POST请求,可以删除批次 action=REVERT
查询要删除的批次ID的参数。 该批次被标记为“不活动”,因此可用于垃圾收集。 将异步收集批次,然后将其标记为“已删除”。
API格式
POST /batches/{BATCH_ID}?action=REVERT
{BATCH_ID}
请求
curl -X POST https://platform.adobe.io/data/foundation/import/batches/{BATCH_ID}?action=REVERT \
-H 'Authorization: Bearer {ACCESS_TOKEN}' \
-H 'x-gw-ims-org-id: {ORG_ID}' \
-H 'x-api-key: {API_KEY}' \
-H 'x-sandbox-name: {SANDBOX_NAME}'
响应
200 OK
修补批次
有时候,可能需要更新您组织的配置文件存储区中的数据。 例如,您可能需要更正记录或更改属性值。 Adobe Experience Platform支持通过更新插入操作或“修补批次”来更新或修补配置文件存储区数据。
要修补批次,需要满足以下条件:
- 为配置文件和属性更新启用的数据集。 这是通过数据集标记完成的,需要特定
isUpsert:true
标签添加到unifiedProfile
数组。 有关显示如何创建数据集或配置现有数据集进行更新插入的详细步骤,请阅读的教程 为配置文件更新启用数据集. - 包含要修补的字段和配置文件的标识字段的Parquet文件。 为批次打补丁的数据格式与常规批次摄取过程类似。 所需的输入是一个Parquet文件,除了要更新的字段外,上传的数据必须包含身份字段,以便与配置文件存储区中的数据匹配。
在为Profile和upsert启用了数据集,并且有一个Parquet文件包含要修补的字段以及必要的标识字段后,您可以执行以下步骤 正在摄取Parquet文件 以通过批量摄取完成修补程序。
重播批次
如果要替换已摄取的批次,可以使用“批量重播”来执行此操作 — 此操作等同于删除旧批次并摄取新批次。
创建批次
首先,您需要创建一个批次,以JSON作为输入格式。 创建批次时,您需要提供数据集ID。 您还需要确保作为批次的一部分上传的所有文件符合链接到所提供数据集的XDM架构。 此外,您还需要提供旧批次作为重播部分中的引用。 在下面的示例中,您将使用ID重播批次 batchIdA
和 batchIdB
.
API格式
POST /batches
请求
curl -X POST https://platform.adobe.io/data/foundation/import/batches \
-H 'Authorization: Bearer {ACCESS_TOKEN}' \
-H 'Content-Type: application/json' \
-H 'x-gw-ims-org-id: {ORG_ID}' \
-H 'x-api-key: {API_KEY}' \
-H 'x-sandbox-name: {SANDBOX_NAME}'
-d '{
"datasetId": "{DATASET_ID}",
"inputFormat": {
"format": "json"
},
"replay": {
"predecessors": ["${batchIdA}","${batchIdB}"],
"reason": "replace"
}
}'
{DATASET_ID}
响应
201 Created
{
"id": "{BATCH_ID}",
"imsOrg": "{ORG_ID}",
"updated": 0,
"status": "loading",
"created": 0,
"relatedObjects": [
{
"type": "dataSet",
"id": "{DATASET_ID}"
}
],
"replay": {
"predecessors": [
"batchIdA", "batchIdB"
],
"reason": "replace"
},
"version": "1.0.0",
"tags": {},
"createdUser": "{USER_ID}",
"updatedUser": "{USER_ID}"
}
{BATCH_ID}
{DATASET_ID}
{USER_ID}
上传文件
现在您已经创建了批次,接下来可以使用 batchId
从之前将文件上传到批处理。 您可以将多个文件上载到批处理。
API格式
PUT /batches/{BATCH_ID}/datasets/{DATASET_ID}/files/{FILE_NAME}
{BATCH_ID}
{DATASET_ID}
{FILE_NAME}
请求
curl -X PUT https://platform.adobe.io/data/foundation/import/batches/{BATCH_ID}/datasets/{DATASET_ID}/files/{FILE_NAME}.json \
-H 'Authorization: Bearer {ACCESS_TOKEN}' \
-H 'Content-Type: application/octet-stream' \
-H 'x-gw-ims-org-id: {ORG_ID}' \
-H 'x-api-key: {API_KEY}' \
-H 'x-sandbox-name: {SANDBOX_NAME}' \
--data-binary "@{FILE_PATH_AND_NAME}.json"
{FILE_PATH_AND_NAME}
acme/customers/campaigns/summer.json
.响应
200 OK
完成批处理
上传完文件的所有不同部分后,您需要表明数据已完全上传,并且批次已准备好进行升级。
API格式
POST /batches/{BATCH_ID}?action=COMPLETE
{BATCH_ID}
请求
curl -X POST https://platform.adobe.io/data/foundation/import/batches/{BATCH_ID}?action=COMPLETE \
-H 'Authorization: Bearer {ACCESS_TOKEN}' \
-H 'x-gw-ims-org-id: {ORG_ID}' \
-H 'x-api-key: {API_KEY}' \
-H 'x-sandbox-name: {SANDBOX_NAME}'
响应
200 OK
附录
以下部分包含有关使用批量摄取在Experience Platform中摄取数据的其他信息。
批量摄取的数据转换
为了将数据文件摄取到 Experience Platform,文件的层次结构必须符合 体验数据模型(XDM) 与要上载到的数据集关联的架构。
有关如何映射CSV文件以符合XDM架构的信息,请参阅 示例转换 文档,以及格式正确的JSON数据文件示例。 文档中提供的示例文件可在此处找到: