Treasure Data CDP連携

Treasure AI Voiceは、組織の録音、文字起こし、AI生成の要約、デバイスインベントリを自社のTreasure Data CDPデータベースに直接プッシュできます。データがCDPに取り込まれると、他の顧客データと結合してセグメントの作成やジャーニーの実行が可能になり、会議や通話で実際に話された内容を分析できます。

この連携は、手動エクスポートと定期エクスポートに対応しています。エクスポートはバックグラウンドで実行され、Treasure AI Voiceは指定したTreasure Data CDPデータベースのテーブルにデータを書き込みます。書き込み先設定と専用の書き込み専用APIキーを保存して、後続の手動実行や管理者がサインインしていない状態での定期実行に使用できます。APIキー本体がコンソールに返されること、エクスポート実行のペイロードに含まれること、監査ログに書き込まれることはありません。

各エクスポートは、CDPデータベース内の4つの論理テーブルに書き込みます。recordings、transcript_segments、summariesテーブルはその実行の対象範囲の録音を反映します。devicesテーブルは録音日時ではフィルタリングされません。初回エクスポートでは全デバイスインベントリが含まれる場合がありますが、差分エクスポートではエクスポートチェックポイント以降に更新されたデバイスが含まれます。

録音は処理ステータスに関わらず含まれます。status列が各録音の状態を反映し、まだ文字起こしや要約が完了していない録音ではtranscription_text、summary_textおよび関連フィールドは空になります。

エクスポートは差分実行にできます。ソースレコードのupdated_atを使用して、前回のチェックポイント以降に変更されたレコードを特定します。日付範囲を指定すると、録音の候補をrecorded_atで絞り込みますが、差分チェックポイントの用途は変わりません。

すべての行には、ソースレコードから導出された安定したunique_idが付与されます。同じソースレコードは常に同じunique_idにマッピングされます。ただしTreasure Data CDPは受信イベントを既存行の置き換えではなく追記するため、同じunique_idを持つ行が複数存在する場合があります。updated_atを使用して最新のコピーを判定してください。

すべての行には、ソースレコードの最終更新時刻をUnixエポック秒で表すupdated_at列も付与されます。ソースの更新時刻が維持されていない古いレコードでは、テーブルごとの代替タイムスタンプがupdated_atに使用されます（各テーブルのupdated_atの説明を参照してください）。

すべての行にはtime列（Unixエポック秒）も付与され、Treasure Data CDPがネイティブの時間パーティショニングに使用します。これによりアナリストは時間で効率的にフィルタリングできます。time値は各テーブルで最も意味のあるタイムスタンプに紐付けられます。recordingsとsummariesはrecorded_at、セグメントは録音日時に発話オフセットを加えた時刻、デバイスは最終ハートビート（報告がない場合はプロビジョニング時刻）です。パーティションのフィルタリングにはtimeを使用し、最新行の判定と重複排除にはupdated_atを使用してください。

スキーマはTreasure AI Voiceが管理します。列は最初の書き込み時にCDPテーブルに追加されるため、事前に宣言する必要はありません。

エクスポートを設定する前に、以下を準備してください：

• Treasure AI VoiceのEnterprise Adminアカウント。
• 組織で連携が有効化されていること（下記参照）。
• 書き込み先データベースに対するImport OnlyまたはGeneral Access権限を持つユーザーのTreasure Data CDP APIキー。
• Treasure Data CDPの日本（AP01）リージョンにある書き込み先データベース。エクスポートは最初の書き込み時に宛先テーブルを作成します。

連携はデフォルトで無効です。有効にするには：

1. Enterprise AdminとしてAI Voice Consoleにサインインします。
2. Exportページを開きます。
3. Treasure Data CDPを選択します。
4. Enable exportをクリックします。

有効化するオプションが表示されない場合は、Treasure AIの担当者までお問い合わせください。

手動エクスポートまたは定期エクスポートを実行する前に、宛先と認証情報を保存します：

1. ExportページのSettingsタブを開きます。
2. 書き込み先のDatabaseを入力します。
3. 専用のWrite-only API Keyを入力します。
4. 必要に応じてAdvancedを展開し、Recordings tableとTranscript segments tableの名前を変更します。
5. Save settingsをクリックします。

データベースとテーブルの設定は、次回Exportページを開いたときに復元されます。認証情報を変更するにはReplaceをクリックして新しいキーを保存します。削除するにはRemoveをクリックします。保存済みキーを削除すると、有効な定期スケジュールも無効になります。

1. Settingsタブに書き込み先データベースと保存済みAPIキーが表示されていることを確認します。
2. 必要に応じてAdvancedを展開し、FromとToの日付範囲を設定します。
3. 必要に応じて以前にエクスポートした録音を再エクスポートを選択します。
4. Run export nowをクリックします。
5. Historyタブで実行状況を確認します。

日付範囲は録音日時で録音をフィルタリングします。Toの日付は、その日の終日を含みます。範囲は録音、文字起こしセグメント、要約に適用されます。録音日時によるデバイスのフィルタリングは行いません。

リクエストはバックグラウンド実行を作成し、エクスポートが完了する前に戻ります。開始後にページを開いたままにする必要はありません。

定期エクスポートでは、保存済みの宛先、テーブル設定、書き込み専用APIキーを使用します。利用できる間隔は以下のとおりです：

• 1時間ごと
• 6時間ごと
• 12時間ごと
• 24時間ごと
• 週次

最短の実行間隔は1時間です。カスタムcron式には対応していません。最初のスケジュール実行は、スケジュールを有効にした時点から開始するか、組織内の既存のエクスポート対象データをバックフィルしてから差分実行へ移行できます。

同じ組織で別のエクスポートが実行中の場合、定期エクスポートは開始されません。

Historyタブには、手動および定期エクスポートの実行履歴が表示されます。各実行は、以下のいずれかのステータスになります：

Historyの一覧には、開始時刻、実行方法、ステータス、受け入れられた行数と拒否された行数、所要時間、実行IDが表示されます。実行詳細では、書き込み先データベース、テーブルごとの件数、失敗情報、イベントサイズが大きすぎて拒否された録音を確認できます。

認証情報が無効な場合や、Treasure Data CDPデータベースへの権限が不足している場合は、非同期実行のためFailedとして表示されることがあります。書き込み先データベースに対するImport OnlyまたはGeneral Access権限をキーのユーザーに付与してから、エクスポートを再実行してください。

シリアライズ後のイベントが1MBの上限を超える録音は、実行詳細に表示されます。その録音行は拒否されますが、文字起こしセグメントと要約は送信される場合があります。録音は後続の実行で一定回数再試行されます。

各エクスポート実行は、TD_CDP_EXPORTイベントタイプとして監査ログに記録されます。ステータスと行数はHistoryタブと実行詳細で確認できますが、監査ログには表示されません。設定と認証情報の変更も記録されます。APIキー本体が監査ログに書き込まれることはありません。

定期エクスポートと日付範囲を指定しないエクスポートでは、差分チェックポイントを使用します。チェックポイント以降に更新された録音とデバイスが次回の実行に含まれます。日付範囲を指定すると、録音の候補をrecorded_atで絞り込みながら、差分チェックを維持します。

以前にエクスポートした録音を再エクスポートを選択すると、その実行ではチェックポイントを使わず、選択した範囲のすべてのレコードを再送信します。

Treasure Data CDPは受信イベントを既存行の置き換えではなく追記します。そのため、再実行によって同じunique_idを持つ行が複数存在する場合があります。ダウンストリームではunique_idを基準に重複排除し、updated_atが最大の行を残してください。

古いレコードでソースの更新時刻が維持されていない場合、updated_atにはテーブルごとの代替タイムスタンプが使用されます（各テーブルのupdated_atの説明を参照してください）。

運用手順と管理者権限については、Administrator Guideを参照してください。

各テーブルはrecording_id（recordings.id）を共有しているため、CDP内で会議データの任意のビューを再構築できます。例：

• summariesをrecording_idでrecordingsに結合すると、要約と完全な録音メタデータを一緒に取得できます。
• transcript_segmentsをrecording_idでrecordingsに結合し、segment_indexで並べ替えると、会話を発話単位で再構築・分析できます。
• transcript_segmentsをspeaker_label（またはspeaker_id）でピボットすると、参加者ごとの発話時間を分析できます。
• recordingsをdevice_serial_number = serial_numberでdevicesに結合すると、会話をデバイスと割り当てられたユーザーに紐付けられます。

WITH ranked AS (
  SELECT
    *,
    ROW_NUMBER() OVER (
      PARTITION BY unique_id
      ORDER BY updated_at DESC
    ) AS rn
  FROM your_table
)
SELECT *
FROM ranked
WHERE rn = 1

ソースの更新時刻が維持されていない古いレコードでは、updated_atにテーブルごとの代替タイムスタンプが使用されます（各テーブルのupdated_atの説明を参照してください）。

エクスポートされるテーブルのスキーマとクエリ例については、スキーマリファレンスを参照してください。