ベストプラクティス

Braze クラウドデータ取り込みを使用すると、データウェアハウスやファイルストレージシステムからBrazeへの直接接続を設定して、関連するユーザーデータやカタログデータを同期できます。このデータをBrazeに同期すると、パーソナライゼーション、トリガー、セグメンテーションなどのユースケースに活用できます。

UPDATED_AT 列を理解する

同期が実行されると、Brazeはデータウェアハウスインスタンスに直接接続し、指定されたテーブルからすべての新しいデータを取得し、Brazeダッシュボードの対応するデータを更新します。同期が実行されるたびに、Brazeは更新されたデータを反映します。

例：定期的な同期

CDI同期における UPDATED_AT の使用方法を示すため、ユーザー属性を更新する定期同期の例を考えてみましょう。

• ファイルストレージソース
  • Amazon S3

サポートされるデータタイプ

クラウドデータ取り込みは、次のデータタイプをサポートしています。

• ユーザー属性（以下を含む）：
  • 階層化カスタム属性
  • オブジェクト配列
  • サブスクリプションステータス
• カスタムイベント
• 購入イベント
• カタログ項目
• ユーザー削除リクエスト

データタイプの問題を回避する

CDIを使用して外部ソース（DatabricksやSnowflakeなど）からデータを同期する場合、同期前にソース列が正しいデータタイプを使用していることを確認してください。よくある問題には以下があります。

• 文字列として保存されたタイムスタンプ：日付列がソースデータベースでvarcharやstringではなく、timestampまたはdatetimeタイプを使用していることを確認してください。
• 文字列として保存された数値：同期前にソースクエリで数値列をintegerまたはfloatタイプにキャストしてください。
• 同期間で一貫性のないタイプ：同期間で列のタイプが変更されると、Brazeが新しいデータを拒否する場合があります。ソーススキーマが一貫していることを確認してください。

Brazeダッシュボードでカスタム属性のデータタイプを強制または変更する方法については、カスタムデータの管理を参照してください。

ユーザーデータは、external ID、ユーザーエイリアス、Braze ID、メール、または電話番号で更新できます。ユーザーはexternal ID、ユーザーエイリアス、またはBraze IDで削除できます。

同期されるデータ

同期が実行されるたびに、Brazeは以前に同期されていない行を調べます。このときに、テーブルまたはビューの UPDATED_AT 列がチェックされます。Brazeは、最後に同期された UPDATED_AT の値よりも後の UPDATED_AT を持つ行を選択してインポートします。境界タイムスタンプと完全に一致する行も、同じタイムスタンプで新しい行が実行間に追加された場合、再同期される可能性があります。

データウェアハウスで、次のユーザーと属性をテーブルに追加し、UPDATED_AT の時間をこのデータを追加する時間に設定します。

1
2
3
4
5
6
7
8
{
    "attribute_1":"abcdefg",
    "attribute_2": {
        "attribute_a":"example_value_1",
        "attribute_b":"example_value_1"
    },
    "attribute_3":"2019-07-16T19:20:30+1:00"
}

1
2
3
4
5
6
{
    "attribute_1":"abcdefg",
    "attribute_2":42,
    "attribute_3":"2019-07-16T19:20:30+1:00",
    "attribute_5":"testing"
}

1
2
3
4
5
{
    "attribute_1":"abcdefg",
    "attribute_4":true,
    "attribute_5":"testing_123"
}

次のスケジュールされた同期時に、Brazeは最新の同期済みタイムスタンプよりも後の UPDATED_AT タイムスタンプを持つすべての行を同期します。Brazeはフィールドを更新または追加するため、毎回ユーザープロファイル全体を同期する必要はありません。同期後、ユーザープロファイルは新しい更新を反映します。

定期的な同期、2回目の実行：2022年7月20日午後12時

1
2
3
4
5
6
7
8
{
    "attribute_1":"abcdefg",
    "attribute_2": {
        "attribute_a":"example_value_2",
        "attribute_b":"example_value_2"
    },
    "attribute_3":"2019-07-16T19:20:30+1:00"
}

1
2
3
4
5
6
{
    "attribute_1":"abcdefg",
    "attribute_2":42,
    "attribute_3":"2019-07-16T19:20:30+1:00",
    "attribute_5":"testing"
}

1
2
3
4
5
{
    "attribute_1":"abcdefg",
    "attribute_4":true,
    "attribute_5":"testing_123"
}

1
2
3
4
5
{
    "attribute_1":"abcdefg",
    "attribute_4":false,
    "attribute_5":"testing_123"
}

customer_9012 の新しい行が追加されましたが、その UPDATED_AT の値（2022-07-16 00:25:30）は保存されたタイムスタンプ（2022-07-19 09:07:23）よりも前であるため、同期されません。ただし、customer_5678 の既存の行は保存されたタイムスタンプと等しい UPDATED_AT の値を持つため、包含境界により再同期されます。この動作の詳細については、UPDATED_AT の時間が同期の時間と同じでないことを確認するを参照してください。保存された UPDATED_AT は 2022-07-19 09:07:23 のままです。

定期的な同期、3回目の実行：2022年7月21日午後12時

1
2
3
4
5
6
7
8
{
    "attribute_1":"abcdefg",
    "attribute_2": {
        "attribute_a":"example_value_1",
        "attribute_b":"example_value_1"
    },
    "attribute_3":"2019-07-16T19:20:30+1:00"
}

1
2
3
4
5
6
{
    "attribute_1":"abcdefg",
    "attribute_2":42,
    "attribute_3":"2019-07-16T19:20:30+1:00",
    "attribute_5":"testing"
}

1
2
3
4
5
{
    "attribute_1":"abcdefg",
    "attribute_4":true,
    "attribute_5":"testing_123"
}

1
2
3
4
5
{
    "attribute_1":"xyz",
    "attribute_4":false,
    "attribute_5":"testing_123"
}

1
2
3
4
5
6
7
8
{
    "attribute_1":"abcdefg",
    "attribute_2": {
        "attribute_a":"example_value_2",
        "attribute_b":"example_value_2"
    },
    "attribute_3":"2019-07-20T19:20:30+1:00"
}

この3回目の実行では、customer_1234 の新しい行が追加され、その UPDATED_AT の値（2022-07-21 08:30:00）は保存されたタイムスタンプよりも後です。この新しい行と、customer_5678 の既存の行（保存されたタイムスタンプと等しい UPDATED_AT を持つ）の両方が同期されます。保存された UPDATED_AT は 2022-07-21 08:30:00 に設定されます。

UPDATED_AT 列にUTCタイムスタンプを使用する

夏時間に関する問題を防ぐために、UPDATED_AT 列はUTCにする必要があります。できる限り、CURRENT_DATE() ではなく SYSDATE() など、UTCのみの関数を使用してください。

重複タイムスタンプによる行の再同期を避ける

CDIは、最後に同期された UPDATED_AT タイムスタンプにおける行数を追跡します。CDIが前回の実行以降に同じタイムスタンプで新しい行が追加されたことを検出すると、包含境界（>=）を使用してそのタイムスタンプのすべての行（すでに処理済みのものを含む）を再選択します。それ以外の場合、CDIは排他境界（>）を使用し、最後に同期された値よりも厳密に後の行のみを選択します。

例えば、同期が UPDATED_AT = 2025-04-01 00:00:00 の5行を処理し、後から同じタイムスタンプで6行目が追加された場合、次の同期はカウントの変化を検出し、6行すべてを再同期します。これにより、重複データや不要なデータポイント消費が発生する可能性があります。

これを避けるには：

• VIEW に対して同期を設定する場合、デフォルト値として CURRENT_TIMESTAMP を使用しないでください。使用すると、同期が実行されるたびにすべてのデータが同期されます。これは、UPDATED_AT フィールドの評価結果がクエリの実行時間になるためです。
• 長時間実行されるパイプラインやクエリがソーステーブルにデータを書き込んでいる場合、同期と同時に実行することを避けるか、挿入される各行に同じタイムスタンプを使用することを避けてください。
• トランザクションを使用して、同じタイムスタンプを持つすべての行を書き込みます。
• ユニークで単調増加する UPDATED_AT の値を使用して、処理済みの行が再選択されることを防ぎます。

例：その後の更新を管理する

この例では、最初にデータを同期し、その後の更新では変更されたデータ（差分）のみを更新する一般的なプロセスを説明します。いくつかのユーザーデータを含むテーブル EXAMPLE_DATA があるとします。1日目のテーブルには次の値があります。

このデータをCDIが期待する形式に変換するには、次のクエリを実行します。

1
2
3
4
5
6
7
8
9
10
11
12
SELECT
    CURRENT_TIMESTAMP AS UPDATED_AT,
    EXTERNAL_ID AS EXTERNAL_ID,
    TO_JSON(
        OBJECT_CONSTRUCT(
            'attribute_1', attribute_1,
            'attribute_2', attribute_2,
            'attribute_3', attribute_3,
            'attribute_4', attribute_4
        )
    ) AS PAYLOAD
FROM EXAMPLE_DATA;

このデータはどれもBrazeに同期されていないため、CDIのソーステーブルにすべて追加します。

同期が実行され、Brazeにより「2023-03-16 15:00:00」まで利用可能なすべてのデータを同期したと記録されます。次に、2日目の朝にETLが実行され、ユーザーテーブルの一部のフィールドが更新されます（強調表示）。

ここで、変更された値のみをCDIソーステーブルに追加する必要があります。古い行を更新するのではなく、これらの行を追加できます。そのテーブルは次のようになります。

CDIは新しい行だけを同期するので、次に実行される同期では最後の5行のみが同期されます。

追加のヒント

消費を最小限に抑えるために、新規の属性または更新された属性のみを書き込む

同期が実行されるたびに、Brazeは以前に同期されていない行を調べます。このときに、テーブルまたはビューの UPDATED_AT 列がチェックされます。Brazeは、最後に同期された UPDATED_AT の値よりも後の UPDATED_AT を持つ行を、ユーザープロファイル上の現在の内容と同一かどうかにかかわらず、選択してインポートします。境界タイムスタンプの行も、新しい行がそのタイムスタンプを共有している場合、再同期される可能性があります。そのため、追加または更新する属性のみを同期することをお勧めします。

CDIを使用したデータポイント使用量は、REST APIやSDKなどの他の取り込み方法と同じです。したがって、ソーステーブルに追加するのは新規または更新された属性のみであることを確認するのは、お客様の責任です。

EXTERNAL_ID を PAYLOAD 列から分離する

PAYLOAD オブジェクトには、external IDまたはその他のIDタイプを含めないでください。

属性を削除する

ユーザーのプロファイルから属性を省略する場合は、null に設定できます。属性を変更せずに残す場合は、更新されるまでBrazeに送信しないでください。属性を完全に削除するには、TO_JSON(OBJECT_CONSTRUCT_KEEP_NULL(...)) を使用します。

増分更新を行う

データの増分更新を行うことで、同時更新時の意図しない上書きを防ぐことができます。

この動作を防ぐ最善の方法は、CDI同期のソースデータが各ユーザーの最新の状態のみを反映していること、あるいは特定のユーザーまたはユーザーと属性の組み合わせに対するすべての更新が単一の行に収められていることを保証することです。

別のテーブルからJSON文字列を作成する

各属性を内部的に独自の列に格納している場合は、それらの列をJSON文字列に変換してBrazeとの同期に使用する必要があります。そのために、次のようなクエリを使用できます。

• snowflake
• redshift
• bigquery
• databricks
• microsoft fabric

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
CREATE TABLE "EXAMPLE_USER_DATA"
    (attribute_1 string,
     attribute_2 string,
     attribute_3 number,
     my_user_id string);

SELECT
    CURRENT_TIMESTAMP as UPDATED_AT,
    my_user_id as EXTERNAL_ID,
    TO_JSON(
        OBJECT_CONSTRUCT (
            'attribute_1',
            attribute_1,
            'attribute_2',
            attribute_2,
            'yet_another_attribute',
            attribute_3)
    )as PAYLOAD FROM "EXAMPLE_USER_DATA";

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
CREATE TABLE "EXAMPLE_USER_DATA"
    (attribute_1 string,
     attribute_2 string,
     attribute_3 number,
     my_user_id string);

SELECT
    CURRENT_TIMESTAMP as UPDATED_AT,
    my_user_id as EXTERNAL_ID,
    JSON_SERIALIZE(
        OBJECT (
            'attribute_1',
            attribute_1,
            'attribute_2',
            attribute_2,
            'yet_another_attribute',
            attribute_3)
    ) as PAYLOAD FROM "EXAMPLE_USER_DATA";

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
CREATE OR REPLACE TABLE BRAZE.EXAMPLE_USER_DATA (attribute_1 string,
     attribute_2 STRING,
     attribute_3 NUMERIC,
     my_user_id STRING);

SELECT
    CURRENT_TIMESTAMP as UPDATED_AT,
    my_user_id as EXTERNAL_ID,
    TO_JSON(
      STRUCT(
        'attribute_1' AS attribute_1,
        'attribute_2'AS attribute_2,
        'yet_another_attribute'AS attribute_3
      )
    ) as PAYLOAD
  FROM BRAZE.EXAMPLE_USER_DATA;

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
CREATE OR REPLACE TABLE BRAZE.EXAMPLE_USER_DATA (
    attribute_1 string,
    attribute_2 STRING,
    attribute_3 NUMERIC,
    my_user_id STRING
);

SELECT
    CURRENT_TIMESTAMP as UPDATED_AT,
    my_user_id as EXTERNAL_ID,
    TO_JSON(
      STRUCT(
        attribute_1,
        attribute_2,
        attribute_3
      )
    ) as PAYLOAD
  FROM BRAZE.EXAMPLE_USER_DATA;

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
CREATE TABLE [braze].[users] (
    attribute_1 VARCHAR,
    attribute_2 VARCHAR,
    attribute_3 VARCHAR,
    attribute_4 VARCHAR,
    user_id VARCHAR
)
GO

CREATE VIEW [braze].[user_update_example]
AS SELECT
    user_id as EXTERNAL_ID,
    CURRENT_TIMESTAMP as UPDATED_AT,
    JSON_OBJECT('attribute_1':attribute_1, 'attribute_2':attribute_2, 'attribute_3':attribute_3, 'attribute_4':attribute_4) as PAYLOAD

FROM [braze].[users] ;

UPDATED_AT タイムスタンプを使用する

Brazeは UPDATED_AT タイムスタンプを使用して、正常に同期されたデータを追跡します。CDIは最後に同期されたタイムスタンプにおける行数も追跡します。同じタイムスタンプで新しい行が実行間に追加された場合、CDIはそのタイムスタンプのすべての行を再同期するため、重複データが発生する可能性があります。詳細とヒントについては、重複タイムスタンプによる行の再同期を避けるを参照してください。

テーブル設定

お客様がベストプラクティスやコードスニペットを共有できるように、GitHubリポジトリを公開しています。独自のスニペットを投稿するには、プルリクエストを作成してください。

データのフォーマット

クラウドデータ取り込みのテーブル設定要件とペイロードのフォーマット要件は、クラウドデータ取り込みのテーブル設定に記載されています。

このページを使用して、以下を区別してください。

• ソーステーブルの要件（必須列、識別子列、UPDATED_AT の動作）
• ペイロードの要件（各データタイプで /users/track オブジェクト形式に一致する必要があるフィールド）

データウェアハウスクエリのタイムアウトを避ける

最適なパフォーマンスを実現し、潜在的なエラーを回避するために、クエリは1時間以内に完了することをお勧めします。クエリがこの時間枠を超える場合は、データウェアハウスの設定を見直すことを検討してください。ウェアハウスに割り当てられたリソースを最適化することで、クエリ実行速度を向上させることができます。

製品の制限事項