ターミナルを開き、次のコマンドを実行して最新のTD Toolbeltをインストールします。
$ td --version
0.15.3Salesforce External Client Apps（ECA）でClient Credentials Flowを使用している場合は、以下の手順に従ってください。
- Setup > Apps > External Client Apps > External Client App Managerに移動します。
- 新しいアプリを追加するか、既存のアプリを編集します。
- SettingsタブでEditをクリックします。
- OAuth Settingsを展開します。
- 必要なOAuth Scopesを追加します。
- Enable Client Credentials Flowをチェックします。
- Saveをクリックします。
- PoliciesタブでEditをクリックします。
- OAuth Policiesを展開します。
- Enable Client Credentials Flowをチェックします。
- Run As (Username) を入力します — これはインテグレーションが実行されるSalesforceユーザーです。
- Saveをクリックします。
保存後、シード設定ファイルに必要となるアプリのConsumer Key（client_id）とConsumer Secret（client_secret）を控えておいてください。
Connected AppをExternal Client Appの代わりに使用している場合は、Setup > Apps > App Managerに移動し、New Connected Appを選択します。(手順は異なる場合があります。使用しているSalesforceのバージョンによって異なります。)
Manage Connected Appsを選択し、次のステップで必要となるConsumer KeyとConsumer Secretを取得します。
安全なアカウントアクセスのために、まだお持ちでない場合は、Salesforce Security Tokenを取得する必要があります。これを行うには、Account > Settings > Reset My Security Tokenに移動し、Reset Security Tokenを選択します。Security Tokenはメールで受け取ります。
テキストエディタを使用して、seed.ymlというファイルを作成します。認証方法に合った設定を選択してください。
以下の設定をコピーして貼り付け、プレースホルダーテキストをECAの認証情報に置き換えます。この設定は、replaceモードが指定されているため、targetフィールドで指定されたAccountオブジェクトをダンプします。利用可能な出力モードの詳細については、付録を参照してください。
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: replacelogin_urlにはSalesforceのMy Domain URLを指定してください（例: yourorg.my.salesforce.com）。Client Credentials Flowでは汎用URLのlogin.salesforce.comはサポートされていません。詳細はSalesforce OAuth 2.0 Client Credentials Flowを参照してください。
この方法ではusername、password、security_tokenは不要です。
Credentials（Username-Password）認証方法は非推奨です。Salesforce External Client Apps（ECA）はUsername-Password OAuth Flowをサポートしていません。上記のClient Credentials方法に移行してください。
以下の設定をコピーして貼り付け、プレースホルダーテキストをSalesforceの認証情報に置き換えます。
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ポータルの完全なリファレンスを参照してください。
アクセスできるオブジェクト、フィールド、レコードは、Salesforceライセンス、セキュリティ設定、およびSalesforce組織のAPIアクセス設定によって異なります。次のステップで認証エラーが表示される場合は、設定を確認してください。
ターミナルで次のコマンドを実行します:
$ td connector:guess seed.yml -o load.ymlConnector:guessは、対象データを自動的に読み取り、データ形式をインテリジェントに推測します。
load.ymlファイルを開くと、ファイル形式、エンコーディング、カラム名、型など、推測されたファイル形式定義が表示されます。
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: replacepreviewコマンドを使用して、システムがファイルをどのように解析するかをプレビューできます。
$ td connector:preview load.ymlシステムが誤ったカラム名または型を検出した場合は、load.ymlを変更して再度プレビューしてください。
Integrationは、"boolean"、"long"、"double"、"string"、"timestamp"型の解析をサポートしています。
ターミナルで、次の例に示すようにロードジョブを送信します。データサイズに応じて、処理に数時間かかる場合があります。ユーザーは、データが保存されるデータベースとテーブルを指定する必要があります。
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にデータベースまたはテーブルが存在しない場合、コマンドは成功しません。そのため、データベースとテーブルを手動で作成するか、次のようにtd connector:issueコマンドで--auto-create-tableオプションを使用して、データベースとテーブルを自動作成してください。
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'オプションを使用したロードファイルの例です:
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)を使用してデータをクエリおよびフィルタリングできます。SOQLを使用すると、単一のオブジェクトまたは相互に関連する複数のオブジェクトからデータを取得できます。必要な列を選択し、独自の条件文で出力をフィルタリングまたはソートできます。制限として、当社のData Connectorは、count() FROM ObjectなどのSOQL構文をサポートしていません。whereこの設定により、カスタムSOQLにフィルタリング条件を使用できますcolumnsこの設定は、Treasure Dataにインポートするデータのカスタムスキーマを定義するために使用されます。ここでは、関心のある列のみを定義できますが、それらが取得しているオブジェクトに存在することを確認してください。そうでない場合、これらの列は結果で使用できません。last_recordこの設定は、前回のロードジョブからの最後のレコードを制御するために使用されます。列名のキーと列の値の値を含むオブジェクトが必要です。キーはSFDC(SOQL)の列名と一致する必要があります。filter_schema_sampleこの設定は、スキーマサンプルデータクエリ中にWHERE条件と最後のレコード設定の適用を強制します。- デフォルトの動作: システムは、スキーマを構築するために十分なデータが取得されることを保証するために、フィルター条件を無視します。
- 有効にする場合: 非常に大きなデータセットでは、フィルターなしでクエリを実行するとタイムアウトまたはエラーが発生する可能性があります。定義したフィルターを使用してクエリ範囲を制限するには、この設定を有効にしてください。
以下は、'append'モードでの出力と組み合わせたincrementalモードを使用したシードファイルの例です。
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)を使用します。
-- 初回実行:
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)を使用したシードファイルの例です。
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を使用したシードファイルの例です。
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モードを使用したシードファイルの例です。
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は'ownership'フィールドに制限されたフィルターで、より少ない列を選択できます。スキーマは、不明なフィールド名、またはSOQLに存在しないフィールド名(つまりDesc)も定義しており、この列はインポート結果で使用できません。
td connector:createコマンドを使用して、新しいスケジュールを作成できます。スケジュールの名前、cron形式のスケジュール、データが保存されるデータベースとテーブル、およびData Connector設定ファイルが必要です。
$ td connector:create \
daily_sfdc_import \
"10 0 * * *" \
td_sample_db \
td_sample_table \
load.ymlcronパラメータは、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_tabletd 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 settd connector:deleteはスケジュールを削除します。
$ td connector:delete daily_sfdc_importSalesforceからTreasure Dataにレコードを取り込むには、2つのオプションがあります:
- Bulk API
- REST API
Bulk APIはより高速な取り込みを提供しますが、24時間以内に10,000バッチ割り当ての制限もあります。ターゲットが大きい場合、利用可能なバッチ割り当て全体が消費され、最終的にジョブが失敗する可能性があります。
すべてのレコードが取り込まれる最初のデータの取り込み中に、バッチ割り当ての制限に遭遇する可能性があります。
すべてのレコードの取り込みがユースケースの1つである場合は、同期転送のみオプションを有効にして、REST APIを使用することを検討してください。REST APIを使用すると、バッチ割り当ての制限を回避できますが、速度が遅くなる可能性があります。