
こんにちは、香田です。
今回はCloud Data FusionのBatch パイプラインでSalesforceからBigQueryにロードする方法を紹介していきます。
ロード元のSalesforceのオブジェクトは取引先、取引先責任者を利用していきます。
Salesforce 認証情報の発行
はじめにSalesforceへAPIアクセスできるように認証情報を発行します。
認証情報としてConsumer Key
、Consumer Secret
が必要となる為、以前投稿した下記を参考に認証情報を発行しておいてください。
SalesforceでConsumer Key、Consumer Secretの発行方法について
Cloud Data Fusion Batch パイプラインの作成
次にCloud Data FusionにてBatch パイプラインを作成していきます。
Salesforceのプラグインを利用する為、[HUB]より[Salesforce Plugins]を選択しDeployします。

Deployが成功したら[Create a pipeline]よりパイプラインを作成していきます。

[Source]よりSalesforceを選択し、[Properties]を設定していきます。
[Reference Name]、[Username]、[Password]、[Consumer Key]、[Consumer Secret]を適宜入力します。
Salesforceの取引先
情報を取得するため、[SObject Name]にAccount
と入力し、[GET SCHEMA]クリックし[Output Schema]に出力されることを確認します。[Validate]を実行し問題なければ設定を閉じます。

次に[Sink]よりBigQueryを選択し、[Properties]を設定していきます。

[Dataset]にsalesforce
、[Table]にaccount
を入力し、[Temporary Bucket Name]を適宜入力します。
実行時にTableが再作成されるように[Operation]を[Insert]にして[Truncate Table]を[True]に設定します。
[Validate]を実行し問題なければ設定を閉じます。

パイプラインをプレビューモードで実行
デプロイする前にプレビューモードで確認します。
[Preview]より[Run]をクリックします。成功すると下記のようにSink BigQueryの[Preview Data]が確認できるはずです。

パイプラインのデプロイ
プレビューモードを終了し、パイプラインの名前を入力し[Deploy]をクリックします。[Run]をクリックししばらくすると[Status]が[Succeeded]になっているはずです。

※下記のようなエラーログで失敗してしまう場合は、サービス アカウント ユーザーの権限の付与を参考に、サービスアカウントの権限を調整してみてください。
PROVISION task failed in REQUESTING_CREATE state for program run [pipeline-name] due to Dataproc operation failure: INVALID_ARGUMENT: User not authorized to act as service account '[service-account-name]'
BigQueryのデータ確認
BigQueryのコンソールへ移動し、テーブルが作成されていることを確認します。

問題なければSalesforceの取引先責任者
の情報を取得するため、作成したBatchパイプラインを複製します。
Source Salesforceの[SObject Name]にContact
を入力し、Sink BigQueryの[Table]にcontact
を入力します。
作成したBatch パイプラインを実行しBigQuery側に取引先責任者
のデータがロードされていれば成功です。
下記のようなクエリを実行することで、Salesforceの取引先責任者と取引先を結合して表示することが可能です。
SELECT
c.Name AS ContactName,
a.Name AS AccountName,
c.Title AS Titlte,
c.Phone AS Phone,
c.Email AS Email
FROM
salesforce.contact AS c
JOIN
salesforce.account AS a
ON
c.AccountId = a.Id
ORDER BY
ContactName

さいごに
Cloud Data FusionのBatch パイプラインでSalesforceからBigQueryにロードする方法いかがでしたでしょうか。
Cloud Data FusionでSalesforceプラグインを利用する際、本記事が参考になれば幸いでございます。
最後までご覧頂きありがとうございます。