Embulkを使ってSalesforceにデータを連携する

はじめに

Embulkを使ってSalesforceにデータ連携をします。
使用するプラグインはembulk-output-salesforceです。

今回書かないこと

• Embulkのセットアップ
• Embulkとは何か？の説明

環境がない方は、公式を参考にインストールしてみてください

どんな定義ファイルになるの？

in側で、Salesforceの項目のAPI参照名を
out側で対象のオブジェクト・接続情報・アクション(insert, upsert)等を設定する

in:
  type: file
  path_prefix: /path/to/salesforce_
  parser:
    charset: UTF-8
    newline: CRLF
    type: csv
    delimiter: ','
    quote: '"'
    escape: ''
    skip_header_lines: 1
    comment_line_marker: null
    allow_extra_columns: true
    allow_optional_columns: false
    columns:
    - {name: Name, type: string}
    - {name: Date__c, type: timestamp, format: '%Y-%m-%d'}
    - {name: DateTime__c, type: timestamp, format: '%Y-%m-%d %H:%M:%S%Z'}
    - {name: Number__c, type: double}
    - {name: Reference__c, type: string}
out:
  type: salesforce
  username: [e-mail]
  password: fuga
  sobject: Account
  action: insert

今回やること

外部システムのデータを外部IDをキーにして連携させることを想定し以下のパターンをします。

• Account オブジェクトへのupsert
• Contact オブジェクトへのupsert & Accountへの親子関係の生成

事前準備

データ

idはわかりやすいようにテキストにしておきます。

accountsテーブルのIDが、contactsテーブルの中にあってリレーションが結べる状態

accounts.csv

id, Name
account_1, 取引先テスト_1
account_2, 取引先テスト_2
account_3, 取引先テスト_3

contacts.csv

id, account_id LastName, FirstName
contact_1, account_1, テスト姓_1, 名
contact_2, account_1, テスト姓_2, 名
contact_3, account_2, テスト姓_3, 名

外部ID項目

ExternalId__cというAPI参照名でデータ型テキストとして外部ID項目を
AccountとContactに作成しました。

セキュリティトークンの取得

セキュリティトークンをまだ持ってない場合は、
個人設定>私のセキュリティトークンのリセットで取得してください。

設定に存在しない場合は、公式のヘルプを参照してください

実装

ディレクトリ構造はこんな感じ

.
├── accounts.yml.liquid
├── contacts.yml.liquid
└── csv
    ├── log
    └── source
        └── accounts.csv

embulk-output-salesforceのインストール

$ embulk gem install embulk-output-salesforce

embulk-output-salesforceの補足

upsertする場合の構成の定義は基本以下の通りで

out:
  type: salesforce
  username: {{env.SALESFORCE_USERNAME}}
  password: {{env.SALESFORCE_PASSWORD}}{{env.SALESFORCE_TOKEN}}
  sobject: Account
  action: upsert
  upsert_key: ExternalId__c
  result_dir: ./csv/log/

{{env.SALESFORCE_PASSWORD}}{{env.SALESFORCE_TOKEN}}には、パスワードの末尾にセキュリティトークンを組み合わせたものに置換してください。

パスワード： 1234
セキュリティトークン：abcd
とすると
例) password: 1234abcd

項目	説明
username	対象salesforce組織のユーザ名
password	対象salesforce組織のユーザパスワード + セキュリティトークンを組み合わせたもの
sobject	対象オブジェクト
action	insert, upsert, update, delete
upsert_key	upsertする場合の外部ID項目
result_dir	実行結果のCSV保存先、成功と失敗でそれぞれ出力される

Accountオブジェクトにupsertする

定義

accounts.yml.liquid

in:
  type: file
  path_prefix: ./csv/source/accounts.csv
  parser:
    charset: UTF-8
    newline: CRLF
    type: csv
    delimiter: ','
    quote: '"'
    escape: ''
    skip_header_lines: 1
    comment_line_marker: null
    allow_extra_columns: true
    allow_optional_columns: false
    columns:
    - {name: ExternalId__c, type: string}
    - {name: Name, type: string}
out:
  type: salesforce
  username: {{env.SALESFORCE_USERNAME}}
  password: {{env.SALESFORCE_PASSWORD}}{{env.SALESFORCE_TOKEN}}
  sobject: Account
  action: upsert
  upsert_key: ExternalId__c
  result_dir: ./csv/log/

実行

$ embulk run accounts.yml.liquid
Next config diff: {"in":{"last_path":"./csv/source/accounts.csv"},"out":{}}

実行完了しました。

result_dirの確認

result_dirの出力を確認します。

エラーなくデータが入ったみたいです。

error_20221218075936943.csv

ExternalId__c,Name,Error

success_20221218075936943.csv

Id,ExternalId__c,Name
0012w00001AyDD5AAN,account_1, 取引先テスト_1
0012w00001AyDD6AAN,account_2, 取引先テスト_2
0012w00001AyDD7AAN,account_3, 取引先テスト_3

Salesforce上での確認

Salesforce上でも確認できました。

Contactオブジェクトにupsertする

ContactオブジェクトとAccountオブジェクトを紐づける際は特殊なので、再度CSVを確認してみる

contacts.csv

id, account_id LastName, FirstName
contact_1, account_1, テスト姓_1, 名
contact_2, account_1, テスト姓_2, 名
contact_3, account_2, テスト姓_3, 名

account_idが参照関係を結ぶためのキーとなるので
SOQLを書く感覚で以下のようになる

ExternalId__cがaccount_1のAccountとContactを結びつけることになる。

    columns:
    - {name: ExternalId__c, type: string}
    - {name: Account.ExternalId__c, type: string}
    - {name: LastName, type: string}
    - {name: FirstName, type: string}

定義

contacts.yml.liquid

in:
  type: file
  path_prefix: ./csv/source/contacts.csv
  parser:
    charset: UTF-8
    newline: CRLF
    type: csv
    delimiter: ','
    quote: '"'
    escape: ''
    skip_header_lines: 1
    comment_line_marker: null
    allow_extra_columns: true
    allow_optional_columns: false
    columns:
    - {name: ExternalId__c, type: string}
    - {name: Account.ExternalId__c, type: string}
    - {name: LastName, type: string}
    - {name: FirstName, type: string}
out:
  type: salesforce
  username: {{env.SALESFORCE_USERNAME}}
  password: {{env.SALESFORCE_PASSWORD}}{{env.SALESFORCE_TOKEN}}
  sobject: Contact
  action: upsert
  upsert_key: ExternalId__c
  result_dir: ./csv/log/

実行

$ embulk run contacts.yml.liquid
Next config diff: {"in":{"last_path":"./csv/source/contacts.csv"},"out":{}}

実行完了しました。

result_dirの確認

result_dirの出力を確認します。

エラーなくデータが入ったみたいです。

error_20221218083022091.csv

ExternalId__c,Account.ExternalId__c,LastName,FirstName,Error

success_20221218083022091.csv

Id,ExternalId__c,Account.ExternalId__c,LastName,FirstName
0032w00000qG2RaAAK,contact_1,, テスト姓_1, 名
0032w00000qG2RbAAK,contact_2,, テスト姓_2, 名
0032w00000qG2TbAAK,contact_3,, テスト姓_3, 名

Salesforce上での確認

Salesforce上でも確認できました。親子関係も再現できてます！

さいごに

Salesforceに参照関係を再現した状態でupsertできましたね。
実際に運用するのであれば、AccountとContactの実行順序を定義しやすく管理しやすくできるように、
argoやdigdagと組み合わせるといいですね。
あと、バッチ的な使い道ではなく短い感覚（15分とか10分とか）で実行する場合は個人的には
エラーのCSVを処理して通知する仕組みがあったほうがいいと思います。

参考