Treasure ワークフローを使用した増分データImportについて

データパイプラインの管理の一環として、データの増分転送を指定し、データ転送の開始タイミングを制御できます。Treasure ワークフローを使用して、必要なデータがアカウントに取り込まれた後に、Treasure Data内でクエリ処理ステップを確実に実行できます。

多くのバッチImport Integrationは、さまざまなユースケースに対応した増分データロード機能をサポートしています。

テーブルが大きすぎて、定期的にテーブル全体を再Importすることができない場合（例：大規模な本番データベースから）
更新されたデータを頻繁に（例：15分ごとに）Importして、データを最新の状態に保つ場合
取り込む行数を最小限に抑えて、Treasure アカウントプランの容量を最も効率的に使用する場合

タイミング、スケジューリング、および増分処理について

増分処理は、Importするテーブルとレコードの列値（例：timeまたはID列）を追跡し、最後の取り込み時にImportされた最高値を使用して、次の取り込みを開始することで機能します。例：

from_date（last_fetched_dateまたは任意の増分フィールド）は、各実行後に更新され保存されます。新しい値は次回の実行時に使用されます。

例えば、Mixpanel Import Integrationを使用する場合：

最初の実行では、すべてのデータをImportします
その後の増分実行では、last_fetched_time（前回の実行からの最大取り込みタイムスタンプ）を使用します

アプローチの決定

Workflowの一部としてImport Integrationをトリガーする場合、以下の2つのアプローチがあります。

Integration自体の増分ロード機能を使用する方法。この場合、事前定義されたSourceのUnique IDがWorkflow定義で参照されます
Workflowのセッション時間（Workflow変数経由）をIntegrationパラメータに注入する機能を利用する方法。このアプローチでは、Import IntegrationパラメータがYAMLファイルで宣言されます

Unique IDベースのアプローチ

長所
- Treasure コンソールでSourceを直接作成および管理でき、セットアップがよりユーザーフレンドリーです
- Integrationが提供するID、ファイル名、または時間ベースのフィールドを使用した増分ロジックをサポートします
- 自己修復動作：1回の取り込みが失敗しても次回が成功すれば、最後に成功した実行からの差分が計算されます
短所
- 遅延データがあるSourceには最適ではありません。デフォルトの増分ロジックでは、見逃されたレコードが取得されない可能性があります
- 過去のWorkflowセッションを再実行しても、過去のデータは再ロードされません。最新の増分値がIntegration設定に保存されています

YAMLファイルベースのアプローチ

長所
- 柔軟なタイミング制御を提供します。Sourceシステムが遅延到着データを生成する場合に便利です
短所
- 増分ロジックが日付によって駆動できる場合にのみ適用されます

Unique ID増分データ転送の設定

Integration Hubで、Import IntegrationのためのAuthenticationを作成します
上記で作成したAuthenticationからSourceを作成し、増分ロード機能を選択します

Integration Hub > SourcesでSourceを見つけ、**...**メニューを選択し、Copy Unique IDを選択します

Data Workbench > Workflowsに移動します
新しいWorkflowを作成するか、既存のWorkflowを選択してWorkflow定義を開きます
クリップボードに保存されたUnique ID（例：s3_v2_import_1725874557）を使用して、td_load >: コマンドを使用します

増分データロードは、Workflowが手動またはスケジュールによって実行されるたびに発生します

YAML増分データ転送の設定

このアプローチでは、Workflowによって提供される時間ベースの変数を使用して、Import Integrationのカスタム設定を作成します。増分フローが時間ベースの場合に、この方法を使用できます。

この部分的な例では、Integration設定ファイルはdaily_load.ymlであり、Workflow定義で参照されます。変数last_session_timeは、多くのWorkflow変数の中から使用されます。Workflowセッション時間と変数の詳細については、Treasure ワークフロー基本を参照してください。

Data Workbench > Workflowsに移動します
新しいWorkflowを作成するか、既存のWorkflowを選択してWorkflow定義を開きます
設定ファイル名を指定したtd_load >: コマンドを使用し、last_session_timeの変数を追加します

schedule:
  daily>: 07:00:00

_export:
  td:
    dest_db: my_sample_db
    dest_table: my_sample_table
  wf:
    start_time: "${last_session_time}"

+data_import:
  td_load>: config/daily_load.yml
  database: ${td.dest_db}
  table: ${td.dest_table}

config/daily_load.ymlを編集して、Workflowによって提供されるstart_time変数を使用します

in:
  type: salesforce
  login_url: https://<YOUR_DOMAIN_NAME>.salesforce.com
  auth_method: token
  username: <YOUR_EMAIL_ADDRESS>
  token: <YOUR_API_TOKEN>
  target: tickets
  start_time: ${wf.start_time} #Workflow変数を使用
out:
  mode: append

Workflowは毎日実行され、last_session_time以降のデータを増分的に取得します

Treasure ワークフローを使用した増分データImportについて

タイミング、スケジューリング、および増分処理について

アプローチの決定

Unique ID増分データ転送の設定

YAML増分データ転送の設定

Was this helpful?