# Salesforce Import Integration CLI ## コマンドラインを使用してSalesforce接続を作成する ### TD Toolbeltのインストール ターミナルを開き、次のコマンドを実行して最新の[TD Toolbelt](https://toolbelt.treasuredata.com/)をインストールします。 ```bash $ td --version 0.15.3 ``` ## Salesforceでの設定 ### Client Credentials FlowのためのExternal Client Appの設定 Salesforce External Client Apps（ECA）でClient Credentials Flowを使用している場合は、以下の手順に従ってください。 #### Settings Tab 1. **Setup** > **Apps** > **External Client Apps** > **External Client App Manager**に移動します。 2. 新しいアプリを追加するか、既存のアプリを編集します。 3. **Settings**タブで**Edit**をクリックします。 4. **OAuth Settings**を展開します。 5. 必要な**OAuth Scopes**を追加します。 6. **Enable Client Credentials Flow**をチェックします。 7. **Save**をクリックします。 ![](/assets/external_client_app_1.1a5bf858a7851f89bb86a4813a3bdef56c3d032b5afc00a870aae64eeccf9224.3074136a.png) #### Policies Tab 1. **Policies**タブで**Edit**をクリックします。 2. **OAuth Policies**を展開します。 3. **Enable Client Credentials Flow**をチェックします。 4. **Run As (Username)** を入力します — これはインテグレーションが実行されるSalesforceユーザーです。 5. **Save**をクリックします。 ![](/assets/external_client_app_2.aa267ddc0670ac270fa240720a9f2e1ea5e7a287e4703fe97f228eca8d6e76ca.3074136a.png) 保存後、シード設定ファイルに必要となるアプリの**Consumer Key**（`client_id`）と**Consumer Secret**（`client_secret`）を控えておいてください。 ### Connected Appの設定（非推奨） Connected AppをExternal Client Appの代わりに使用している場合は、Setup > Apps > App Managerに移動し、**New Connected App**を選択します。([手順は異なる場合があります](https://help.salesforce.com/search#q=connected+apps&f:@sflanguage=%5Ben_US%5D&firstQueryMeta=%5Bobject+Object%5D)。使用しているSalesforceのバージョンによって異なります。) ![](/assets/salesforce-app1.7415e65f64f26ecb256ee3f9cd44ebc650aa76c8370d1ee4519df072cffa18ed.6d7a3f8b.png) **Manage Connected Apps**を選択し、次のステップで必要となるConsumer KeyとConsumer Secretを取得します。 安全なアカウントアクセスのために、まだお持ちでない場合は、Salesforce Security Tokenを取得する必要があります。これを行うには、Account > Settings > Reset My Security Tokenに移動し、**Reset Security Token**を選択します。Security Tokenはメールで受け取ります。 ## シード設定ファイル(seed.yml)の作成 テキストエディタを使用して、seed.ymlというファイルを作成します。認証方法に合った設定を選択してください。 ### Client Credentials（External Client Appsに推奨） 以下の設定をコピーして貼り付け、プレースホルダーテキストをECAの認証情報に置き換えます。この設定は、replaceモードが指定されているため、targetフィールドで指定されたAccountオブジェクトをダンプします。利用可能な出力モードの詳細については、付録を参照してください。 ```yaml in: type: sfdc_v2 auth_method: client_credentials client_id: "CLIENT_ID" # アプリのconsumer key client_secret: "CLIENT_SECRET" # アプリのconsumer secret login_url: yourorg.my.salesforce.com # My Domain URLが必須。login.salesforce.comは非対応 incremental: false # 'full dump'、以下のScheduled executionを参照 target: Account # インポートしたいSalesforceオブジェクト out: mode: replace ``` My Domain URLが必須 `login_url`にはSalesforceの**My Domain URL**を指定してください（例: `yourorg.my.salesforce.com`）。Client Credentials Flowでは汎用URLの`login.salesforce.com`はサポートされていません。詳細は[Salesforce OAuth 2.0 Client Credentials Flow](https://help.salesforce.com/s/articleView?id=xcloud.remoteaccess_oauth_client_credentials_flow.htm)を参照してください。 この方法では`username`、`password`、`security_token`は不要です。 ### Credentials（非推奨） Deprecated **Credentials**（Username-Password）認証方法は非推奨です。Salesforce External Client Apps（ECA）はUsername-Password OAuth Flowをサポートしていません。上記の**Client Credentials**方法に移行してください。 以下の設定をコピーして貼り付け、プレースホルダーテキストをSalesforceの認証情報に置き換えます。 ```yaml in: type: sfdc_v2 username: "USERNAME" # ユーザー名 password: "PASSWORD" # パスワード security_token: "SECURITY_TOKEN" # Security Token client_id: "CLIENT_ID" # アプリのconsumer key client_secret: "CLIENT_SECRET" # アプリのconsumer secret login_url: login.salesforce.com # サンドボックスの場合はtest.salesforce.com incremental: false # 'full dump'、以下のScheduled executionを参照 target: Account # インポートしたいSalesforceオブジェクト out: mode: replace ``` すべてのSalesforceオブジェクトがtargetオプションでサポートされています。一般的なオブジェクトには次のものがあります: - Opportunity - Contact - Lead - Account - Event - Task Salesforceオブジェクトの詳細については、Salesforce Developerポータルの[完全なリファレンス](https://developer.salesforce.com/docs/atlas.en-us.api.meta/api/data_model.htm)を参照してください。 アクセスできるオブジェクト、フィールド、レコードは、Salesforceライセンス、セキュリティ設定、およびSalesforce組織のAPIアクセス設定によって異なります。次のステップで認証エラーが表示される場合は、設定を確認してください。 ## Guess Fieldsコマンドの実行(load.ymlの生成) ターミナルで次のコマンドを実行します: ```bash $ td connector:guess seed.yml -o load.yml ``` Connector:guessは、対象データを自動的に読み取り、データ形式をインテリジェントに推測します。 load.ymlファイルを開くと、ファイル形式、エンコーディング、カラム名、型など、推測されたファイル形式定義が表示されます。 ```yaml in: type: sfdc_v2 username: "USERNAME" password: "PASSWORD" security_token: "SECURITY_TOKEN" client_id: "CLIENT_ID" client_secret: "CLIENT_SECRET" login_url: login.salesforce.com ... filters: ... out: mode: replace ``` previewコマンドを使用して、システムがファイルをどのように解析するかをプレビューできます。 ```bash $ td connector:preview load.yml ``` システムが誤ったカラム名または型を検出した場合は、load.ymlを変更して再度プレビューしてください。 Integrationは、"boolean"、"long"、"double"、"string"、"timestamp"型の解析をサポートしています。 ## ロードジョブの実行 ターミナルで、次の例に示すようにロードジョブを送信します。データサイズに応じて、処理に数時間かかる場合があります。ユーザーは、データが保存されるデータベースとテーブルを指定する必要があります。 ```bash td connector:issue load.yml --database td_sample_db \ --table td_sample_table \ --time-column createddate ``` 上記のコマンドは、*database(td_sample_db)*と*table(td_sample_table)*を既に作成していることを前提としています。TDにデータベースまたはテーブルが存在しない場合、コマンドは成功しません。そのため、[データベースとテーブルを手動で作成](https://docs.treasuredata.com/smart/project-product-documentation/data-management)するか、次のようにtd connector:issueコマンドで--auto-create-tableオプションを使用して、データベースとテーブルを自動作成してください。 ```bash td connector:issue load.yml \ --database td_sample_db --table td_sample_table \ --time-column createddate --auto-create-table ``` "—time-column"オプションを使用して、Time Format列を"Partitioning Key"に割り当てることができます。 SFDCで削除されたレコードを取り込む場合は、include_deleted_records: trueを指定します。デフォルトでは、取り込まれたレコードには、以前にSalesforceで削除したレコードは含まれません。 以下は、'include_deleted_records'オプションを使用したロードファイルの例です: ```yaml in: type: sfdc_v2 username: "USERNAME" password: "PASSWORD" security_token: "SECURITY_TOKEN" client_id: "CLIENT_ID" client_secret: "CLIENT_SECRET" login_url: login.salesforce.com include_deleted_records: true # 削除されたレコードを取り込む ... filters: ... out: mode: replace ``` ## コマンドラインを使用したスケジュール実行 定期的なSFDCインポートのために、定期的なData Connector実行をスケジュールできます。高可用性を確保するために、スケジューラーを慎重に管理しています。この機能を使用することで、ローカルデータセンターに`cron`デーモンを配置する必要がなくなります。 スケジュールされたインポートの場合、Data Connector for SFDCは、指定されたtargetに一致するすべてのObjectsをインポートします。 スケジュール実行は、SFDCからデータを取得する際のData Connectorの動作を制御する追加の設定パラメータをサポートしています: - `incremental` この設定は、各Objectに関連付けられたネイティブのタイムスタンプフィールドの1つに基づいて、Data ConnectorがSFDCからデータを取得する方法を制御するロードモードを制御するために使用されます。 - `incremental: true` (デフォルト) このモードでは、data connectorは、connectorの前回の実行以降に更新された指定されたSalesforce Object型のレコードのみを取得します。このモードは、前回のスケジュール実行以降に変更されたObject targetsのみを取得したい場合に便利です。このモードは通常、'append'モードを使用して宛先テーブルにデータを書き込むことと組み合わせて使用されます。 - `incremental_columns` (必須) このオプションは、SFDCから必要なデータのみをロードするために、incrementalモードに必要です。カスタムSOQLを使用する場合は、これらの列が選択されていることを確認してください。 - `incremental: false` このモードでは、data connectorは、最後に更新された時期に関係なく、指定されたSalesforce Object型のすべてのレコードを取得します。このモードは、'replace'モードを使用して宛先テーブルにデータを書き込むことと組み合わせて使用するのが最適です。 - `soql` この設定により、カスタムSOQL([Salesforce Object Query Language](https://developer.salesforce.com/docs/atlas.en-us.soql_sosl.meta/soql_sosl/sforce_api_calls_soql.htm))を使用してデータをクエリおよびフィルタリングできます。SOQLを使用すると、単一のオブジェクトまたは相互に関連する複数のオブジェクトからデータを取得できます。必要な列を選択し、独自の条件文で出力をフィルタリングまたはソートできます。制限として、当社のData Connectorは、count() FROM ObjectなどのSOQL構文をサポートしていません。 - `where` この設定により、カスタムSOQLにフィルタリング条件を使用できます - `columns` この設定は、Treasure Dataにインポートするデータのカスタムスキーマを定義するために使用されます。ここでは、関心のある列のみを定義できますが、それらが取得しているオブジェクトに存在することを確認してください。そうでない場合、これらの列は結果で使用できません。 - `last_record` この設定は、前回のロードジョブからの最後のレコードを制御するために使用されます。列名のキーと列の値の値を含むオブジェクトが必要です。キーはSFDC(SOQL)の列名と一致する必要があります。 - `filter_schema_sample` この設定は、スキーマサンプルデータクエリ中にWHERE条件と最後のレコード設定の適用を強制します。 - デフォルトの動作: システムは、スキーマを構築するために十分なデータが取得されることを保証するために、フィルター条件を無視します。 - 有効にする場合: 非常に大きなデータセットでは、フィルターなしでクエリを実行するとタイムアウトまたはエラーが発生する可能性があります。定義したフィルターを使用してクエリ範囲を制限するには、この設定を有効にしてください。 以下は、'append'モードでの出力と組み合わせたincrementalモードを使用したシードファイルの例です。 ```yaml in: type: sfdc_v2 username: "USERNAME" password: "PASSWORD" security_token: "SECURITY_TOKEN" client_id: "CLIENT_ID" client_secret: "CLIENT_SECRET" login_url: login.salesforce.com target: Account use_rest: false incremental: true incremental_columns: - LastModifiedDate out: mode: append ``` 上記の設定には、レコードを段階的にロードするために使用される`LastModifiedDate`という列名があります。Data Connectorは、次のようにその設定でSOQL(SFDC Query Language)を使用します。 ```sql -- 初回実行: SELECT Id, Name, ... FROM Account ORDER BY LastModifiedDate -- 2回目の実行: SELECT Id, Name, ... FROM Account WHERE (LastModifiedDate > 2015-10-21T10:33:07+0000) ORDER BY LastModifiedDate -- '2015-10-21T10:33:07+0000'は初回実行からの最新の変更レコード時刻 -- 3回目の実行: SELECT Id, Name, ... FROM Account WHERE (LastModifiedDate > 2015-10-30T22:30:41+0000) ORDER BY LastModifiedDate -- '2015-10-30T22:30:41+0000'は2回目の実行からの最新の変更レコード時刻 ``` WHERE句が各実行で更新され、変更されたレコードのみがロードされることがわかります。 以下は、'replace'モードでの出力と組み合わせた非incrementalモード(`incremental: false`)を使用したシードファイルの例です。 ```yaml in: type: sfdc_v2 username: "USERNAME" password: "PASSWORD" security_token: "SECURITY_TOKEN" client_id: "CLIENT_ID" client_secret: "CLIENT_SECRET" login_url: login.salesforce.com incremental: false target: Account out: mode: replace ``` 上記の設定では、Data Connectorは常にすべてのレコードをロードし、既存のすべてのレコードをロードされたレコードで置き換えます。 以下は、'append'モードでの出力と組み合わせたincrementalモードとlast_recordを使用したシードファイルの例です。 ```yaml in: type: sfdc_v2 username: "USERNAME" password: "PASSWORD" security_token: "SECURITY_TOKEN" client_id: "CLIENT_ID" client_secret: "CLIENT_SECRET" login_url: login.salesforce.com target: Account incremental: true incremental_columns: - LastModifiedDate last_record: - {"key": "LastModifiedDate", "value": "2017-01-01T00:00:00Z"} out: mode: append ``` 上記の設定では、Integrationは2017-01-01 00:00:00以降のすべてのレコードをロードします。 以下は、出力用のカスタムSOQL、where条件、columnsスキーマと組み合わせたincrementalモードを使用したシードファイルの例です。 ```yaml in: type: sfdc_v2 username: "USERNAME" password: "PASSWORD" security_token: "SECURITY_TOKEN" client_id: "CLIENT_ID" client_secret: "CLIENT_SECRET" _url: .salesforce.com target: Account incremental: true incremental_columns: - LastModifiedDate soql: "select Id, Name, Description, LastModifiedDate from Account" where: "ownership = 'Public'" columns: - name: Id type: string - name: Name type: string - name: Desc type: string - name: LastModifiedDate type: string out: mode: append ``` ご覧のように、[SOQLは](https://developer.salesforce.com/docs/atlas.en-us.soql_sosl.meta/soql_sosl/sforce_api_calls_soql_select.htm)'ownership'フィールドに制限されたフィルターで、より少ない列を選択できます。スキーマは、不明なフィールド名、またはSOQLに存在しないフィールド名(つまりDesc)も定義しており、この列はインポート結果で使用できません。 ## スケジュールの作成 `td connector:create`コマンドを使用して、新しいスケジュールを作成できます。スケジュールの名前、cron形式のスケジュール、データが保存されるデータベースとテーブル、およびData Connector設定ファイルが必要です。 ``` $ td connector:create \ daily_sfdc_import \ "10 0 * * *" \ td_sample_db \ td_sample_table \ load.yml ``` `cron`パラメータは、3つの特別なオプションも受け入れます: `@hourly`、`@daily`、`@monthly`。 デフォルトでは、スケジュールはUTCタイムゾーンで設定されます。`-t`または`--timezone`オプションを使用して、タイムゾーンでスケジュールを設定できます。`--timezone`オプションは、'Asia/Tokyo'、'America/Los_Angeles'などの拡張タイムゾーン形式のみをサポートしています。PST、CSTなどのタイムゾーンの略語は*サポートされておらず*、予期しないスケジュールにつながる可能性があります。 ## スケジュールのリスト表示 `td connector:list`コマンドを入力することで、スケジュールされたエントリのリストを表示できます。 ``` $ td connector:list +-------------------+--------------+----------+-------+--------------+-----------------+------------------------+ | Name | Cron | Timezone | Delay | Database | Table | Config | +-------------------+--------------+----------+-------+--------------+-----------------+------------------------+ | daily_sfdc_import | 10 0 * * * | UTC | 0 | td_sample_db | td_sample_table | {"type"=>"sfdc", ... } | +-------------------+--------------+----------+-------+--------------+-----------------+------------------------+ ``` ## スケジュールの設定と履歴の表示 `td connector:show`は、スケジュールエントリの実行設定を表示します。 ``` % td connector:show daily_sfdc_import Name : daily_sfdc_import Cron : 10 0 * * * Timezone : UTC Delay : 0 Database : td_sample_db Table : td_sample_table ``` `td connector:history`は、スケジュールエントリの実行履歴を表示します。個々の実行の結果を調査するには、`td job jobid`を使用します。 ``` % td connector:history daily_sfdc_import +--------+---------+---------+--------------+-----------------+----------+---------------------------+----------+ | JobID | Status | Records | Database | Table | Priority | Started | Duration | +--------+---------+---------+--------------+-----------------+----------+---------------------------+----------+ | 578066 | success | 10000 | td_sample_db | td_sample_table | 0 | 2015-04-18 00:10:05 +0000 | 160 | | 577968 | success | 10000 | td_sample_db | td_sample_table | 0 | 2015-04-17 00:10:07 +0000 | 161 | | 577914 | success | 10000 | td_sample_db | td_sample_table | 0 | 2015-04-16 00:10:03 +0000 | 152 | | 577872 | success | 10000 | td_sample_db | td_sample_table | 0 | 2015-04-15 00:10:04 +0000 | 163 | | 577810 | success | 10000 | td_sample_db | td_sample_table | 0 | 2015-04-14 00:10:04 +0000 | 164 | | 577766 | success | 10000 | td_sample_db | td_sample_table | 0 | 2015-04-13 00:10:04 +0000 | 155 | | 577710 | success | 10000 | td_sample_db | td_sample_table | 0 | 2015-04-12 00:10:05 +0000 | 156 | | 577610 | success | 10000 | td_sample_db | td_sample_table | 0 | 2015-04-11 00:10:04 +0000 | 157 | +--------+---------+---------+--------------+-----------------+----------+---------------------------+----------+ 8 rows in set ``` ## スケジュールの削除 `td connector:delete`はスケジュールを削除します。 ``` $ td connector:delete daily_sfdc_import ``` ## 同期転送のみを使用 SalesforceからTreasure Dataにレコードを取り込むには、2つのオプションがあります: - Bulk API - REST API Bulk APIはより高速な取り込みを提供しますが、24時間以内に10,000バッチ割り当ての制限もあります。ターゲットが大きい場合、利用可能なバッチ割り当て全体が消費され、最終的にジョブが失敗する可能性があります。 すべてのレコードが取り込まれる最初のデータの取り込み中に、バッチ割り当ての制限に遭遇する可能性があります。 すべてのレコードの取り込みがユースケースの1つである場合は、**同期転送のみ**オプションを有効にして、REST APIを使用することを検討してください。REST APIを使用すると、バッチ割り当ての制限を回避できますが、速度が遅くなる可能性があります。