Cloud Data FusionでSalesforceのデータをBigQueryに取り込んでみる

はじめに

データアナリティクス事業本部のkobayashiです。

クラスメソッド BigQuery Advent Calendar 2020 の14日目のエントリになります。

最近、データパイプラインGCPのフルマネージドサービスであるCloud Data Fusionを色々試しています。他のETL,ELTを行えるデータパイプラインツールも操作していますが、Cloud Data Fusionはなかなか面白く、調査を楽しんでいます。

• Cloud Data Fusion Google Cloud

今回はCloud Data FusionのHUBでPluginsを扱ってみたいと思います。

Cloud Data Fusion HUB

Cloud Data FusionのHUBは再利用可能なコンポーネント（各種データベースのドライバ、AWS、Azureなどのクラウドサービスへのコネクタ）やサンプルパイプラインなどが利用できます。

今回はその中のSalesforce Pluginsを使ってみます。

事前準備

事前準備としてSalesforceの設定を行う必要があります。

• アプリケーションマネージャーから新規接続アプリケーションの追加
• 接続アプリケーション管理でIP制限を変更

新規接続アプリケーションの追加

手順1）設定 > アプリケーション > アプリケーションマネージャで新規接続アプリケーションを押下する。

手順2）新規接続アプリケーションの設定を行い保存を押下する。

• 基本情報
  • 接続アプリケーション名 : GCP Cloud Data Fusionを入力
  • API参照名 : GCP Cloud Data Fusionを入力
  • 取引先責任者メール : ユーザーのメールアドレスを入力
• API(OAuth設定の有効化)
  • OAuth設定の有効化 : チェックする
  • デバイスフローで有効化 : チェックする
  • 選択したOAuth範囲 : Web経由のデータへのアクセスを提供（web)を選択

下図の様にGCP Cloud Data Fusionのアプリケーションが登録されます。コンシューマキーとコンシューマシークレットを確認するには右側のプルダウンメニューを開き参照を選択します。

すると次の画面で図の様にコンシューマキーとコンシューマシークレットが確認できるのでこれを後の設定で使用します。

次に接続元であるCloud Data Fusionから接続できるように接続制限を変更します。

接続アプリケーション管理でIP制限を変更

手順1）設定 > アプリケーション > 接続アピりケーション > 接続アプリケーションを管理するでGCP Cloud Data Fusionの左の編集を押下する。

手順2）IP制限の緩和で有効化されたデバイスのIP制限を緩和を選択し保存をします。

これで事前設定は完了ですのでパイプラインを作成します。

パイプラインの作成

HUBからSalesforceプラグインを追加

SalesforceをデータソースとするためにはじめにHUBからSalesforceのプラグインをデプロイします。

手順1）Studioからパイプラインの作成を開始し、右上のHUBを押下する。

手順2）検索フォームにsalesforceを入力すると、Salesforceのプラグインが表示されるのでSalesforce Pluginsを押下する。

手順3）モーダルが表示されるのでDeployを押下する。

手順4）再度別のモーダルが表示されるのでそのままFinishを押下する。

これでSalesforceのノードが追加されるので設定を行っていきます。

パイプラインの設定

手順1）SourceにSalesforceノードが表示されるので押下してノードを追加する。

手順2）SalesforceノードのPropertiesを押下し、Salesforceの設定を行う。値を入力後、GET SCHEMAを押下し右側に取得したいカラムが表示される事を確認する。

• Username : Salesforceのユーザー名を入力
• Password : 上記ユーザーのパスワードを入力
• Comsumer Key : Salesforceのアプリケーションマネージャで作成したコンシューマ鍵を入力
• Comsumer Secret : Salesforceのアプリケーションマネージャで作成したコンシューマの秘密を入力
• SObject Query
  • SObject Name : Leadを入力

今回はSalesforceのリードを取り込んでみたいのでSObject QueryのSObject NameにLeadを入力しましたが、SOQL Queryも使えるので取得したいデータに応じてSOQLもしくはSObject Queryを使います。

手順3）左のメニューのSink > BigQueryを選択し、Salesforceノードを接続する。

手順4）BigQueryノードのPropertiesを押下し、SinkのBigQuery設定を行う。

• Reference Name : BigQueryを入力
• Dataset : data_set_salesforceを入力
• Table : leadを入力

以上で設定は完了したので、パイプラインの設定を確認します。

パイプラインの確認

手順1）Pipeline Studioの画面上部のPreviewを押下する。

手順2）Preview画面のRunを押下してPreviewを実行する。

手順3）Preview完了後、Preview Dataを押下して内容を確認します。

下記の様にデータが表示されていることが確認できます。

Previewで問題がなければPipeline Studioの画面上部のDeployを押下してパイプラインをデプロイします。

パイプラインの実行

デプロイが終わるとパイプラインを実行することができるようになリます。

手順1）画面上部のRunを押下してパイプラインを実行する。

Successが表示されデータパイプラインの実行が終わったあとに、実行結果をBigQueryで確認すると変換されたデータが登録されていることがわかります。

まとめ

Cloud Data Fusionを使ってSalesforceのデータをBigqueryに取り込むデータパイプラインを構築し実行してみました。このパイプラインもノーコーディングで簡単に構築できました。SourceノードとSinkノードだけの単純なパイプラインでしたがTransformやAnalyticsを使うことでデータの加工も行えます。

最後まで読んで頂いてありがとうございました。

クラスメソッド BigQuery Advent Calendar 2020 15日目は、 ナガマサ さんです。お楽しみに。