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

十月 28, 2025

背景与应用场景

大家好，我是一名 Salesforce 集成工程师。在我的日常工作中，最常遇到的挑战之一就是如何在不同系统之间高效、可靠地移动大量数据。当数据量从几百条上升到数百万甚至上亿条时，传统的同步方法，如逐条调用的 REST API 或 SOAP API，就显得力不心裁。它们的同步特性、较低的事务限制和 API 调用次数限制，使得处理大规模数据集变得非常缓慢且容易失败。

这正是 Bulk API 发挥作用的地方。Salesforce Bulk API 是一个基于 REST 的专用接口，专为异步处理大量记录的加载、查询、更新和删除而设计。今天，我们将深入探讨其最新版本——Bulk API 2.0。与它的前身 Bulk API 1.0 相比，2.0 版本极大地简化了工作流程，让我们集成工程师能更专注于数据本身，而不是复杂的批处理管理。

典型的应用场景包括：

1. 初始数据迁移： 当企业首次实施 Salesforce 时，需要将来自旧 CRM、ERP 或其他遗留系统的大量客户、联系人、产品等数据一次性导入 Salesforce。这通常涉及数百万条记录。

2. 定期数据同步： 企业需要定期将财务系统中的订单数据、数据仓库中的客户分析结果或营销自动化平台的活动记录同步到 Salesforce。这些任务通常在夜间执行，数据量巨大。

3. 数据归档与清理： 随着时间的推移，Salesforce 组织中会积累大量非活动数据。使用 Bulk API 2.0 可以高效地查询并导出这些数据进行归档，或执行大规模的数据清理和去重操作。

4. 与大数据平台集成： 将 Salesforce 数据导出到 Snowflake、BigQuery 等数据湖或数据仓库进行深度分析，或将分析结果回写到 Salesforce，都需要 Bulk API 2.0 强大的数据处理能力。

作为一名集成工程师，掌握 Bulk API 2.0 是构建稳健、可扩展的 Salesforce 数据集成解决方案的关键技能。

原理说明

Bulk API 2.0 的核心设计理念是简化和自动化。它与 Bulk API 1.0 最大的区别在于，开发者不再需要手动将数据分割成多个批次 (batch) 并分别上传。在 2.0 中，你只需要创建一个作业 (Job)，上传完整的 CSV 文件，然后关闭作业即可。Salesforce 会在后端自动为你处理数据的分块和并行处理，极大地降低了集成的复杂性。

整个工作流程是异步的 (asynchronous)，这意味着你提交一个作业后，不会立即得到处理结果。你需要通过轮询 (polling) 的方式来检查作业的状态，直到它完成或失败。这对于需要长时间运行的大数据任务来说是至关重要的设计。

一个典型的 Bulk API 2.0 数据加载 (Ingest) 流程如下：

1. 创建作业 (Create a Job)： 你向 Salesforce 发送一个 POST 请求，指定要操作的对象 (e.g., `Account`)、操作类型 (e.g., `insert`, `update`, `upsert`, `delete`) 以及数据格式 (CSV)。Salesforce 会返回一个唯一的作业 ID (Job ID)。

2. 上传数据 (Upload Job Data)： 你将准备好的 CSV 数据文件通过一个 PUT 请求上传到指定的端点，该端点包含了上一步获取的 Job ID。这个步骤可以一次性上传所有数据。

3. 关闭作业 (Close the Job)： 数据上传完成后，你发送一个 PATCH 请求，将作业的状态从 `Open` 更新为 `UploadComplete`。这个动作会通知 Salesforce：“我的数据已经全部上传完毕，请开始处理吧。”

4. 监控作业状态 (Monitor Job Status)： 你需要定期发送 GET 请求来查询作业的状态。作业状态会经历 `UploadComplete` -> `InProgress` -> `JobComplete` (成功) 或 `Failed` (失败) 的转变。

5. 获取处理结果 (Get Job Results)： 作业完成后，你可以通过专门的端点下载处理结果。结果分为两个文件：一个是成功处理的记录列表，另一个是处理失败的记录列表。失败文件中会包含具体的错误信息 (e.g., `sf__Error` 列)，这对于调试和数据修正至关重要。

对于数据查询 (Query) 操作，流程类似，只是不需要上传数据，而是提交一个 SOQL 查询语句，然后等待 Salesforce 准备好结果集供你下载。

示例代码

以下示例将演示如何使用 cURL 命令通过 Bulk API 2.0 插入一组新的客户 (Account) 记录。这些示例严格遵循 Salesforce 官方文档，请确保将 `MyDomainName`、`4X.0` 和 `YOUR_SESSION_ID` 替换为你的实际值。

1. 创建一个插入作业

我们首先创建一个作业，告诉 Salesforce 我们准备插入 `Account` 对象的数据。`lineEnding` 字段指定了 CSV 文件中的换行符类型，`LF` 是最常见的选择。

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

成功的响应会返回一个 JSON 对象，其中包含了我们后续步骤需要的 `id` (Job ID)。

2. 上传 CSV 数据

假设我们创建了一个名为 `account_data.csv` 的文件，内容如下：

Name,Industry,NumberOfEmployees
"Bulk API 2.0 Account 1","Technology",500
"Bulk API 2.0 Account 2","Finance",2000

现在，我们使用上一步返回的 Job ID 来上传这个文件。注意，请求方法是 `PUT`，并且 `Content-Type` 必须是 `text/csv`。

curl -X PUT https://MyDomainName.my.salesforce.com/services/data/v4X.0/jobs/ingest/YOUR_JOB_ID/batches \
-H "Authorization: Bearer YOUR_SESSION_ID" \
-H "Content-Type: text/csv" \
--data-binary @account_data.csv

3. 关闭作业

数据上传完毕后，我们通过发送一个 PATCH 请求来关闭作业，通知 Salesforce 开始处理。

curl -X PATCH https://MyDomainName.my.salesforce.com/services/data/v4X.0/jobs/ingest/YOUR_JOB_ID \
-H "Authorization: Bearer YOUR_SESSION_ID" \
-H "Content-Type: application/json; charset=UTF-8" \
-d '{
  "state" : "UploadComplete"
}'

4. 检查作业状态

现在，我们需要轮询作业状态，直到 `state` 变为 `JobComplete` 或 `Failed`。

curl -X GET https://MyDomainName.my.salesforce.com/services/data/v4X.0/jobs/ingest/YOUR_JOB_ID \
-H "Authorization: Bearer YOUR_SESSION_ID"

响应会包含作业的详细信息，包括 `state`、`recordsProcessed`、`numberRecordsFailed` 等关键指标。

5. 获取成功和失败的结果

作业完成后，我们可以分别下载成功和失败的记录。这些结果通常是以 CSV 格式提供的。

获取成功记录：

curl -X GET https://MyDomainName.my.salesforce.com/services/data/v4X.0/jobs/ingest/YOUR_JOB_ID/successfulResults/ \
-H "Authorization: Bearer YOUR_SESSION_ID"

获取失败记录：

curl -X GET https://MyDomainName.my.salesforce.com/services/data/v4X.0/jobs/ingest/YOUR_JOB_ID/failedResults/ \
-H "Authorization: Bearer YOUR_SESSION_ID"

失败结果的 CSV 文件会比原始文件多一列 `sf__Error`，其中详细说明了每条记录失败的原因，例如“REQUIRED_FIELD_MISSING”（必填字段缺失）等。

注意事项

作为集成工程师，在设计和实施 Bulk API 2.0 解决方案时，必须考虑以下几点：

1. 权限 (Permissions)： 执行 API 调用的集成用户必须拥有 "API Enabled" 系统权限。此外，为了创建、监控和中止批量作业，用户需要 "Manage Data Integrations" 权限。当然，用户还必须对操作的对象和字段拥有相应的 CRUD (Create, Read, Update, Delete) 权限。

2. API 限制 (API Limits)： Bulk API 2.0 受 Salesforce 治理限制的约束。最重要的是每日记录处理限制，根据 Salesforce 版本，通常是每天 1.5 亿条记录。还有一个不太为人所知但同样重要的限制：在滚动的 15 分钟内，一个组织最多只能提交 15,000 个批次 (batch) 的数据。虽然 Bulk API 2.0 会自动创建批次，但上传一个非常大的文件（例如，超过 150GB）可能会触及这个限制。因此，对于超大规模的数据，建议在客户端将文件拆分为几个较小的文件，并创建多个作业。

3. 错误处理 (Error Handling)： 稳健的错误处理机制是集成成功的关键。你的集成流程必须能够：

正确解析 `failedResults` CSV 文件。
根据 `sf__Error` 列中的错误信息对失败原因进行分类。
实现重试逻辑。例如，对于由记录锁定 (record locking) 引起的临时性错误，可以稍后重试。对于数据格式错误，应将记录记录到异常队列中，供人工审查。
在作业本身失败 (state: `Failed`) 时，能够从作业信息中获取错误详情并发出警报。

4. 数据格式与大小： Bulk API 2.0 严格要求使用 UTF-8 编码的 CSV 文件。单个上传的 CSV 文件大小不能超过 150 MB。确保你的 CSV 文件头部（第一行）与 Salesforce 对象的 API 名称完全匹配。

5. PK Chunking for Queries： 在使用 Bulk API 2.0 执行查询时，如果目标表非常大（例如，超过 1000 万条记录的 `Task` 或 `Account` 对象），标准查询可能会超时。这时应使用 PK Chunking。通过在查询请求头中添加 `Sforce-Enable-PKChunking: TRUE`，你可以指示 Salesforce 基于记录的主键 (Primary Key) 自动将查询分割成多个小块并行执行，从而显著提高大表查询的成功率和性能。

总结与最佳实践

Bulk API 2.0 是 Salesforce 提供的一款强大工具，它彻底改变了我们集成工程师处理大规模数据的方式。其简化的 RESTful 框架和自动化的批处理机制，使我们能够构建更高效、更可靠、更易于维护的数据集成解决方案。

以下是我在实践中总结的最佳实践：

选择正确的工具： 当记录数超过 2,000-10,000 条时，优先考虑 Bulk API 2.0。对于少量实时性要求高的数据，标准 REST API 可能是更好的选择。
优化数据准备： 在上传数据之前，尽可能在源系统或 ETL (Extract, Transform, Load) 工具中完成数据清洗、转换和验证。这可以最大限度地减少 Salesforce 端的处理错误，节省宝贵的 API 处理时间和调试精力。
实施智能轮询： 在监控作业状态时，不要以固定且过于频繁的间隔进行轮询。采用指数退避 (exponential backoff) 策略，即每次查询失败或状态未变时，将下一次查询的等待时间加倍，这可以有效避免耗尽 API 调用次数。
并行处理，但需谨慎： 如果你需要处理海量数据，可以创建多个作业并行执行以提高吞吐量。但要警惕，对同一对象进行大量的并行写入操作可能会导致记录锁定竞争，从而增加失败率。
日志与监控： 建立完善的日志系统，记录每个 Bulk 作业的 ID、状态、处理记录数和失败记录数。集成监控系统（如 Datadog, Splunk）来跟踪这些指标，并在作业失败或失败率超过阈值时自动告警。

通过遵循这些原则并充分利用 Bulk API 2.0 的强大功能，你可以充满信心地应对任何大规模 Salesforce 数据集成的挑战。

Salesforce中文网