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

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

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

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


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

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

![](../../../../../images/products/customer-data-platform/integration-hub/batch/import/image-20190829-234802.png)

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

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

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


## アプローチの決定

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

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


**Unique IDベースのアプローチ**

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


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

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


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

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


![](../../../../../images/products/customer-data-platform/integration-hub/batch/import/about-incremental-data-import-using-td-workflow-2024-09-10.png)

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


![](../../../../../images/products/customer-data-platform/integration-hub/batch/import/using-td-workflow-with-td-integrations-2024-08-09.png)

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


![](../../../../../images/products/customer-data-platform/integration-hub/batch/import/about-incremental-data-import-using-td-workflow-2024-09-10-1.png)

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


## YAML増分データ転送の設定

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

この部分的な例では、Integration設定ファイルは**daily_load.yml**であり、Workflow定義で参照されます。変数**last_session_time**は、多くの[Workflow変数](https://docs.digdag.io/workflow_definition.html#using-variables)の中から使用されます。Workflowセッション時間と変数の詳細については、[Treasure ワークフロー基本](/products/customer-data-platform/data-workbench/workflows/treasure-workflow-terms-and-concepts)を参照してください。

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


```yaml
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}
```

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


```yaml
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
```

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