批量摄取开发人员指南
本文档提供了在Adobe Experience Platform中使用批量摄取API端点的全面指南。 有关批量摄取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
可以通过以下对要删除的批次ID使用action=REVERT
查询参数执行的POST请求来删除批次。 该批次被标记为“不活动”,因此可用于垃圾收集。 该批次将异步收集,然后将其标记为“已删除”。
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支持通过更新插入操作或“修补批次”来更新或修补配置文件存储数据。
要修补批次,需要满足以下条件:
- 为配置文件和属性更新启用的数据集。 这是通过数据集标记完成的,需要向
unifiedProfile
数组添加特定的isUpsert:true
标记。 有关说明如何创建数据集或配置现有数据集以进行更新插入的详细步骤,请参阅启用数据集以进行配置文件更新的教程。 - 包含要修补的字段和配置文件标识字段的Parquet文件。 为批次打补丁的数据格式与常规批次摄取过程类似。 必需的输入是一个Parquet文件,除了要更新的字段之外,上传的数据还必须包含身份字段,以便与配置文件存储中的数据匹配。
一旦为配置文件和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数据文件示例。 文档中提供的示例文件可在此处找到: