Salesforce Bulk API 2.0 を使ってみる

※ これから記載する事項は、私が所属する会社とは一切関係のない事柄です。

今回は[注文管理](https://developer.salesforce.com/docs/atlas.en-us.order_management_developer_guide_html.meta/order_management_developer_guide_html/order_management_importing_data_into_data_mgmt.htm)などの大量データのインポートで利用される Bulk API 2.0 を利用してみたいと思います。


## Bulk API 2.0 とは（2023/07時点）

まず Bulk API は非同期でレコード操作を行えるため、大量データを扱う際に向いていますが、Bulk API 2.0 は下記の点でさらに優れているようです。（ヘルプ「[Bulk API 2.0 と Bulk API の違いは?](https://developer.salesforce.com/docs/atlas.ja-jp.api_asynch.meta/api_asynch/bulk_common_diff_two_versions.htm)」より）

* クライアント側のコード記述を削減する。
* ジョブ状況の監視を簡単にする。
* 失敗したレコードを自動再試行する。
* 並列処理をサポートする。
* 取得またはクエリワークフローの完了に必要なコール数を削減する。
* バッチ管理を簡素化する。


一方で、あまり大量なデータではない（2000レコード以下）ような場合は、REST (Composite など) または SOAP による「一括」の同期呼び出しを行うことが推奨されています。

:::note info
レコード操作を API を利用して行う場合にどの API を利用したらいいかわからない場合はヘルプ「[どの API を使用するか?](https://help.salesforce.com/s/articleView?id=sf.integrate_what_is_api.htm&type=5)」を参照してみてください。
:::




## 今回やること

今回は下記の CSV ファイルを取り込んで取引先を追加したいと思います。

```text:account.csv
Name,Description,NumberOfEmployees
TestAccount1,Description of TestAccount1,30
TestAccount2,Another description,40
TestAccount3,Yet another description,50
```

完成イメージはこちらです。
![スクリーンショット 2023-07-28 14.44.15.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2110894/acf8e03f-ac45-85dc-9c50-65b52c34fc2b.png)




## 手順

### 1. 接続アプリケーションの作成とアクセストークンの取得

今回は「[OAuth 2.0 ユーザ名パスワードフロー](https://help.salesforce.com/s/articleView?id=sf.remoteaccess_oauth_username_password_flow.htm&type=5)」を利用して API を利用するための認証を行います。

:::note alert
現在では「OAuth 2.0 ユーザ名パスワードフロー」はデフォルトで無効なので他のフローを利用するか、このフローを[有効にする設定](https://help.salesforce.com/s/articleView?id=release-notes.rn_security_username-password_flow_blocked_by_default.htm&release=244&type=5)を行ってください。
![スクリーンショット 2023-08-23 13.31.38.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2110894/1401cc90-cb42-3e7d-a002-e9491fc900ad.png)
:::


##### 接続アプリケーションの作成

**設定 > Lightning Experience アプリケーションマネージャ** に遷移し、右上の [新規接続アプリケーション] をクリックし、下記のような接続アプリケーションを設定しました。
![スクリーンショット 2023-07-28 13.58.18.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2110894/884c84fe-465f-7c16-bbec-3448f5b876fe.png)


**設定 > Lightning Experience アプリケーションマネージャ** から作成したアプリケーションの右横のメニューリストから [参照] を選択しアプリケーションの詳細を表示させた上で [コンシューマの詳細を管理] ボタンをクリックする 

![スクリーンショット 2023-07-28 13.58.36.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2110894/18e84205-2a29-7290-d284-6652ceddd0f5.png)


認証後、下記のように認証情報が表示されるのでコピーしておく。

![スクリーンショット 2023-07-28 13.59.22.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2110894/3623cdca-375d-9689-93ee-0e15fa54aad1.png)



##### アクセストークンの取得

下記のように HTTP リクエストを行います。

```http:Request
POST /services/oauth2/token HTTP/1.1
Host: login.salesforce.com
Content-Type: application/x-www-form-urlencoded

grant_type=password&username={ユーザID}&password={ユーザパスワード}&client_id={接続アプリケーションのコンシューマ鍵}&client_secret={接続アプリケーションのコンシューマの秘密}
```

するとレスポンスの `access_token` にアクセストークンが入っているので取得しておきます。後に記載している `{accessToken}` をこのアクセストークンに置き換えてください。

```json:Response
{
 "access_token": "{アクセストークン}",
 "instance_url": "https://XXXXXXXX.my.salesforce.com",
 "id": "https://login.salesforce.com/id/XXXXXXXX/XXXXXXXX",
 "token_type": "Bearer",
 "issued_at": "1690520603751",
 "signature": "XXXXXXXX"
}
```



### 2. ジョブの作成

まず[ジョブの作成](https://developer.salesforce.com/docs/atlas.ja-jp.api_bulk_v2.meta/api_bulk_v2/create_job.htm)を行います。
下記のような HTTP リクエストで今回は CSV を利用して取引先（Account）を作成（insert）するためのジョブを作成します。

```http:Request
POST /services/data/v57.0/jobs/ingest HTTP/1.1
Host: XXXXXX.my.salesforce.com
Content-Type: application/json
Authorization: Bearer {accessToken}

Request Body:
{
  "operation" : "insert",
  "object" : "Account",
  "contentType" : "CSV"
}
```

下記のようなレスポンスが返ってくるのでのちに jobId として利用するので `id` を取得しておきましょう。 

```json:Response
{
 "id": "XXXXXXXX",
 "operation": "insert",
 "object": "Account",
 "createdById": "XXXXXXXX",
 "createdDate": "2023-07-28T05:09:42.000+0000",
 "systemModstamp": "2023-07-28T05:09:42.000+0000",
 "state": "Open",
 "concurrencyMode": "Parallel",
 "contentType": "CSV",
 "apiVersion": 57.0,
 "contentUrl": "services/data/v57.0/jobs/ingest/XXXXXXXX/batches",
 "lineEnding": "LF",
 "columnDelimiter": "COMMA"
}
```




### 3. CSV アップロード

次に [CSV のアップロード](https://developer.salesforce.com/docs/atlas.ja-jp.api_bulk_v2.meta/api_bulk_v2/upload_job_data.htm)を行います。
下記のリクエストで CSV をアップロードしています。 `Content-Type` が `text/csv` になっていることがポイントです。リクエストが成功すると HTTP ステータスとして `201 Created` が返却されます。

```http:Request
PUT /services/data/v57.0/jobs/ingest/{jobId}/batches HTTP/1.1
Host: XXXXXX.my.salesforce.com
Content-Type: text/csv
Authorization: Bearer {accessToken}

Request Body:
Name,Description,NumberOfEmployees
TestAccount1,Description of TestAccount1,30
TestAccount2,Another description,40
TestAccount3,Yet another description,50
```



### 4. 処理の開始

CSV のアップロードが終了したら、[処理を開始](https://developer.salesforce.com/docs/atlas.ja-jp.api_bulk_v2.meta/api_bulk_v2/close_job.htm)します。
下記のように `UploadComplete` というステートを送信するとジョブが終了し、キューの送信され、処理が開始されます。

```http:Request
PATCH /services/data/v57.0/jobs/ingest/{jobId} HTTP/1.1
Host: XXXXXX.my.salesforce.com
Content-Type: application/json
Authorization: Bearer {accessToken}

Request Body:
{
    "state": "UploadComplete"
}
```

リクエストが成功すると下記のようにジョブの状態が返却されます。

```json:Response
{
 "id": "XXXXXXXX",
 "operation": "insert",
 "object": "Account",
 "createdById": "XXXXXXXX",
 "createdDate": "2023-07-28T05:09:42.000+0000",
 "systemModstamp": "2023-07-28T05:09:42.000+0000",
 "state": "UploadComplete",
 "concurrencyMode": "Parallel",
 "contentType": "CSV",
 "apiVersion": 57.0
}
```




### 5. ジョブの状況の取得

[処理の状況](https://developer.salesforce.com/docs/atlas.ja-jp.api_bulk_v2.meta/api_bulk_v2/get_job_info.htm)を確認します。
下記のリクエストでジョブの状況を確認します。 

```http:Request
GET /services/data/v57.0/jobs/ingest/{jobId} HTTP/1.1
Host: XXXXXX.my.salesforce.com
Authorization: Bearer {accessToken}
```

下記のようなレスポンスが返却されますが、`state` が `JobComplete` になっていたら処理が完了しています。その他の `state` の値については[要求の処理方法](https://developer.salesforce.com/docs/atlas.ja-jp.api_bulk_v2.meta/api_bulk_v2/how_requests_are_processed.htm)をご覧ください。

```json:Response
{
 "id": "XXXXXXXX",
 "operation": "insert",
 "object": "Account",
 "createdById": "XXXXXXXX",
 "createdDate": "2023-07-28T05:09:42.000+0000",
 "systemModstamp": "2023-07-28T05:09:53.000+0000",
 "state": "JobComplete",
 "concurrencyMode": "Parallel",
 "contentType": "CSV",
 "apiVersion": 57.0,
 "jobType": "V2Ingest",
 "lineEnding": "LF",
 "columnDelimiter": "COMMA",
 "numberRecordsProcessed": 3,
 "numberRecordsFailed": 0,
 "retries": 0,
 "totalProcessingTime": 482,
 "apiActiveProcessingTime": 391,
 "apexProcessingTime": 54
}
```





### 6. 処理に成功したレコードの取得

最後に[正常に処理されたレコードの CSV を取得](https://developer.salesforce.com/docs/atlas.ja-jp.api_bulk_v2.meta/api_bulk_v2/get_job_successful_results.htm)します。
下記のリクエストで処理に成功したレコードの CSV を確認します。 他に[失敗したレコード](https://developer.salesforce.com/docs/atlas.ja-jp.api_bulk_v2.meta/api_bulk_v2/get_job_failed_results.htm)や[未処理のレコード](https://developer.salesforce.com/docs/atlas.ja-jp.api_bulk_v2.meta/api_bulk_v2/get_job_unprocessed_results.htm)の取得も可能です。

```http:Request
GET /services/data/v57.0/jobs/ingest/{jobId}/successfulResults HTTP/1.1
Host: XXXXXX.my.salesforce.com
Accept: text/csv
Authorization: Bearer {accessToken}
```

下記のような CSV で返却されます。

```text:result.csv
"sf__Id","sf__Created",Name,Description,NumberOfEmployees
"XXXXXXXX","true","TestAccount1","Description of TestAccount1","30"
"XXXXXXXX","true","TestAccount2","Another description","40"
"XXXXXXXX","true","TestAccount3","Yet another description","50"
```





## オープンソースライブラリの Bulk API 状況

今回は純粋な HTTP API を利用しましたが、下記のようなライブラリでも Bulk API の利用は可能ですが一部 Bulk 2.０には対応してないようです。

* Java
    * [wsc](https://github.com/forcedotcom/wsc) (Bulk API 2.0 に対応してなさそう)
    * [salesforce-bulkv2-java](https://github.com/endolabs/salesforce-bulkv2-java)
* Node.js
    * [jsforce](https://github.com/jsforce/jsforce) ([Bulk API 2.0 は beta で出ているらしい](https://zenn.dev/yktakaha4/articles/salesforce_bulk2_with_jsforce))
* Python
    * [simple-salesforce](https://github.com/simple-salesforce/simple-salesforce) （[Bulk 2.0 の作成中の形跡](https://github.com/simple-salesforce/simple-salesforce/tree/feature/bulk_api_2.0)はあるが現状未対応のよう）
    * [salesforce-bulk-python](https://github.com/mrpowerus/salesforce-bulk-python)
* Golang
    * [simpleforce](https://github.com/simpleforce/simpleforce) (Bulk API には対応していない)
    * [go-sfdc](https://github.com/g8rswimmer/go-sfdc)