Salesforce Bulk API の極意：アーキテクトのための大量データ処理戦略

日付:

背景と応用シーン

Salesforceプラットフォームを効果的に活用する上で、大量のデータを扱う要件は避けて通れません。日々の業務では、数十、数百といった単位のレコードを扱うことが一般的ですが、初期データ移行、既存システムとの定期的なデータ同期、大規模なデータ更新・削除といったシナリオでは、数万、数十万、場合によっては数百万という膨大なレコードを処理する必要があります。 このような大量データ処理の要件に対し、Salesforceの標準API（REST APIなど）は、同期的な処理モデルとレコード数に制限があるため、効率的ではありません。同期処理では、各リクエストが完了するまで次の処理が待機するため、処理時間が長くなり、ネットワークの遅延やAPIコール数の制限によって、全体のパフォーマンスが著しく低下する可能性があります。

ここで真価を発揮するのが、Salesforce Bulk API（バルクAPI）です。Bulk APIは、大量のデータを非同期的に処理するために設計された特殊なAPIであり、高効率なデータ連携を実現します。主な応用シーンとしては、以下のようなものが挙げられます。

  • 初期データ移行（Initial Data Migration）：レガシーシステムからSalesforceへの移行時に、既存の顧客、商品、商談データなどを一度に大量にインポートする場合。
  • 定期的なETLプロセス（Extract, Transform, Load Process）：外部のデータウェアハウスや基幹システムからSalesforceへ、あるいはSalesforceから外部システムへ、定期的にデータを連携・同期する場合。
  • 外部システムとのデータ同期（Data Synchronization with External Systems）：例えば、Webサイトからの大量のリード情報や、ERPシステムからの受注情報をSalesforceに一括で取り込む場合。
  • 大規模なデータ更新・削除（Mass Updates/Deletes）：組織全体の製品価格の一括更新や、特定の条件に合致する不要なレコードの一括削除など。

Bulk APIは、これらのシナリオにおいて、従来のAPIでは困難であった、または非効率であったデータ操作を強力にサポートします。その非同期処理（asynchronous processing）とジョブベース（job-based）のアーキテクチャが、大量データ処理の鍵となります。

原理説明

Bulk APIは、その名の通り「バルク」、つまり「大量」のデータ処理に特化したRESTful APIです。標準のREST APIが個々のレコードに対する同期的な操作（create, read, update, delete）を主とするのに対し、Bulk APIはデータ操作をジョブ（Job）として定義し、そのジョブに複数のバッチ（Batch）を割り当てて非同期に実行するモデルを採用しています。

非同期処理（Asynchronous Processing）とは、リクエストを送信した後、すぐに結果を待たずに次の処理に進む方式です。Salesforceのサーバー側でジョブがバックグラウンドで処理され、完了後にその結果を確認できます。これにより、クライアント側のアプリケーションが長時間待機する必要がなくなり、効率的なリソース利用が可能になります。

Bulk API 1.0とBulk API 2.0の二つのバージョンが存在しますが、現在ではシンプルさと効率性からBulk API 2.0の利用が推奨されています。Bulk API 2.0では、バッチの自動処理や、よりシンプルなAPI設計により、開発者の負担が軽減されています。本記事ではBulk API 2.0を中心に説明します。

Bulk API 2.0の処理フローは、以下の主要なステップで構成されます。

1. ジョブの作成（Create a Job）

まず、実行したい操作（挿入、更新、削除、アップサート）と対象オブジェクトを指定して、インジェストジョブ（Ingest Job）を作成します。この際、データのフォーマット（CSVが推奨）、外部ID（Upsertの場合）なども指定します。

2. データバッチのアップロード（Upload Data Batches）

作成したジョブに対して、CSV形式のデータをストリームとしてアップロードします。Bulk API 2.0では、開発者が明示的にバッチを分割する必要はありません。Salesforceが内部的にデータを最適化されたバッチに分割し、並行して処理します。これにより、大規模なデータファイルでも効率的に処理が進められます。

3. ジョブのクローズ（Close the Job）

すべてのデータアップロードが完了したら、ジョブをクローズします。これにより、Salesforceはアップロードされた全データの処理を開始します。

4. ジョブステータスの監視（Monitor Job Status）

ジョブは非同期で実行されるため、定期的にジョブのステータスを問い合わせて、処理の進捗を確認する必要があります。ステータスには、Open, UploadComplete, InProgress, Aborted, Failed, JobCompleteなどがあります。

5. 結果の取得（Retrieve Results）

ジョブが完了（JobComplete）すると、成功したレコード、失敗したレコード、未処理のレコードの詳細をCSV形式でダウンロードできます。失敗したレコードには、エラーメッセージが含まれるため、問題の特定とデバッグに役立ちます。

これらのステップを通じて、Bulk APIは大量のデータをSalesforce内外で効率的に移動させることを可能にし、システムインテグレーションやデータ管理の強力な基盤となります。

サンプルコード

ここでは、Bulk API 2.0を使用したAccountオブジェクトへのデータ挿入を例に、cURLコマンドによる操作手順を示します。これらの例は、Salesforceの公式開発者ドキュメントに基づいています。

事前準備として、Salesforce組織への認証とアクセストークンの取得が必要です。通常はOAuth 2.0フロー（例えばJWT Bearer FlowやWeb Server Flow）を利用して、セッションIDまたはアクセストークンを取得します。ここでは、YOUR_ACCESS_TOKEN、YOUR_INSTANCE_URL、YOUR_API_VERSIONを各自の環境に合わせて置き換えてください。

1. ジョブの作成（Create a Job）

Accountオブジェクトにデータを挿入するためのジョブを作成します。CSV形式のデータを受け入れるように設定します。 

curl -X POST "YOUR_INSTANCE_URL/services/data/vYOUR_API_VERSION/jobs/ingest" \
-H "Authorization: Bearer YOUR_ACCESS_TOKEN" \
-H "Content-Type: application/json" \
-d '{
    "object": "Account",
    "contentType": "CSV",
    "operation": "insert"
}'

コメント:

  • YOUR_INSTANCE_URL: SalesforceのインスタンスURL (例: https://your-domain.my.salesforce.com)
  • YOUR_API_VERSION: 使用するAPIバージョン (例: 58.0)
  • Authorization: Bearer YOUR_ACCESS_TOKEN: 認証に使用するアクセストークン。
  • object: "Account": 操作対象のSalesforceオブジェクト。
  • contentType: "CSV": アップロードするデータの形式。Bulk API 2.0ではCSVが推奨されます。
  • operation: "insert": 実行するデータ操作の種類（insert, update, upsert, delete）。
このリクエストが成功すると、レスポンスとしてjobIdを含むJSONが返されます。このjobIdは以降の操作で必要になります。

2. データバッチのアップロード（Upload Data Batches）

作成したジョブにCSVデータをアップロードします。YOUR_JOB_IDを前のステップで取得したIDに置き換えます。ファイルは通常のHTTP POSTリクエストのボディとして送信します。 

curl -X PUT "YOUR_INSTANCE_URL/services/data/vYOUR_API_VERSION/jobs/ingest/YOUR_JOB_ID/batches" \
-H "Authorization: Bearer YOUR_ACCESS_TOKEN" \
-H "Content-Type: text/csv" \
--data-binary "@./accounts.csv"

コメント:

  • YOUR_JOB_ID: 前のステップで取得したジョブID。
  • Content-Type: text/csv: アップロードするデータのMIMEタイプ。
  • --data-binary "@./accounts.csv": accounts.csvという名前のファイルからCSVデータを読み込み、リクエストボディとして送信します。ファイルのパスは適宜調整してください。CSVファイルはヘッダー行を含み、Salesforceのフィールド名と一致させる必要があります。

accounts.csv の例: 

Name,BillingCity,BillingState
Test Account 1,San Francisco,CA
Test Account 2,New York,NY

3. ジョブのクローズ（Close the Job）

すべてのデータアップロードが完了したら、ジョブをクローズしてSalesforceに処理開始を指示します。 

curl -X PATCH "YOUR_INSTANCE_URL/services/data/vYOUR_API_VERSION/jobs/ingest/YOUR_JOB_ID" \
-H "Authorization: Bearer YOUR_ACCESS_TOKEN" \
-H "Content-Type: application/json" \
-d '{
    "state": "UploadComplete"
}'

コメント:

  • state: "UploadComplete": ジョブの状態を「UploadComplete」に設定することで、Salesforceにデータ処理を開始するよう通知します。

4. ジョブステータスの監視（Monitor Job Status）

ジョブのステータスを定期的に確認し、処理の進捗を監視します。 

curl -X GET "YOUR_INSTANCE_URL/services/data/vYOUR_API_VERSION/jobs/ingest/YOUR_JOB_ID" \
-H "Authorization: Bearer YOUR_ACCESS_TOKEN"

コメント:

  • このリクエストは、ジョブの現在の状態（state）、処理されたレコード数（numberRecordsProcessed）、失敗したレコード数（numberRecordsFailed）などの詳細を含むJSONレスポンスを返します。
  • stateがJobCompleteまたはFailedになるまで、このリクエストを定期的に実行します。

5. 結果の取得（Retrieve Results）

ジョブが完了したら、成功したレコードと失敗したレコードの詳細をダウンロードできます。

成功したレコードの取得

curl -X GET "YOUR_INSTANCE_URL/services/data/vYOUR_API_VERSION/jobs/ingest/YOUR_JOB_ID/successfulResults" \
-H "Authorization: Bearer YOUR_ACCESS_TOKEN" \
-H "Accept: text/csv"

コメント:

  • Accept: text/csv: レスポンス形式をCSVとして指定します。
  • このリクエストは、成功した各レコードのIDと成功ステータスを含むCSVファイルを返します。

失敗したレコードの取得

curl -X GET "YOUR_INSTANCE_URL/services/data/vYOUR_API_VERSION/jobs/ingest/YOUR_JOB_ID/failedResults" \
-H "Authorization: Bearer YOUR_ACCESS_TOKEN" \
-H "Accept: text/csv"

コメント:

  • このリクエストは、失敗した各レコードのID、エラーメッセージ、および元のデータを含むCSVファイルを返します。この情報を使用して、データの問題を特定し、修正できます。

注意事項

Salesforce Bulk APIを効果的かつ安全に利用するためには、いくつかの重要な注意点を理解しておく必要があります。

権限（Permissions）

Bulk APIを利用するユーザーは、適切な権限を持っている必要があります。

  • API参照の有効化（API Enabled）: プロファイルまたは権限セットで「API の有効化」が許可されている必要があります。
  • オブジェクト権限（Object Permissions）: 対象となるSalesforceオブジェクト（例: Account, Contact）に対する「参照（Read）」、「作成（Create）」、「編集（Edit）」、「削除（Delete）」のいずれか、またはすべての権限が必要です。実行する操作（insert, update, deleteなど）に応じて適切な権限が付与されていることを確認してください。

API制限（API Limits）

Salesforceは、システムの安定性と公平なリソース利用を保つために、様々なAPI制限（API Limits）を設けています。Bulk APIもこれらの制限の対象となります。

  • APIリクエスト制限（Daily API Request Limits）: Salesforce組織全体で24時間あたりに実行できるAPIリクエストの総数に制限があります。Bulk APIの各コールもこの制限にカウントされます。
  • 並行ジョブ数（Concurrent Jobs）: 一度に実行できるジョブの数には制限があります。多数のジョブを同時に実行しようとすると、キューに入れられるか、エラーが発生する可能性があります。
  • ジョブごとのファイルサイズ/レコード数: Bulk API 2.0では、データファイルのアップロードに関するいくつかの制限があります。例えば、1つのジョブでアップロードできる総データサイズや、単一のCSVファイル内の最大レコード数などです。これらの制限はSalesforceのバージョンやエディションによって異なる場合があるため、最新のドキュメントで確認することが重要です。

これらの制限を超過すると、APIコールが拒否されたり、ジョブが失敗したりする可能性があります。利用計画を立てる際には、SalesforceのAPI制限ドキュメントを必ず参照し、自社の要件に合わせた適切な設計を行うことが不可欠です。

エラー処理（Error Handling）

大量のデータを処理する際にエラーはつきものです。適切なエラー処理戦略が不可欠です。

  • 結果ファイルの確認: ジョブ完了後、successfulResultsとfailedResultsのファイルを必ずダウンロードし、内容を詳細に確認します。特にfailedResultsファイルには、失敗した理由を示すエラーメッセージが含まれており、問題解決の重要な手がかりとなります。
  • HTTPステータスコード: APIリクエストのHTTPステータスコード（例: 4xx, 5xx）を監視し、認証エラー、不正なリクエスト、サーバーエラーなどの即時エラーに対応します。
  • 部分的な成功: Bulk APIは部分的な成功（一部のレコードは成功し、一部は失敗）を許容します。すべてのレコードが処理されたかどうか、また成功率を分析し、必要に応じて失敗したレコードに対する再試行ロジックを実装します。

データ品質（Data Quality）

Bulk APIを使ってデータを挿入・更新する際、Salesforceの入力規則（Validation Rules）、ワークフロールール、Apexトリガーなどが適用されます。これらのビジネスロジックによってデータが無効と判断された場合、レコードは失敗として処理されます。

  • アップロード前にデータのクリーンアップと検証を行い、Salesforceのデータモデルとビジネスルールに合致していることを確認することが重要です。
  • 特に、参照整合性（参照関係のあるオブジェクトのIDの存在確認など）は重要なチェックポイントです。

パフォーマンス（Performance）

Bulk APIは高効率ですが、外部システムからのネットワークレイテンシーや、Salesforce組織内のカスタマイズ（複雑なトリガー、大規模なワークフロー、ロールアップサマリーフィールドなど）によっては、処理速度が影響を受ける可能性があります。大規模なデータ処理を行う前に、サンドボックス環境でテストを行い、パフォーマンスボトルネックを特定し、最適化を図ることが推奨されます。

まとめとベストプラクティス

Salesforce Bulk APIは、その強力な非同期処理能力により、大規模なデータ操作を効率的かつ信頼性の高い方法で実行するための不可欠なツールです。データ移行、システム連携、一括更新といった多様なシナリオにおいて、標準APIの限界を超えるパフォーマンスを提供します。

最後に、Bulk APIを最大限に活用するためのベストプラクティスをまとめます。

1. Bulk API 2.0の利用

可能な限り、Bulk API 2.0を使用してください。これは、APIのシンプルさ、自動バッチ処理、改善されたエラーレポート機能により、開発と運用が大幅に容易になります。明示的なバッチ管理が不要なため、開発者はデータファイルそのものに集中できます。

2. CSV形式の活用

データアップロードには、CSV（Comma Separated Values）形式を推奨します。CSVは、JSONやXMLよりもデータサイズが小さく、解析が高速であるため、Bulk APIのパフォーマンスを最大化するのに最も効率的な形式です。

3. 外部ID（External ID）の最適化された利用

Upsert操作（挿入と更新を兼ねる操作）を実行する際には、外部ID（External ID）フィールドを積極的に活用してください。外部IDは、Salesforceの標準IDの代わりに外部システムでレコードを一意に識別するために使用できるカスタムフィールドです。外部IDを使用することで、Salesforceはレコードを高速に検索し、挿入または更新を効率的に実行できます。これにより、Lookup関係の解決も簡素化されます。

4. エラーログの厳密な監視と分析

ジョブ完了後は、必ずエラーログ（failedResults）をダウンロードし、失敗したレコードとエラーメッセージを詳細に分析してください。この情報は、データの品質問題を特定し、再試行の戦略を立てる上で非常に重要です。エラー原因を突き止め、データソースの修正やSalesforceのビジネスロジックの調整を行うことで、将来的なエラーを減らすことができます。

5. サンドボックス環境での徹底的なテスト

本番環境（Production Environment）で大規模なBulk API操作を実行する前に、必ずサンドボックス環境（Sandbox Environment）で徹底的なテストを実施してください。これにより、API制限、データ品質の問題、予期せぬトリガーや自動化による影響を事前に特定し、リスクを軽減できます。特に、パフォーマンスのボトルネックやガバナ制限（Governor Limits）への影響を評価することが重要です。

6. ガバナ制限と自動化への配慮

Bulk APIによるデータ操作は、Salesforce組織内のApexトリガー、ワークフロー、プロセスビルダー、フローなどの自動化ロジックを起動します。これらの自動化は、大量のレコードが処理される際に、予期せぬガバナ制限（例: SOQLクエリ制限、CPU時間制限）に抵触し、ジョブが失敗する原因となることがあります。大規模なデータロード時には、これらの自動化がBulk APIの処理にどのような影響を与えるかを事前に分析し、必要に応じて一時的に無効化する、またはバッチ処理に最適化されたロジックに調整するなどの対策を検討してください。

これらのベストプラクティスを遵守することで、Salesforce Bulk APIを最大限に活用し、ビジネスの成長を支える堅牢で効率的なデータ連携ソリューションを構築することができます。

コメント

コメントを投稿