Batch Export Integration

このインテグレーションは、Treasure AIからBatchへデータを直接エクスポートします。Batchは、モバイルプッシュ、Webプッシュ、アプリ内メッセージング、メール、SMSにわたるオムニチャネルキャンペーンのための顧客エンゲージメントプラットフォーム（CEP）です。これは2つのターゲットを持つ1つのコネクタです。target: Profile（デフォルト）を設定するとProfile Mass Update APIを通じてBatchプロファイルを更新し、target: Audienceを設定するとAudience APIを通じてセグメントメンバーシップをBatchオーディエンスにプッシュします。

このインテグレーションは、各クエリ結果の行をBatchの操作にマッピングします。識別子、ネイティブ属性、カスタム属性、イベントはすべて、クエリ内の列名プレフィックス規約で宣言されるため、列ごとの設定は不要です。Profileモードでは、Attributes同期（1行で1プロファイルを更新）またはEvents同期（1行で1イベントを記録し、識別子ごとにグループ化）を実行できます。Audienceモードでは、1行が1つのオーディエンスメンバーシップエントリにマッピングされます。2つのターゲットの比較については、ターゲットマッピングの概要を参照してください。

• Treasure AIの基本的な知識。
• Batchのプロファイル、custom ID、installation識別子、ネイティブ属性、イベントに関する基本的な知識。
• Batch REST APIキー。
• Batchプロジェクトキー。
• identifier__列名プレフィックスで必要な識別子列を公開するクエリ:
  • target: Profileの場合: identifier__custom_id、またはidentifier__apikeyとidentifier__installation_idの組み合わせ。
  • target: Audienceの場合: identifier__id。

セキュリティポリシーで IP ホワイトリストが必要な場合は、接続を成功させるために Treasure Data の IP アドレスを許可リストに追加する必要があります。

リージョンごとに整理された静的 IP アドレスの完全なリストは、次のリンクにあります: IP Addresses for Integrations

• 配列値の属性はすべて置換専用です。string_array__カスタム属性と$topic_preferencesは常に完全な配列として送信されます。Batch APIは増分操作をサポートしていますが、このバージョンでは増分の$add/$remove操作はサポートされていません。クエリで目的の配列全体を出力する必要があります。
• batch__プレフィックスのオプトアウトはありません。batch__regionという名前の列は常にネイティブの$region属性にマッピングされます。代わりにカスタム属性として送信するには、プレフィックスのない名前に列をエイリアスしてください。
• Profile Mass Update API（target: Profile）は、受理応答またはエラー応答を即座に返し、オーディエンス形式のインデックストークンやポーリングフローは提供しません。Audience API（target: Audience）はインデックストークンを返し、ポーリングをサポートします。その動作については、オーディエンスのインデックス作成とポーリングを参照してください。両者を混同しないでください。
• スループット（1秒あたり10,000プロファイル更新）はクライアント側で能動的にスロットリングされません。429応答のみがバックオフをトリガーします。
• 列名は自動的に小文字に正規化されるため、大文字や大小文字が混在した名前も受け付けられます。小文字化した後でも^[a-z0-9_]{1,30}$に一致しない名前（例：ハイフンやドットを含む、または30文字を超える名前）は、クエリでエイリアスする必要があります。大小文字の変換以外の実行時のリネームはサポートされていません。
• レート制限（429）とサーバーエラー（5xx、ネットワークタイムアウト）は指数バックオフで再試行されます。400と401応答は、ペイロードまたは認証の問題を示すため、即座に失敗します。
• Audienceモードには部分エラーの概念がありません。BatchはオーディエンスのエラーをトップレベルのHTTPステータスとしてのみ報告します。ignore_partial_errorsは適用されません。

Treasure AIでは、クエリを実行する前にデータ接続を作成して設定する必要があります。データ接続の一部として、Batchにアクセスするための認証を提供します。

1. TD Consoleを開きます。
2. Integrations Hub > Catalogに移動します。
3. Catalogの右端にある検索アイコンをクリックして、Batchと入力します。
4. Batchコネクタにカーソルを合わせて、Create Authenticationを選択します。

New Authenticationダイアログが開きます。

1. BatchのREST API Keyを入力します。
2. BatchのProject Keyを入力します。
3. 接続の名前を入力します。
4. Continueを選択します。

インテグレーションを設定するには：

1. TD Consoleを開きます。
2. Data Workbench > Queriesに移動します。
3. New Queryをクリックしてクエリを作成するか、エクスポートしたい保存済みクエリを開きます。
4. クエリエディタの上部で、Export Resultsを選択します。
5. 作成したBatch認証を選択します。
6. エクスポートパラメータを設定します。

1. 列プレフィックスの検出：インテグレーションは各クエリ結果の列名を検査し、識別子（identifier__）、ネイティブ属性（batch__）、型プレフィックス付きカスタム属性（date__、url__、string_array__、object__、array__）、またはプレーンなカスタム属性（プレフィックスなし、Embulkの列型から型を推論）に分類します。
2. 識別子の抽出：Profileモードでは、インテグレーションは各行についてidentifier__値を読み取り、custom_idまたはinstallation形式（apikey + installation_id）を使用してidentifiersブロックを構築します。Audienceモードでは、identifier__idを読み取り、ids[]エントリを構築します。
3. 属性またはイベントのマッピング：Attributesモードでは、ネイティブ列は予約された$…ワイヤキーで出力され、カスタム列はラップされたキーまたはプレーンキーで出力されます。Eventsモードでは、各行が1つのイベントを構築し、イベントは識別子ごとにevents[]配列にグループ化されます。Audienceモードでは、各行の識別子以外の列がids[].attributesになります。
4. 行ごとの検証：インテグレーションは、送信前にキー形式、値の制限、ネイティブ属性の型、および25 kBのシリアライズされた属性サイズを検証します。
5. バッチ処理：操作はバッファリングされ、バッチ単位（最大batch_size、リクエストあたり最大10,000操作）でフラッシュされます。
6. 再試行と応答処理：429およびサーバーエラーはバックオフで再試行されます。Profileモードでは、BatchのSUCCESS_WITH_PARTIAL_ERRORS応答はignore_partial_errorsに従って処理されます。Audienceモードでは、非同期のインデックストークンがオーディエンスのインデックス作成設定に従ってポーリングされます。

コネクタには、条件付きフィールドを持つ単一の設定サーフェスがあります。適用対象列は、パラメータが両方のターゲットに共通か、ProfileまたはAudienceに固有かを示します。

targetがProfileの場合はこの設定を使用します。

targetがAudienceの場合はこの設定を使用します。

識別子と属性の型は、クエリ結果スキーマの列名プレフィックスで宣言されます。列の設定は不要です。

列名は大小文字を区別せずにマッチングおよびシリアライズされます。インテグレーションはプレフィックスを検出してBatchのワイヤキーを導出する前に各列名を小文字化するため、IDENTIFIER__Custom_Id、BATCH__Email_Address、プレーンなLoyaltyTierはそれぞれidentifier__custom_id、batch__email_address、loyaltytierとして扱われます。そのため、カスタム属性キーは常に小文字でBatchに送信されます（Batchはカスタムキーが^[a-z0-9_]{1,30}$に一致することを要求します）。Eventsモードでは、batch__nameの値も検証および送信の前に小文字化されるため、Purchase_Completedはpurchase_completedになります。

同じプレフィックスモデルが両方のターゲットに適用されますが、各プレフィックスは意味をなす場所でのみ受け付けられます。あるターゲットで拒否される列は、エクスポート時に設定エラーになります。

Attributesモードでサポートされるbatch__ネイティブ属性名は、email_address、email_marketing、phone_number、sms_marketing、email_open_tracking_consent、language、timezone、region、topic_preferencesです。Eventsモードでは、batch__プレフィックスは代わりに予約済みイベントフィールドname（必須）、time、label、tagsを保持します。ネイティブ属性セットはprofile_sync_typeによってゲートされます。プロファイルネイティブ名はEventsモードで拒否され、イベント予約名はAttributesモードで拒否されます。

コネクタは、プレーンな（プレフィックスなしの）カスタム属性のBatch型をTreasure AIの列型から推論します。

オーディエンスモードでは、単一の識別子列identifier__idを使用します。Profileの識別子サフィックス（identifier__custom_id、identifier__apikey、identifier__installation_id）は、このモードでは拒否されます。profile_sync_typeとignore_partial_errorsも無視されます — オーディエンスにはEventsモードや部分エラーの概念がありません。

識別子以外のすべての列はids[].attributesの下のカスタム属性になり、1つのidあたり最大15属性（プロファイルの50に対して）に制限されます。サポートされるのはスカラープレフィックスdate__、url__、およびプレーン（プレフィックスなし）のみです。コンテナプレフィックスstring_array__、object__、array__はサポートされておらず、設定エラーになります。ネイティブプロファイル属性とイベントはオーディエンスには適用されないため、このモードではbatch__プレフィックスはサポートされていません — batch__列が存在すると設定エラーになります。アクションは完全にaudience_sync_modeから決まります。

create_audience_if_not_exists: trueの場合、ジョブはまずオーディエンスが存在するかどうかを確認し、メンバーシップをプッシュする前にそれを作成します（POST /audiences/create、audience_typeとオプションのaudience_display_nameを使用）。falseの場合、オーディエンスが存在しないとジョブは失敗します。

このモードではomit_null_fieldsは適用されません。Batch Audience APIはnull属性を拒否するため、この設定に関わらずnull属性のセルは常にペイロードから省略されます。

omit_null_fieldsパラメータは、Profileモードでnull値のセルの送信方法を制御します。Audienceモードでは効果がありません — Batch Audience APIはnull属性を拒否するため、この設定に関わらずnull属性のセルは常に省略されます。

オーディエンスのインデックス作成とは、メンバーシップのプッシュ（追加、削除、または置換）をライブオーディエンスに反映するBatchの非同期処理です。すべての/audiences/{update,replace,create}呼び出しは非同期であり、indexing_tokenを返します。すべてのバッチが送信された後、コネクタは収集したすべてのトークンについてGET /audiences/viewをポーリングし、それぞれが終端のindexing_state（APPLIEDまたはREPLACED）に達するか、indexing_timeout_secsが経過するまで続けます。replaceモードでは、バッチ1のreplaceトークンは、追加継続バッチが送信される前に、さらに完了までポーリングされます。これにより、非同期の消去処理が競合して、追加されたばかりの継続バッチを消去することを防ぎます。このバリアはindexing_timeout_secs: 0の場合はスキップされます。

この例では、ネイティブ属性（メール、言語、地域）とカスタム属性（サインアップ日、プロファイルURL、ロイヤルティ階層）を含む顧客プロファイルをエクスポートします。

SELECT
  customer_id          AS identifier__custom_id,
  email                AS batch__email_address,
  marketing_consent    AS batch__email_marketing,
  phone                AS batch__phone_number,
  preferred_language   AS batch__language,
  timezone             AS batch__timezone,
  country_code         AS batch__region,
  CAST(signup_ts AS VARCHAR) AS date__signup_date,
  profile_url          AS url__profile_url,
  loyalty_tier,
  interests            AS string_array__interests,
  preferences_json     AS object__preferences
FROM customer_profiles
WHERE TD_TIME_RANGE(time, TD_DATE_TRUNC('day', TD_SCHEDULER_TIME() - 86400, 'UTC'), NULL)

サンプルクエリ結果

これは1つのPOST /2.11/profiles/mass-updateリクエストを生成します。各行は、identifiers.custom_idブロックと、$email_address、$language、$region、date(signup_date)、url(profile_url)、loyalty_tierワイヤキーを使用するattributesオブジェクトを持つ1つの操作になります。batch__regionの値frはFRに正規化されます。

必要な設定

この例では、Eventsモードで行ごとに1つの購入イベントを記録します。batch__name列は必須です。イベントは各識別子の下にグループ化されます。

SELECT
  customer_id          AS identifier__custom_id,
  'purchase'           AS batch__name,
  CAST(purchased_at AS VARCHAR) AS batch__time,
  product_category     AS category,
  order_amount         AS amount
FROM purchases
WHERE TD_TIME_RANGE(time, TD_DATE_TRUNC('day', TD_SCHEDULER_TIME() - 86400, 'UTC'), NULL)

サンプルクエリ結果

両方の行はcustom_id abc123を共有しているため、events[]配列に2つのエントリを持つ1つの操作にグループ化されます。categoryとamount列はevents[].attributesの下のイベント属性になります。

必要な設定

この例では、プロファイルを更新する代わりに、Treasure AIのセグメントメンバーシップをBatchオーディエンスにプッシュします。マッピング、同期モード、インデックス作成の動作については、Audience固有の注意事項、オーディエンス同期モード、オーディエンスのインデックス作成とポーリングを参照してください。

SELECT
  customer_id       AS identifier__id,
  CAST(signup_ts AS VARCHAR) AS date__signup_date,
  loyalty_tier
FROM customers
WHERE segment = 'vip'

サンプルクエリ結果

これは1つのPOST /2.11/audiences/replaceリクエストを生成します（複数バッチにわたる大規模なセグメントの場合は、1つの/audiences/replaceリクエストの後に1つ以上の/audiences/updateリクエストが続きます）。各行は1つのids[]エントリになります：{"id": "abc123", "attributes": {"date(signup_date)": "2024-01-01T00:00:00Z", "loyalty_tier": "gold"}}。

必要な設定

Scheduled JobsとResult Exportを使用して、出力結果を定期的にBatchに書き込むことができます。

Treasure AIのスケジューラ機能は、高可用性を実現するための定期的なクエリ実行をサポートしています。

Scheduled Jobs と Result Export を使用して、指定したターゲット宛先に出力結果を定期的に書き込むことができます。

Treasure Data のスケジューラー機能は、高可用性を実現するために定期的なクエリ実行をサポートしています。

2 つの仕様が競合するスケジュール仕様を提供する場合、より頻繁に実行するよう要求する仕様が優先され、もう一方のスケジュール仕様は無視されます。

例えば、cron スケジュールが '0 0 1 * 1' の場合、「月の日」の仕様と「週の曜日」が矛盾します。前者の仕様は毎月 1 日の午前 0 時 (00:00) に実行することを要求し、後者の仕様は毎週月曜日の午前 0 時 (00:00) に実行することを要求するためです。後者の仕様が優先されます。

1. Data Workbench > Queries に移動します

2. 新しいクエリを作成するか、既存のクエリを選択します。

3. Schedule の横にある None を選択します。

4. ドロップダウンで、次のスケジュールオプションのいずれかを選択します:

   ドロップダウン値	説明
   Custom cron...	Custom cron... の詳細を参照してください。
   @daily (midnight)	指定されたタイムゾーンで 1 日 1 回午前 0 時 (00:00 am) に実行します。
   @hourly (:00)	毎時 00 分に実行します。
   None	スケジュールなし。

 *    *    *    *    *
 -    -    -    -    -
 |    |    |    |    |
 |    |    |    |    +----- day of week (0 - 6) (Sunday=0)
 |    |    |    +---------- month (1 - 12)
 |    |    +--------------- day of month (1 - 31)
 |    +-------------------- hour (0 - 23)
 +------------------------- min (0 - 59)

次の名前付きエントリを使用できます:

• Day of Week: sun, mon, tue, wed, thu, fri, sat.
• Month: jan, feb, mar, apr, may, jun, jul, aug, sep, oct, nov, dec.

各フィールド間には単一のスペースが必要です。各フィールドの値は、次のもので構成できます:

1. (オプション) Delay execution を有効にすることで、クエリの開始時刻を遅延させることができます。

クエリに名前を付けて保存して実行するか、単にクエリを実行します。クエリが正常に完了すると、クエリ結果は指定された宛先に自動的にエクスポートされます。

(オプション) Delay execution を有効にすることで、クエリの開始時刻を遅延させることができます。

Audience Studio で activation を作成することで、segment データをターゲットプラットフォームに送信することもできます。

1. Audience Studio に移動します。
2. parent segment を選択します。
3. ターゲット segment を開き、右クリックして、Create Activation を選択します。
4. Details パネルで、Activation 名を入力し、前述の Configuration Parameters のセクションに従って activation を設定します。
5. Output Mapping パネルで activation 出力をカスタマイズします。

• Attribute Columns
  • Export All Columns を選択すると、変更を加えずにすべての列をエクスポートできます。
  • + Add Columns を選択して、エクスポート用の特定の列を追加します。Output Column Name には、Source 列名と同じ名前があらかじめ入力されます。Output Column Name を更新できます。+ Add Columns を選択し続けて、activation 出力用の新しい列を追加します。
• String Builder
  • + Add string を選択して、エクスポート用の文字列を作成します。次の値から選択します:
    • String: 任意の値を選択します。テキストを使用してカスタム値を作成します。
    • Timestamp: エクスポートの日時。
    • Segment Id: segment ID 番号。
    • Segment Name: segment 名。
    • Audience Id: parent segment 番号。

6. Schedule を設定します。

• スケジュールを定義する値を選択し、オプションでメール通知を含めます。

7. Create を選択します。

Treasure Workflow内で、データコネクタを使用してデータをエクスポートするように指定できます。

詳細はUsing Workflows to Export Data with the TD Toolbeltを参照してください。

サンプルワークフロー設定

timezone: UTC

_export:
  td:
    database: sample_datasets

+td-result-into-batch:
  td>: queries/batch_export.sql
  result_connection: batch
  result_settings:
    target: Profile
    profile_sync_type: Attributes
    batch_size: 10000
    skip_invalid_records: true
    ignore_partial_errors: true
    omit_null_fields: true

• Batch Profile Mass Update API documentation
• Batch Profile Attributes specification
• Batch Audience API documentation
• Batch Integration Overview