SalesforceのデータをAmazon AppFlowでSnowflakeに増分連携してみた
データ事業本部の鈴木です。
SalesforceのデータをAppFlowでSnowflakeに連携したいことがあったので、どのように実現できるか確認しました。
SalesforceからAmazon S3の連携はよく例があるのですが、Snowflakeへは意外と情報が少なかったので、現時点でどのようにできるかをご紹介できればと思います。
もちろんS3に連携して、外部ステージから読み込むのもアリと思います。今回紹介する機能も内部的にはそのようになっていました。
S3への連携は以下が参考になります。
以降にSalesforceのデータをSnowflakeに増分連携するAppFlowの設定と、データ連携した様子をご紹介します。
1. データ連携元・連携先の準備
i. Salesforce環境の準備
Salesforce Developer Edition へのサインアップを行い、Developer Editionの環境を用意しました。
今回は商談オブジェクトのデータを使いました。
Salesforce標準オブジェクトは以下が参考になりました。
ii. Snowflake環境の準備
検証向けに、AppFlow向けのユーザーを作成しました。
CREATE USER appflow_user
PASSWORD = '<ユーザー向けのパスワードを入力する>'
DEFAULT_ROLE = SYSADMIN
DEFAULT_WAREHOUSE = COMPUTE_WH;
今回はあくまで検証なので、SYSADMINを付与しておきました。実際は最小権限のロールを付与することをお勧めします。
その次に、AppFlow向けのデータベース・スキーマを作成しました。また、その下にSalesforceから取得する商談向けのテーブルを作成しました。
実際にシステムを構築する際は、ここでSalesforceから取得するデータを格納するためのテーブルを作成します。
商談オブジェクトのカラム全てを取り込むと、AppFlowのフロー作成画面での設定が動作確認の目的に対して煩雑なため、ここではいくつかに絞っておきました。
CREATE DATABASE APPFLOW_DB;
CREATE SCHEMA APPFLOW_DB.APPFLOW_SCHEMA;
CREATE OR REPLACE TABLE APPFLOW_DB.APPFLOW_SCHEMA.SALESFORCE_OPPORTUNITY (
ID STRING NOT NULL,
NAME STRING,
ACCOUNT_ID STRING,
OWNER_ID STRING,
CREATED_DATE TIMESTAMP_NTZ,
LAST_MODIFIED_DATE TIMESTAMP_NTZ,
IS_DELETED BOOLEAN,
LOADED_AT TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP(),
PRIMARY KEY (ID)
);
また、AppFlowがSalesforceから取得したデータをSnowflakeにロードする際に使用するストレージ統合と外部ステージを作成しました。
この操作はガイドに解説があるのでそちらに譲ります。
S3バケットおよびIAMロールはAppFlowフローを作成するのと同じAWSアカウントで作成しておくこととします。
最終的に、以下の状態にしておきました。
2. AppFlowの準備
i. Salesforce接続の作成
まずはSalesforceへの接続を作成しました。
これは簡単で、接続 > 接続を管理より、Salesforceのコネクタを選択して、接続を作成を押しました。
今回はデータソースがDeveloper Edition環境だったので、OAuth グラントタイプは認証コード、Salesforce 環境はProductionを選びました。
PrivateLinkを有効にすることもできますが、Salesforce側でライセンスが必要なため、今回は無効としました。
接続するを押すと、データソースとの認証向けのポップアップが出るので、接続を作成するアカウントで認証しました。
認証ができればSalesforce接続の作成は完了です。
ii. Snowflake接続の作成
次にSnowflakeへの接続を作成しました。
Salesforceと同じく、接続 > 接続を管理より、Snowflakeのコネクタを選択して、接続を作成を押しました。
接続するSnowflake環境の設定を入力しました。
特に以下はポイントでした。
- ウェアハウス：ユーザーが権限のあるウェアハウスの名前を入力する
- ステージ名：S3外部ステージの完全修飾ステージ名
- バケットの詳細：上記のS3外部ステージを設定したS3のパス
- アカウント名：アカウントロケーター（
SELECT CURRENT_ACCOUNT()で取得できる値） - ユーザー名：AppFlowが利用するSnowflakeユーザー
- パスワード：上記ユーザーのパスワード
私がハマったところで、ユーザーがウェアハウスに権限がない場合、フロー設定時にデータのCOPY先テーブルが表示されない現象が起こりました。ユーザーに権限があるはずなのに表示されない場合はウェアハウス向けの権限もあるか確認してみてください。
ステージ名とバケットの詳細も、同一のS3外部ステージを指しているので注意してください。後で記載するように、フローは実行時にバケットの詳細に設定したパス+フロー名のプレフィクスに連携データを配置します。そのため、バケットの詳細に設定したパスがS3外部ステージで指定した場所に対応していないと、Snowflake側で実行されるCOPYコマンドのCOPY元が異なる場所を向いてしまい、空振りや誤ったデータのロードに繋がる可能性がありました。
また、フロー名がバケットの詳細に設定したパスに既に含まれていると、バケットの詳細に設定したパスだけでオブジェクトのパスが決まるように見えました。バケットの詳細は一度設定すると編集で変えてもフロー実行時の出力先が変わらないようにも見えたため、もし同じ現象が起これば再度接続を作り直してフローに設定すると反映されました。
アカウント名はSELECT CURRENT_ACCOUNT()で取得できるアカウントロケーターを設定する必要がありました。
こちらは必要なパラメータを入力し、接続するを押して無事にエラーなく作成されたら作業完了です。
iii. フローの作成
フロー > フローを作成より、新規にフローを作成しました。
フロー名と説明を入力し、ほかはデフォルトとしました。
フローを設定で、作成した接続を指定しました。
送信元ではSalesforceオブジェクトとして商談を、送信先は作成した商談テーブルをSnowflakeオブジェクトとして選択しました。この設定から、連携オブジェクトにつき1本フローを作成する必要があることが分かります。
エラー処理も設定可能です。
フロートリガーも設定しました。
今回は増分転送を確認したかったので、スケジュール実行を設定しました。
続いてデータ送信元/送信先でデータフィールドをマッピングしました。
今回は手動でマッピングしました。
送信元から送信先フィールドへのマッピングで対応のペアを選び、マップフィールドを押して、マッピングを追加しました。
ここでポイントとなるのが、その他の設定にある削除されたレコードのインポートです。増分転送の場合はデータ送信元で削除された場合はデータ送信先にもその情報を送りたいと思うので、その場合はここをチェックします。送信元が小さいデータで、フローで連携したデータで毎回全量洗い替えするような場合は不要と思います。
連携フローではフィルタを設定することも可能でした。今回はなしで進めました。
最後に設定を見直して作成を完了しました。
フローは最初は下書きステータスなので、フローをアクティブ化でスケジュール実行を開始しました。
以下のようになればスケジュール実行が開始します。
3. AppFlowフローの増分転送結果を確認する
i. 結果の確認
ここまでの説明でフローは増分転送で実行されているので、後は転送を待つだけとなります。
フローの実行履歴を確認するとどのような転送がいつ行われたのか概要を把握できました。
Snowflakeの連携先テーブルを確認すると、Salesforceのデータが連携できていました。
いくつか私が検証用に作成・削除データが入っていますが、増分実行の場合、初回は直近30日以内のデータが連携されます。
IS_DELETEDカラムがTRUEの場合、削除済みレコードとなります。
レコードの状態はフロー実行断面のものが連携されました。
AppFlow向けテスト商談は作成後すぐに削除したため、削除後の状態のみ連携されました。
AppFlow向け商談newは作成して削除する間に一度連携を挟んでいるため、IS_DELETEDカラムがTRUEのものとFALSEのものが両方あります。
ほかのレコードもIS_DELETEDカラムがTRUEですが、これは事前に作成・削除していたレコードです。
オブジェクトはSalesforceの商談ページから、適当な名前でデータを作成しました。
個別の商談ページから削除を押すと、ゴミ箱に移動し、論理削除がされます。これがIS_DELETEDカラムがTRUEの状態です。
ii. オブジェクトのSnowflakeへの連携方法確認
S3上のオブジェクトも確認しておきます。
以下のようにバケットの詳細に設定したパス+フロー名のプレフィクスに連携データが配置されていました。
フローに設定したユーザーでクエリ履歴を確認すると、COPYコマンドで上記のパスからデータをロードしたことが分かります。
これらのことから、フローは一度S3にSalesforceより取得したデータを出力し、外部ステージ経由でCOPYコマンドでロードをすることが伺えました。もし認証など設定が上手くできているのにテーブルにロードできない場合は、ステージ上の取得データとCOPYコマンドの参照元が一致しているか確認してみると良いです。
最後に
SalesforceのデータをAmazon AppFlowでSnowflakeに増分連携してみたのでご紹介しました。
以前からある機能ではありますが、意外と設定方法の情報が少なかったり、細かな挙動は試してみないと分からなかったので、改めてまとめてみました。
若干設定にコツが必要なところもありましたが、価格も安いですし、必要に応じてPrivateLink経由の通信もできるため、かなり良い仕組みと思います。
参考になりましたら幸いです。
