Salesforce Bulk API 2.0 大规模数据集成：集成工程师综合指南

背景与应用场景

大家好，我是一名 Salesforce 集成工程师。在我的日常工作中，核心任务之一就是确保 Salesforce 与企业内部的其他系统（如 ERP、数据仓库、营销自动化平台等）之间能够高效、可靠地交换数据。当数据量较小时，标准的 REST API 或 SOAP API 足以胜任实时或近实时的同步任务。但当我们面临大规模数据处理场景时，例如：

初始数据迁移：将旧系统中的数百万条客户、订单或产品数据一次性导入 Salesforce。
定期批量同步：每晚从数据仓库同步数十万条销售记录到 Salesforce 以更新报表和仪表盘。
数据归档或清理：定期从 Salesforce 中导出或删除大量陈旧数据。
与ETL工具集成：作为数据抽取、转换和加载（Extract, Transform, Load）流程的一部分，与 MuleSoft, Informatica, Talend 等工具集成。

在这些场景下，逐条调用标准 API 不仅效率低下，而且会极快地消耗掉组织的每日 API 调用限制。这正是 Salesforce Bulk API（批量 API）发挥其核心价值的地方。Bulk API 专为异步处理大量数据集而设计，它将数据分块处理，优化了服务器资源，并提供了更高的 API 调用限额，是所有大规模数据集成项目的首选工具。

本文将以集成工程师的视角，深入探讨 Bulk API 2.0，这是目前推荐使用的版本。我们将从其工作原理入手，结合官方代码示例，详细阐述其使用方法、关键注意事项以及在集成项目中的最佳实践。

原理说明

Bulk API 2.0 相比于其前身（Bulk API 1.0），在架构上进行了简化，为开发者提供了更加流畅和统一的体验。其核心思想是异步处理。作为集成工程师，理解这个异步流程至关重要，因为它直接影响我们如何设计集成作业的轮询、监控和错误处理逻辑。

整个流程完全基于 RESTful 原则，使用 JSON 或 CSV 格式传输数据。一个典型的 Bulk API 2.0 作业（Job）遵循以下步骤：

创建作业 (Create a Job)：客户端向 Salesforce 发送一个 HTTP POST 请求，定义要执行的操作（如 `insert`, `update`, `upsert`, `delete`, `hardDelete`），操作的对象（如 `Account`, `Contact`），以及数据格式（`CSV`, `JSON`）。Salesforce 收到请求后，会创建一个作业，并返回一个唯一的作业 ID 和一个用于上传数据的临时存储 URL。
上传数据 (Upload Data)：客户端将准备好的 CSV 或 JSON 数据文件，通过一个 HTTP PUT 请求上传到上一步返回的 URL。Bulk API 2.0 简化了数据分块（Chunking）的过程，我们只需上传完整的数据文件，Salesforce 会在后台自动将其分割成多个小块进行并行处理。
关闭作业 (Close the Job)：数据上传完成后，客户端需要发送一个 HTTP PATCH 请求，将作业的状态更新为 `UploadComplete`。这个操作会通知 Salesforce：“数据已准备就绪，请开始处理”。一旦作业关闭，就不能再上传任何数据。
监控作业状态 (Monitor Job Status)：由于处理是异步的，客户端需要通过定期发送 HTTP GET 请求来轮询作业的状态。作业状态会经历 `Open` -> `UploadComplete` -> `InProgress` -> `JobComplete`（或 `Failed`, `Aborted`）等阶段。
获取结果 (Retrieve Results)：当作业状态变为 `JobComplete` 时，客户端可以下载处理结果。Salesforce 会提供三个独立的结果文件：成功记录、失败记录和未处理记录。失败记录文件中会包含具体的错误信息，这对于调试和数据修正至关重要。

这个框架使得 Bulk API 2.0 能够高效地处理海量数据，同时将对 Salesforce 实时性能的影响降到最低。对于我们集成工程师来说，这意味着我们需要构建一个能够管理这个完整生命周期的客户端程序或集成流程。

示例代码

以下示例将演示如何使用 Bulk API 2.0 插入一组新的客户（Account）记录。我们将使用 cURL 命令来模拟 HTTP 请求，这在集成开发初期的测试和调试阶段非常实用。请将 `MyDomainName` 和 `bearer-token` 替换为您自己的 Salesforce 实例域名和有效的 OAuth 2.0 访问令牌。

第 1 步：创建批量插入作业

我们首先创建一个作业，指定我们要对 `Account` 对象执行 `insert` 操作，数据格式为 `CSV`。

# Endpoint: /services/data/vXX.X/jobs/ingest
# Method: POST
# XX.X 代表您的 Salesforce API 版本, 例如 58.0

curl https://MyDomainName.my.salesforce.com/services/data/v58.0/jobs/ingest \
-H "Authorization: Bearer bearer-token" \
-H "Content-Type: application/json; charset=UTF-8" \
-H "Accept: application/json" \
-d '{
  "object" : "Account",
  "contentType" : "CSV",
  "operation" : "insert",
  "lineEnding" : "LF"
}'

注释：

`object`: 指定要操作的 SObject，这里是 `Account`。
`contentType`: 数据格式，我们使用 `CSV`。
`operation`: 操作类型，这里是 `insert`。
`lineEnding`: 指定行尾符，`LF` (Line Feed) 是推荐的格式。

Salesforce 将返回一个 JSON 响应，其中包含作业的详细信息，最重要的是 `id` 和 `contentUrl`：

{
  "id": "750R0000000z0f9IAA",
  "operation": "insert",
  "object": "Account",
  "createdById": "005R0000000j1bFIAQ",
  "createdDate": "2023-01-10T21:29:43.000+0000",
  "systemModstamp": "2023-01-10T21:29:43.000+0000",
  "state": "Open",
  "concurrencyMode": "Parallel",
  "contentType": "CSV",
  "apiVersion": 58.0,
  "contentUrl": "jobs/ingest/750R0000000z0f9IAA/batches",
  "lineEnding": "LF",
  "columnDelimiter": "COMMA"
}

第 2 步：上传数据

接下来，我们将 CSV 数据上传到上一步返回的 `contentUrl` 指定的端点。注意，这里的请求方法是 `PUT`。

# Endpoint: a 组合的 URL，即实例 URL + contentUrl
# Method: PUT

# 准备一个名为 "request.csv" 的文件，内容如下：
# Name,Description,Phone
# "Test Account 1","Description for Test Account 1","(111) 555-1212"
# "Test Account 2","Description for Test Account 2","(222) 555-1212"

curl https://MyDomainName.my.salesforce.com/services/data/v58.0/jobs/ingest/750R0000000z0f9IAA/batches \
-H "Authorization: Bearer bearer-token" \
-H "Content-Type: text/csv" \
-T "request.csv"

注释：

`-H "Content-Type: text/csv"`: 明确告诉服务器我们上传的是 CSV 数据。
`-T "request.csv"`: cURL 命令，用于将本地文件 `request.csv` 的内容作为请求体上传。

如果上传成功，Salesforce 会返回 `201 Created` 状态码。

第 3 步：关闭作业

数据上传完毕后，必须关闭作业以启动处理流程。

# Endpoint: /services/data/vXX.X/jobs/ingest/{jobId}
# Method: PATCH

curl https://MyDomainName.my.salesforce.com/services/data/v58.0/jobs/ingest/750R0000000z0f9IAA/ \
-H "Authorization: Bearer bearer-token" \
-H "Content-Type: application/json; charset=UTF-8" \
-H "Accept: application/json" \
-d '{
  "state" : "UploadComplete"
}'

注释：

请求体中的 `"state" : "UploadComplete"` 是关键，它触发 Salesforce 开始处理数据。

第 4 步：检查作业状态并获取结果

我们可以轮询作业信息端点来监控进度。

# Endpoint: /services/data/vXX.X/jobs/ingest/{jobId}
# Method: GET

curl https://MyDomainName.my.salesforce.com/services/data/v58.0/jobs/ingest/750R0000000z0f9IAA/ \
-H "Authorization: Bearer bearer-token" \
-H "Accept: application/json"

当返回的 JSON 中的 `state` 字段变为 `JobComplete` 时，表示处理已完成。此时，我们可以下载成功和失败的记录列表。

# 获取成功记录的结果
curl https://MyDomainName.my.salesforce.com/services/data/v58.0/jobs/ingest/750R0000000z0f9IAA/successfulResults/ \
-H "Authorization: Bearer bearer-token" \
-H "Accept: text/csv"

# 获取失败记录的结果
curl https://MyDomainName.my.salesforce.com/services/data/v58.0/jobs/ingest/750R0000000z0f9IAA/failedResults/ \
-H "Authorization: Bearer bearer-token" \
-H "Accept: text/csv"

注释：

成功的结果文件会包含原始数据，并附加 `sf__Id` 和 `sf__Created` 列。
失败的结果文件会包含原始数据，并附加 `sf__Error` 列，其中详细说明了失败原因。

注意事项

作为集成工程师，成功实现一个健壮的 Bulk API 集成不仅仅是调用 API，更重要的是处理好各种边界条件和限制。

权限 (Permissions)

执行集成的用户必须拥有正确的权限。在用户的简档（Profile）或权限集（Permission Set）中，请确保：

API Enabled：这是所有 API 访问的基础权限。
对象和字段权限：用户必须对目标对象拥有创建、读取、更新、删除的权限（CRUD），并对所有涉及的字段具有相应的读写权限。
Bulk API Hard Delete：如果作业操作是 `hardDelete`，则用户必须显式拥有此权限。

API 限制 (API Limits)

Bulk API 2.0 有其自身的限制，必须在集成设计时加以考虑：

作业数量：在 24 小时滚动周期内，最多可以创建 100,000 个作业。
数据大小：上传的原始数据文件大小不能超过 150 MB。
记录数量：一个作业中包含的记录数没有硬性上限，但受文件大小限制。
处理时间：Salesforce 会在系统资源可用时处理作业，没有保证的处理完成时间（SLA）。对于时间敏感的集成，需要设计合理的超时和重试逻辑。

错误处理 (Error Handling)

健壮的错误处理是集成成功的关键。当作业完成后，务必下载并解析 `failedResults` 文件。文件中的 `sf__Error` 列会提供宝贵的调试信息，例如验证规则失败、必填字段缺失、触发器（Trigger）异常等。集成流程应能够：

自动解析错误文件。
根据错误类型对失败记录进行分类。
对于可恢复的错误（如暂时性的记录锁定），实现自动重试机制。
对于数据质量问题，将错误记录和原因记录到日志系统或发送通知，以便人工干预。

并发模式 (Concurrency Mode)

在创建作业时，可以指定 `concurrencyMode` 为 `Parallel`（默认）或 `Serial`。

Parallel（并行）：Salesforce 会将数据块并行处理，速度更快。但如果数据之间存在依赖关系（例如，同时更新同一父记录下的多个子记录），或目标对象上有复杂的自动化逻辑（如触发器、流程），可能会导致记录锁定（Record Locking）错误。

Serial（串行）：

总结与最佳实践

Salesforce Bulk API 2.0 是集成工程师工具箱中不可或缺的利器，是解决大规模数据处理挑战的钥匙。它通过异步、批量的处理模式，在保证 Salesforce 平台稳定性的前提下，提供了卓越的数据吞吐能力。

作为最佳实践，我建议在您的集成项目中遵循以下原则：

明确使用场景：当记录数超过 2,000 条时，优先考虑使用 Bulk API。对于需要实时响应的单条记录操作，继续使用标准 REST API。
数据质量先行：在上传数据之前，在源系统或ETL流程中进行尽可能多的数据清洗和验证。这能显著减少因数据格式、必填字段等问题导致的错误，提高一次性成功率。
设计容错和重试机制：不要假设每次作业都会 100% 成功。您的集成流程必须能够优雅地处理作业失败、部分记录失败的情况，并具备智能重试逻辑。
理解目标 Org 的自动化：在设计集成前，充分了解目标 Salesforce Org 中的触发器、流程、验证规则等自动化逻辑。这些逻辑会影响 Bulk API 的性能，并可能是导致错误的主要原因。与 Salesforce 管理员或开发人员密切沟通至关重要。
监控 API 消耗：将 Salesforce API 限制的监控纳入您的运维体系。通过 Salesforce `Limits` REST API 资源或平台事件来跟踪消耗情况，避免因超出限制而导致服务中断。
选择合适的并发模式：在项目初期就评估数据特性和对象逻辑，明智地选择 `Parallel` 或 `Serial` 模式，以平衡性能和数据一致性。

通过遵循这些原则并深入理解 Bulk API 2.0 的工作机制，您将能够构建出既高效又可靠的 Salesforce 数据集成解决方案，为企业的数据驱动决策提供坚实的基础。

Salesforce中文网