| name | td-agg-llm |
| description | Treasure Data上の複数DB/Tableからデータを1つのDatabaseに集約し、ID Unificationで共通IDを付与するワークフロー(td_agg_llm)の設定・デプロイを支援する。GitHubからワークフローを取得した後の設定カスタマイズ、デプロイ、トラブルシューティングに使用する。「td_agg_llm」「LLM向けデータ集約」「ID統合ワークフロー」「データ集約WF」「LLMデータ統合」「enrichedテーブル集約」などで発動する。 |
td_agg_llm ワークフロー設定スキル
ユーザーがGitHubからtd_agg_llmワークフローを取得した後、対話形式で設定をカスタマイズし、Treasure Dataにデプロイするまでを支援する。
ワークフロー概要
td_agg_llmは以下の処理を行うDigdagワークフロー:
- +main: 複数DB/Tableからデータを1つの集約DBにコピー
- +call_unification: ID Unification APIで共通IDを付与
- +push_llm_db:
enriched_* テーブルを集約DBにコピー
- +drop_tmp_table: 中間テーブル(enriched_*以外)を削除
参考: https://advanced.cdp-academy-lab.com/id_unif/intro.html
ファイル構成
td_agg_llm/
├── td_agg_llm.dig # メインワークフロー
├── config/
│ └── agg_datas.yml # データソース定義
├── unification_setting.yml # ID Unification設定
└── queries/
├── main.sql # データ集約SQL
├── target_table.sql # enriched_*テーブル一覧取得
└── drop_tmp_table.sql # 中間テーブル一覧取得
対話フロー
ユーザーにステップバイステップで以下を聞き取り、設定ファイルを生成する。
Step 1: 基本設定
以下を確認する:
- 集約先Database名 (デフォルト:
td_llm_support)
- リージョン →
td_endpoint を決定
- US:
api-cdp.treasuredata.com
- EU:
api-cdp.eu01.treasuredata.com
- JP:
api-cdp.ap1.treasuredata.com
- AP02:
api-cdp.ap02.treasuredata.com
- AP03:
api-cdp.ap03.treasuredata.com
- unification_name (デフォルト:
agg_llm) — cdp_unification_<この値> がID Unificationの出力DB名になる
- canonical_id_name (デフォルト:
unif_id) — 名寄せ後の共通IDカラム名
Step 2: データソース定義 (config/agg_datas.yml)
集約したいテーブルを聞き取る。各テーブルについて:
db: コピー元Databasetbl: コピー元Tableoutput_tbl_name: 集約先DBでのTable名where_condition: フィルタ条件（任意。TD_INTERVAL 等）
テンプレート:
datas:
data1:
db: <database名>
tbl: <table名>
output_tbl_name: <出力テーブル名>
where_condition:
tdx databases や tdx tables <db> でDatabase/Table一覧を確認可能。ユーザーのデータを探す際に活用する。
Step 3: ID Unification設定 (unification_setting.yml)
名寄せキーとテーブルマッピングを設定する:
- keys: 名寄せに使うキー（email, user_id, phone 等）
- tables: Step 2で定義した各テーブルのカラムとキーの対応
- canonical_ids: 名寄せ実行設定
merge_by_keys: 使用するキーの配列merge_iterations: マージ反復回数（デフォルト5）
テンプレート:
name: ${unification_name}
keys:
- name: <キー名>
invalid_texts: ['']
tables:
- database: ${td.database}
table: <output_tbl_name>
key_columns:
- {column: <カラム名>, key: <キー名>}
canonical_ids:
- name: ${canonical_id_name}
merge_by_keys: [<キー1>, <キー2>]
merge_iterations: 5
incremental_merge_iterations: 3
tdx describe <db>.<table> でカラム情報を確認し、名寄せに使えるキーカラムを特定する。
Step 4: 設定ファイル生成
聞き取り結果をもとに以下を生成:
td_agg_llm.dig の _export セクションを更新config/agg_datas.yml を生成unification_setting.yml を生成
Step 5: デプロイ
tdx wf push --project td_agg_llm <ワークフローディレクトリ>
デプロイ後、Secret td.apikey の設定を案内:
tdx wf secret --project td_agg_llm --set "td.apikey=TD1 <APIキー>"
重要: 値には必ず先頭に TD1 (半角スペース含む) を付けること。ID Unification API の内部 callback で authorization ヘッダーとしてそのまま使用されるため、TD1 が無いと Invalid authorization header エラーになる。
重要な注意点
- ID Unification機能がアカウントで有効になっている必要がある
+main はHive(Tez)エンジン、+push_llm_db/+drop_tmp_table はTrino(Presto)エンジンを使用- データ量が多い場合は
where_condition で期間を絞ることを推奨
unification_setting.yml の tables セクションは agg_datas.yml の output_tbl_name と一致させる- スケジュール実行する場合は
td_agg_llm.dig の schedule セクションのコメントアウトを解除
トラブルシューティング
| 症状 | 原因・対処 |
|---|
+call_unification で401エラー | Secret td.apikey が未設定または無効 |
callback で Invalid authorization header | Secret td.apikey の値に TD1 プレフィックスが付いていない。TD1 <APIキー> の形式で再設定する |
+call_unification で失敗 | ID Unification機能が未有効 / td_endpoint が間違っている |
enriched_* テーブルが空 | unification_setting.yml のキーマッピングを確認。実際のカラム名と一致しているか |
+main が遅い | where_condition で TD_INTERVAL を使って期間を絞る |
