Azure Data Factory にて Salesforce をシンクとした場合のデータ連携パターン

## 概要

Azure Data Factory にて Salesforce をシンクとした場合のデータ連携パターンを共有します。

## 基本的な方針

### データ連携パターン

Azure Data Factory にて Salesforce をシンクとする場合には、パイプラインの実装方法とデータの操作方法の組み合わせで構築する必要があります。

パイプラインの実装方法としては下記の方法があります。

- Salesforce コネクタによるコピーアクティビティ
- Salesforce REST API による Web アクティビティ
- Salesforce Bulk API 2.0 による Web アクティビティ

データの操作方法としては下記があります。HARD DELETE については、デフォルトで実施することができないため、データ連携パターンの対象外とします。

- INSERT
- UPSERT （外部 ID による UPDATE or INSERT）
- UPDATE （Salesforce ID による UPDATE or INSERT）
- DELETE（論理削除）
- HARD DELETE（物理削除）

Salesforce コネクタによるコピーアクティビティでは、 INSERT と UPSERT のみをサポートされています。

> 操作の書き込み動作。
> 使用可能な値: Insert および Upsert。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/b4f3db79-d61e-0578-daa1-aa2fa9e98684.png)

引用元：[Salesforce との間でデータをコピーする - Azure Data Factory & Azure Synapse | Microsoft Learn](https://learn.microsoft.com/ja-jp/azure/data-factory/connector-salesforce?tabs=data-factory#salesforce-as-a-sink-type)

以上の情報を整理すると、Azure Data Factory にて Salesforce をシンクとする場合には下記の方法があります。

1. コピーアクティビティ（Salesforce コネクタ）による INSERT
2. コピーアクティビティ（Salesforce コネクタ）による UPSERT
3. Web アクティビティ（Salesforce REST API）による INSERT
4. Web アクティビティ（Salesforce REST API）による UPSERT
5. Web アクティビティ（Salesforce REST API）による UPDATE
6. Web アクティビティ（Salesforce REST API）による DELETE
7. Web アクティビティ（Salesforce Bulk API 2.0）による INSERT
8. Web アクティビティ（Salesforce Bulk API 2.0）による UPSERT
9. Web アクティビティ（Salesforce Bulk API 2.0）による UPDATE
10. Web アクティビティ（Salesforce Bulk API 2.0）による DELETE

### Salesforce REST API による Web アクティビティの実装方針

ソースから参照アクティビティにて json 形式の配列データを取得し、 ForEach アクティビティにて単一の json 形式をボディに渡して処理を実行します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/5a54d95f-a872-04e6-4b4e-f9e8877531fb.png)

参照アクティビティの出力サイズに上限があることに注意が必要です。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/205e9ecd-8f41-4746-58e3-74e705064c97.png)

引用元：[ルックアップ アクティビティ - Azure Data Factory & Azure Synapse | Microsoft Learn](https://learn.microsoft.com/ja-jp/azure/data-factory/control-flow-lookup-activity#supported-capabilities)

### Salesforce Bulk API 2.0 による Web アクティビティの実装方針

Salesforce Bulk API 2.0 を利用する際には、 CSV 形式データを反映する必要があります。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/704305ce-c8ff-9a4e-581e-c7f6a5be9a38.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/92b9c3d1-3bd3-e467-c1df-7bc4c48a6658.png)

引用元：[ステップ 5: 一括更新/挿入 | Bulk API 2.0 および Bulk API 開発者ガイド | Salesforce Developers](https://developer.salesforce.com/docs/atlas.ja-jp.api_asynch.meta/api_asynch/walkthrough_upsert.htm)

CSV 形式のデータを指定するために、ストレージ上の CSV ファイルの内容を REST API で取得した値を Salesforce REST API 実行時の Body に指定します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/47b20e78-7cd3-8f69-bd21-6b8ef851a4a4.png)

Web アクティビティの出力サイズに上限があることに注意が必要です。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/78307433-60e0-4c6e-f42b-ec97912ba13f.png)

引用元：[Web アクティビティ - Azure Data Factory & Azure Synapse | Microsoft Learn](https://learn.microsoft.com/ja-jp/azure/data-factory/control-flow-web-activity)

### Salesforce Bulk API 2.0 の実行方法

Salesforce Bulk API 2.0 により Salseforce へ書き込みを実施する場合には、下記の API を順に実行する必要があります。

1. [ジョブの作成 | Bulk API 2.0 および Bulk API 開発者ガイド | Salesforce Developers](https://developer.salesforce.com/docs/atlas.ja-jp.api_asynch.meta/api_asynch/create_job.htm)
2. [ジョブデータのアップロード | Bulk API 2.0 および Bulk API 開発者ガイド | Salesforce Developers](https://developer.salesforce.com/docs/atlas.ja-jp.api_asynch.meta/api_asynch/upload_job_data.htm)
3. [アップロードの完了 | Bulk API 2.0 および Bulk API 開発者ガイド | Salesforce Developers](https://developer.salesforce.com/docs/atlas.ja-jp.api_asynch.meta/api_asynch/close_job.htm)
4. [ジョブ情報の取得 | Bulk API 2.0 および Bulk API 開発者ガイド | Salesforce Developers](https://developer.salesforce.com/docs/atlas.ja-jp.api_asynch.meta/api_asynch/get_job_info.htm)

ジョブの実行状況をジョブ情報の取得の`state`の値で確認できます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/aee0dcb4-ba58-7414-887e-e6980a7dd965.png)

引用元：[ジョブ情報の取得 | Bulk API 2.0 および Bulk API 開発者ガイド | Salesforce Developers](https://developer.salesforce.com/docs/atlas.ja-jp.api_asynch.meta/api_asynch/get_job_info.htm)

失敗を厳密に管理したい場合には、`stae`が`Failed`であることを監視するだけでなく、`numberRecordsFailed`の値も監視する必要があります。レコードの書き込みが失敗したとしても`state`が`Job Complete`になるためです。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/bd132bc4-cf1f-2fbd-401e-5d9d28072e9b.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/2a86b261-9121-e46e-4da4-a4e174340fe1.png)

引用元：[ジョブ情報の取得 | Bulk API 2.0 および Bulk API 開発者ガイド | Salesforce Developers](https://developer.salesforce.com/docs/atlas.ja-jp.api_asynch.meta/api_asynch/get_job_info.htm)


### データ連携パターン

#### 1. コピーアクティビティ（Salesforce コネクタ）による INSERT

#### 利用する Salesforce の API 

Salesforce コネクタを利用する場合には Salesforce Bulk API 2.0 が利用されます。

> Salesforce Bulk API 2.0 を使用して、データのクエリと取り込みを行います。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/708f961b-7195-1224-f780-b81e4c7eae6d.png)

引用元：[Salesforce との間でデータをコピーする - Azure Data Factory & Azure Synapse | Microsoft Learn](https://learn.microsoft.com/ja-jp/azure/data-factory/connector-salesforce?tabs=data-factory#salesforce-bulk-api-20-limits)

#### パイプライン

##### パイプラインの全体像

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/ee50216d-27cb-2cb9-8c82-bd74e0ca4c7c.png)

#### 実行確認

##### カスタムオブジェクトを準備

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/4686efa1-13da-1268-2e86-1690e5ae4712.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/9756d397-a0e3-bb10-dd8c-c4b96015f8d3.png)

#### CSV ファイルを配置

```csv
"Name","ExternalId__c"
"A Company","1"
"B Compan","2"
"C Company","3"
```

##### パイプラインを実行

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/b8bd4557-f7a4-82bb-627c-ed365811b837.png)

##### データの確認

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/7e704e34-79dc-583b-e1f2-3379d60fc044.png)

#### 2. コピーアクティビティ（Salesforce コネクタ）による UPSERT

#### 利用する Salesforce の API 

Salesforce コネクタを利用する場合には Salesforce Bulk API 2.0 が利用されます。

> Salesforce Bulk API 2.0 を使用して、データのクエリと取り込みを行います。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/708f961b-7195-1224-f780-b81e4c7eae6d.png)

引用元：[Salesforce との間でデータをコピーする - Azure Data Factory & Azure Synapse | Microsoft Learn](https://learn.microsoft.com/ja-jp/azure/data-factory/connector-salesforce?tabs=data-factory#salesforce-bulk-api-20-limits)

#### パイプライン

##### パイプラインの全体像

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/89ea4324-b098-5c28-e99c-0c66f3550529.png)

#### コピーアクティビティのシンクの設定

`書き込む動作`を`Upsert`に、`外部ID フィールド`を`ADFTable01__c`に設定

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/eaff6df0-7c67-7474-97c2-a31efe75980e.png)

#### 実行確認

##### カスタムオブジェクトを準備

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/55439e37-7a80-15ff-f5f6-0559e1f5ebd1.png)

#### CSV ファイルを配置

```csv
"ExternalId__c","Name"
"3","CCC Company"
"4","D Company"
```

##### パイプラインを実行

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/ac19b41a-df5b-4f9c-c7b4-d2a68a1362ce.png)

##### データの確認

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/f1322640-d46b-e88a-bae7-27bf53ec8e21.png)

#### 3. Web アクティビティ（Salesforce REST API）による INSERT

#### 利用する Salesforce の API

下記の API を利用します。

- [sObject Basic Information を使用したレコードの作成 | REST API 開発者ガイド | Salesforce Developers](https://developer.salesforce.com/docs/atlas.ja-jp.api_rest.meta/api_rest/resources_sobject_basic_info_post.htm)

#### パイプライン

##### パイプラインの全体像

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/86794a7c-7ea2-5546-b99c-1d589b447b13.png)

#### 実行確認

##### カスタムオブジェクトを準備

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/90a0091a-f1f7-353e-efd0-0380605f8f86.png)

#### CSV ファイルを配置

```csv
"Name","ExternalId__c"
"A Company","1"
"B Compan","2"
"C Company","3"
```

##### パイプラインを実行

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/37cef1b1-0d26-693a-6c6d-51818636c5bf.png)

##### データの確認

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c2e0ad49-e644-d773-e52a-6f90c30ae8b5.png)

#### 4. Web アクティビティ（Salesforce REST API）による UPSERT

#### 利用する Salesforce の API 

下記の API を利用します。

- [sObject Rows by External ID を使用したレコードの更新/挿入 | REST API 開発者ガイド | Salesforce Developers](https://developer.salesforce.com/docs/atlas.ja-jp.248.0.api_rest.meta/api_rest/resources_sobject_upsert_patch.htm)

#### パイプライン

##### パイプラインの全体像

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/d4f4bf82-43e5-1a08-f92e-9345547b7bd9.png)

#### json のデータから外部 ID 項目の除外

Body に渡す json データにて外部 ID 項目があるとエラーとなるため、 下記のような式により外部 ID のキーと値を除外します。

```text
@replace(
    string(item()),
    concat(
        '"',
        pipeline().parameters.SalesforceExternalIdFieldName,
        '":"',
        item()[pipeline().parameters.SalesforceExternalIdFieldName],
        '",'
    ),
    ''
)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/d131302c-0f2c-b5b7-eacc-a97bea6f41a0.png)

#### 実行確認

##### カスタムオブジェクトを準備

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/0045559e-f24a-eef0-4df0-0d81fdaa6a7d.png)

#### CSV ファイルを配置

```csv
"ExternalId__c","Name"
"3","CCC Company"
"4","D Company"
```

##### パイプラインを実行

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/bcef54e4-c612-d000-6ce3-d62b95954d88.png)

##### データの確認

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c65c0ff0-0d7f-5d88-4ec8-32e1e74bd7d7.png)

#### 5. Web アクティビティ（Salesforce REST API）による UPDATE

#### 利用する Salesforce の API 

下記の API を利用します。

- [sObject Rows を使用したレコードの更新 | REST API 開発者ガイド | Salesforce Developers](https://developer.salesforce.com/docs/atlas.ja-jp.api_rest.meta/api_rest/resources_sobject_retrieve_patch.htm)

#### パイプライン

##### パイプラインの全体像

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/29ac3aec-6b5a-55eb-6290-4ef5094f13a8.png)

#### json のデータから Salesforce ID 項目の除外

Body に渡す json データにて Salesforce ID 項目があるとエラーとなるため、 下記のような式により Id のキーと値を除外します。

```text
@replace(
    string(item()),
    concat(
        '"Id":"',
        item().id,
        '",'
    ),
    ''
)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/adefce9a-0393-78d9-89c4-c250b483c415.png)

#### 実行確認

##### カスタムオブジェクトを準備

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/1fc08411-712b-d560-f98b-eaeade258846.png)

#### CSV ファイルを配置

Id 列は Salesforce ID 項目であるため、既存の ID を指定する必要があります。

```csv
"Id","Name"
"a02IT00001x1f9u","BBB Company"
```

##### パイプラインを実行

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/4e2b316e-9a3b-892c-e6dc-3f20bf41438c.png)

##### データの確認

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/78b90cde-c95f-816f-f927-88b961613aa3.png)

#### 6. Web アクティビティ（Salesforce REST API）による DELETE

#### 利用する Salesforce の API 

下記の API を利用します。

- [sObject Rows を使用したレコードの削除 | REST API 開発者ガイド | Salesforce Developers](https://developer.salesforce.com/docs/atlas.ja-jp.api_rest.meta/api_rest/resources_sobject_retrieve_delete.htm)

#### パイプライン

##### パイプラインの全体像

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/8452209d-d8d8-bd6d-f6b6-f3c4da2d8c6a.png)

#### 実行確認

##### カスタムオブジェクトを準備

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/b6d40991-e70b-0a0b-51cd-871c03f5cf85.png)

#### CSV ファイルを配置

Id 列は Salesforce ID 項目であるため、既存の ID を指定する必要があります。

```csv
"Id"
"a02IT00001x1f9p"
"a02IT00001x1f9u"
```

##### パイプラインを実行

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/dd890186-f563-2049-915f-4f99a7d8615a.png)

##### データの確認

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/55260d23-b5c0-ba41-9c4e-49089cdd4db7.png)

#### 7. Web アクティビティ（Salesforce Bulk API 2.0）による INSERT

#### 利用する Salesforce の API 

ジョブの作成 API にて`operation`に`insert`を指定します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/f289dd1e-b146-e353-f7fd-466027a7477d.png)

引用元：[ジョブの作成 | Bulk API 2.0 および Bulk API 開発者ガイド | Salesforce Developers](https://developer.salesforce.com/docs/atlas.ja-jp.api_asynch.meta/api_asynch/create_job.htm)

#### パイプライン

##### パイプラインの全体像

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/1cb01321-6559-1a3e-81be-3d0ae5ee5647.png)

#### 実行確認

##### カスタムオブジェクトを準備

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c0559f9e-0ba9-0f13-38a6-3378e712c71c.png)

#### CSV ファイルを配置

```csv
"Name","ExternalId__c"
"A Company","1"
"B Compan","2"
"C Company","3"
```

##### パイプラインを実行

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/939ce690-4baf-5aa0-dc2d-0e092db444ae.png)

##### データの確認

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/ebae5240-104f-39d8-742a-2763ce3ea8a0.png)

#### 8. Web アクティビティ（Salesforce Bulk API 2.0）による UPSERT

#### 利用する Salesforce の API 

ジョブの作成 API にて`operation`に`upsert`を指定します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/f289dd1e-b146-e353-f7fd-466027a7477d.png)

引用元：[ジョブの作成 | Bulk API 2.0 および Bulk API 開発者ガイド | Salesforce Developers](https://developer.salesforce.com/docs/atlas.ja-jp.api_asynch.meta/api_asynch/create_job.htm)

#### パイプライン

##### パイプラインの全体像

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/d4d42cf3-e964-6c17-77d3-c3739f1d4b06.png)

#### 実行確認

##### カスタムオブジェクトを準備

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/5551b075-8765-698f-ee8c-153386d023ec.png)

#### CSV ファイルを配置

```csv
"ExternalId__c","Name"
"3","CCC Company"
"4","D Company"
```

##### パイプラインを実行

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/773986ef-8d10-7d7c-4389-4a5253ca4ef7.png)

##### データの確認

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/8c02210a-d4a4-7bad-552a-54b95b041081.png)

#### 9. Web アクティビティ（Salesforce Bulk API 2.0）による UPDATE

#### 利用する Salesforce の API 

ジョブの作成 API にて`operation`に`update`を指定します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/f289dd1e-b146-e353-f7fd-466027a7477d.png)

引用元：[ジョブの作成 | Bulk API 2.0 および Bulk API 開発者ガイド | Salesforce Developers](https://developer.salesforce.com/docs/atlas.ja-jp.api_asynch.meta/api_asynch/create_job.htm)

#### パイプライン

##### パイプラインの全体像

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/1dd6053e-073f-43d4-36cf-2d021cb6d346.png)

#### 実行確認

##### カスタムオブジェクトを準備

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/b94e296f-97a6-7f24-5a06-b1591fa03921.png)

#### CSV ファイルを配置

```csv
"Id","Name"
"a02IT00001x1fA5","BBB Company"
```

##### パイプラインを実行

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/5b92de30-0a41-ee3a-b540-1ec35e592dbe.png)


##### データの確認

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/59264502-6370-c766-6795-231591ac5f60.png)

#### 10. Web アクティビティ（Salesforce Bulk API 2.0）による DELETE

#### 利用する Salesforce の API 

ジョブの作成 API にて`operation`に`delete`を指定します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/f289dd1e-b146-e353-f7fd-466027a7477d.png)

引用元：[ジョブの作成 | Bulk API 2.0 および Bulk API 開発者ガイド | Salesforce Developers](https://developer.salesforce.com/docs/atlas.ja-jp.api_asynch.meta/api_asynch/create_job.htm)

#### パイプライン

##### パイプラインの全体像

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/b3571474-c51c-4cfc-70d3-ab740b80c8b6.png)


#### 実行確認

##### カスタムオブジェクトを準備

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/39d85445-2ad1-dc71-9e5f-9ec41be61f87.png)

#### CSV ファイルを配置

```csv
"Id"
"a02IT00001x1fA6"
"a02IT00001x1fA5"
```

##### パイプラインを実行

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/cb568ac9-9b13-808c-74cb-74df58b99c9d.png)

##### データの確認

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c2411a2c-4517-0867-05fd-9febe5eaf77e.png)