Snowflake データのインポート

AmplitudeのSnowflake連携を使用すると、SnowflakeデータをAmplitudeプロジェクトに直接取り込むことができます。この連携では、選択したデータタイプに応じて、Snowflakeデータをインポートするための4つの戦略がサポートされています。

制限事項

• 1 つの Snowflake SQL クエリの最大実行時間は 12 時間です。 より多くのコンピューティングリソースをSnowflakeウェアハウスに割り当てることで（たとえば、ウェアハウスのサイズを拡大することによって）、クエリパフォーマンスを最適化し、ランタイムを削減できます。

既存のSnowflake認証情報を再利用する

新しいSnowflakeのインポートまたはエクスポートを作成する場合、Amplitudeを使用すると、以前に保存した認証情報を再入力する代わりに、組織から選択できます。 再利用可能な認証情報には、他のインポートやエクスポートからの接続、および他のプロジェクトで作成された接続が含まれます（権限がある場合）。

既存の認証情報を再利用するには、新しい接続を作成する際の認証情報入力ステップで「既存の認証情報を使用」を選択します。Amplitudeは、お客様の組織から保存されたすべての認証情報をリストアップします。 別のプロジェクトの資格情報を再利用するには、両方のプロジェクトでデータウェアハウス接続を作成するための権限が必要です。

連携を設定する

Snowflakeソースを設定するには、次の手順を実行してください。

1. 接続の設定と検証
2. データを選択
3. インポート戦略を選択する
4. データをマッピングする
5. 同期をスケジュールする

接続の設定と検証

AmplitudeプロジェクトのデータソースとしてSnowflakeを追加するには、以下の手順に従ってください。

1. Amplitude データで、「カタログ」→「ソース」に移動します。

2. 「ウェアハウスソース」セクションで、「Snowflake」をクリックします。

3. 接続したい Snowflake インスタンスに必要な認証情報を入力してください。

   • アカウント：Snowflakeアカウント識別子です。大文字と小文字を区別します。 これは、Snowflake URL の snowflakecomputing.com の前の最初の部分です。 アカウント名に ".snowflakecomputing.com" を含めないでください。
   • データベース：Amplitudeがデータを検索できるデータベースの名前です。
   • Warehouse：AmplitudeがSQLを実行するために使用します。
   • ユーザー名：Amplitudeが認証に使用します。
   • パスワード：Amplitudeが認証に使用します。

   Amplitudeは、Snowflake用のパスワードベースの認証とキーペア認証を提供しています。

- If you want to use password authentication, select Password and enter your password in the Password field. - Key pair authentication (Recommended): If you want to use key pair authentication, select Key pair and then click Generate Key. Then provide the organization and account names in the format ORGNAME-ACCOUNTNAME.

4. （オプション）Snowflake用Stageを使用したS3ストレージ連携の設定（オープンベータ版）：Stageとのストレージ連携を設定することで、AmplitudeがSnowflakeデータにアクセスする際にセキュリティを強化できます。詳細については、SnowflakeのSnowflakeストレージ連携ドキュメントを参照してください。

5. 自動生成されたSQLクエリをコピーし、Snowflakeで実行して、Amplitudeに適切な権限を付与します。

6. クエリを実行したら、［Next］をクリックして接続をテストします。

7. テストが成功したら、もう一度 [次へ] をクリックしてデータ選択ステージに進みます。

データタイプを選択してください

選択したデータタイプによって、構成に使用できる戦略と設定が定義されます。

インポート戦略を選択する

選択したデータタイプに応じて、以下の戦略から選択してください。

どのデータタイプがどのインポート戦略と互換性があるかを理解するには、次の表を参照してください。

データをマッピングする

選択したインポート戦略に応じて、データをSQL文でマッピングしてデータを変換するか（タイムスタンプ、フル同期）、データ選択ツールを使用して列名をAmplitudeプロパティに直接マッピングします。

Eventデータタイプおよび「追加専用」または「タイムスタンプの取り込み」については、オプションで**「ユーザープロパティ**の_同期」または「グループプロパティの同期」_を選択し、イベント内の対応するプロパティを同期します。

同期をスケジュールする

ソースの名前を入力し、同期頻度を設定します。 同期をスケジュールするには、5 分ごとから毎月まで設定できます。 毎日の同期は、1 日の特定の時間帯に実行できます。 週単位と月単位の同期は、特定の曜日と時間帯に実行できます。

ユースケースに最適な連携を選択する

連携戦略を選択する際には、次の点を考慮してください。

• 完全同期: データセット全体を定期的に取り込む必要があり、どの行が変更されたかを追跡できない場合に、このオプションを選択します。この方法は、増分追跡が不可能な小規模なデータセットに最適です。 この方法は、すべてのデータを毎回取り込むために必要なオーバーヘッドがあるため、大規模なデータセットには適していません。

• タイムスタンプのインポート: このオプションは、単調に増加するタイムスタンプ列を使用してデータをインクリメンタルインポートできる場合に選択します。この列は、Snowflakeがレコードをロードするタイミングを示します。これは効率的で、タイムスタンプ付きの新しいデータを追加する場合にうまく機能します。

• 追加のみ同期: このオプションを選択すると、Amplitudeのエンリッチメントサービスを引き続き使用している間にも、SnowflakeのCDC機能によって検出された変更に基づいてデータをインポートできます。この方式は、CDCからの読み取りINSERT操作のみをサポートします。

• ミラー同期：このオプションを選択すると、Snowflake の CDC 機能によって検出された変更に基づいて、Snowflake 内のデータを、INSERT UPDATE、DELETE および操作と直接ミラーリングできます。 この方法では、エンリッチメントサービスを無効にし、Snowflake データのミラーを Amplitude に保持します。UPDATE およびDELETE 操作は、Amplitude 内のデータを変更します。

次の表を使用して、インポート戦略を一目で比較できます。

CDC の前提条件と考慮事項

ミラー同期を使用する場合は、次の点に注意してください。

• 変更追跡を有効にする：ソーステーブルまたはビューの変更追跡を有効にします。Snowflakeのドキュメントにある「ビューと基礎となるテーブルに対する変更追跡の有効化」を参照してください。

• データリテンション設定：DATA_RETENTION_TIME_IN_DAYS1 以上である必要がありますが、Amplitude は少なくとも 7 日間を推奨しています。そうでない場合、変更ベースのインポートは失敗します。 詳細については、Snowflakeのドキュメントにあるタイムトラベルを参照してください。 DATA_RETENTION_TIME_IN_DAYSに設定すると、変更追跡が無効になり、接続が回復不可能になります。0 この問題が発生した場合は、ソースを再作成してください。

• 変更追跡を無効にする：Snowflakeで変更追跡を無効にするか、値よりも長い間Amplitudeソースを切断した場合、DATA_RETENTION_TIME_IN_DAYSAmplitudeは過去の変更を追跡する機能を失います。 この場合は、接続を再作成してください。 イベントの重複を避けるためには、すべてのイベントに ID がinsert_id設定されていることを確認し、7 日以内に接続を再作成してください。

• 一意かつ不変insert_id：予期しない問題が発生した場合にデータの重複を防ぐため、インポートするデータに各行の一意かつ不変な識別子が含まれていることを確認してくださいinsert_id。Amplitudeの重複排除と insert_id の詳細については、「イベント重複排除」を参照してください。

• 複雑なSQLステートメント：データソースが複雑なSQLSELECTステートメント（例えば句付き）として表現されている場合、SnowflakeアカウントにデータソースをラップするJOINビューを作成して、変更ベースのインポート戦略で使用してください。VIEWSnowflakeでビューとCDCを使用する場合の考慮事項については、「ビュー上のストリーム」を参照してください。

• JOIN 付きビュー：Snowflake CDC は効率的ですが、JOIN を含むビューを使用するとパフォーマンスに影響を及ぼす可能性があります。 代わりに、結合されたデータをユーザー プロファイルとして同期することを検討してください。

• テーブルの削除と再作成を回避する: このシナリオでは、Snowflake CDCは変更をキャプチャしないため、同じ名前のテーブルを削除して再作成しないでください。dbtなどのツールを使用して増分モデルを活用することで、テーブルの置換を防ぐことができます。

• スキーマ変更の処理: CDC は、CDC が追跡するテーブルまたはビューにデフォルトNULL値を持つ新しい列を追加することをサポートしています。 Amplitudeは他の種類のスキーマ変更を推奨しません。 Snowflake CDCは、DMLステートメントからの変更のみを反映します。データを論理的に変更するDDLステートメント（デフォルト値を使用して新しい列を追加したり、既存の列を削除したり、列名を変更したりするなど）は、Amplitudeに送信される将来のデータに影響を与えますが、SnowflakeはDDLステートメントによる変更で履歴データを更新しません。 その結果、Amplitudeは履歴データのこれらの更新を反映しません。

• Amplitudeのエンリッチメントサービスが無効化されています：Mirror Syncを使用している場合、AmplitudeはID解決、プロパティとアトリビューションの同期、位置情報の解決などのエンリッチメントサービスを無効化し、真実のソースとの同期を維持します。

• ユーザープライバシーAPI：ユーザープライバシーAPIは、以前に取り込まれたデータを削除し、Amplitudeがユーザーに関する新しい情報を処理することを妨げることはありません。 CDC を使用する場合、ユーザーに関するデータの送信を停止してからユーザープライバシー API を使用してユーザーを削除する必要があります。 これにより、Amplitudeは次の同期時にユーザーを再作成しません。

  Amplitudeのシステムからエンドユーザーに関連付けられたすべてのデータを削除するには、データウェアハウスからユーザーを削除するだけでは不十分です。 このプロセスでは、Amplitudeがユーザーのデータをシステムから確実に削除できるようにするため、ユーザープライバシーAPIリクエストが必要です。

• ミラー同期イベントとミューテーションは不明なユーザーをサポートしていません。 行にはユーザーIDが含まれている必要があります。そうしないと、Amplitudeはイベントをドロップします。大量の匿名イベントが発生している場合、Amplitudeはこのモードの使用を推奨しません。

変更データキャプチャ（CDC）ミラー同期への移行

Amplitudeは、データの送信と変更をテストするために新しいプロジェクトを作成することをお勧めします。 データが正しくマッピングされ、変更されていることを確認したら、メインプロジェクトで次の手順を実行してください。

1. 既存の接続を WHERE time < {cutOffDate} のようなフィルタリング定義を持つように変更します。ここで、time はイベント時刻であり、明日はエポックからのミリ秒単位ですcutOffDate。
2. 前の手順で設定した cutOffDateまで待ちます。
3. 既存のソース接続に新しいデータが流れ込まないことを確認してください。
4. WHERE time >= {cutOffDate}のようなフィルタリング定義を使用して新しいソースを作成します。ここで、time はイベント時刻であり、明日はエポックからのミリ秒単位cutOffDateです。
5. ステップ 1 で変更したソース接続を削除します。

データフィールド

SQLクエリーを作成するときに、データ型の必須フィールドを含めます。これらの表は、各データタイプの必須フィールドとオプションフィールドの概要を示しています。 イベント用にサポートされているその他のフィールドの一覧については、HTTP V2 API のドキュメントを参照してください。また、ユーザープロパティ用にサポートされているその他のフィールドについては、Identify API のドキュメントを参照してください。 これらのリストにない列をすべて event_propertiesまたは user_properties のいずれかに追加します。 それ以外の場合は、Amplitudeはそれらを無視します。

イベント

サポートされているその他のフィールドについては、HTTP V2 API のドキュメントを参照してください。

ユーザープロパティ

サポートされているその他のフィールドについては、Identify API のドキュメントを参照してください。

グループプロパティ

group_propertiesの各グループプロパティは、groups のすべてのグループに適用されます。

グループプロパティを使用するには：

• グループプロパティを設定します。 以下は、Snowflakeグループプロパティのインポートでこれを行う方法の例です。

  SQL

  コピー

  SELECT OBJECT_CONSTRUCT('customerId', account_id)                      AS "groups",          -- must be JSON
        OBJECT_CONSTRUCT('companyName', name, 'customerType', type) AS "group_properties" -- must be JSON
  FROM "AMPLITUDE"."DWH"."ACCOUNTS"
  

• 関連付けられたグループプロパティを持つイベントを送信します。 ユーザーIDとグループが存在する場合、これらはプレースホルダイベントにすることができます。 Snowflakeイベントインポートでの以下の指定：

  SQL

  コピー

  "groups": {"customerId": <account_id>}
  

SQLクエリの例

データ選択手順を簡単にするために、ここでは開始するためのいくつかのサンプルSQLスニペットを示します。

イベントデータの例

SELECT
    EVENT_TYPE_COLUMN AS "event_type",
    EVENT_PROPERTIES_VARIANT_COLUMN AS "event_properties",
    TIME_EPOCH_MS_COLUMN AS "time",
    USER_ID_COLUMN AS "user_id",
    USER_PROPERTIES_VARIANT_COLUMN AS "user_properties"
FROM DATABASE_NAME.SCHEMA_NAME.TABLE_OR_VIEW_NAME

ユーザープロパティの例

SELECT
    USER_ID_COLUMN AS "user_id",
    USER_PROPERTIES_VARIANT_COLUMN AS "user_properties"
FROM DATABASE_NAME.SCHEMA_NAME.TABLE_OR_VIEW_NAME

グループプロパティの例

SELECT
    GROUPS_OBJ AS "groups",
    GROUP_PROPS_OBJ AS "group_properties"
FROM DATABASE_NAME.SCHEMA_NAME.TABLE_OR_VIEW_NAME

共通スニペット

JSON オブジェクトを作成します。

OBJECT_CONSTRUCT('city', CITY, 'state', STATE) as "user_properties"

タイムスタンプ列をミリ秒に変換します。

DATE_PART('EPOCH_MILLISECOND', TIMESTAMP_COLUMN) as "time"

ミリ秒を時間ベースのインポートに必要なTIMESTAMP_NTZ形式に変換します。 この例では、scale に設定された引数を使用して、ミリ秒に変換します。3 詳細については、Snowflakeのドキュメントを参照してください。

TO_TIMESTAMP_NTZ(TIME_COLUMN_IN_MILLIS, 3) as "update_time_column"

タイムゾーン付きのタイムスタンプ列を、TIMESTAMP_NTZ時間ベースのインポートに必要な形式に変換します。

TO_TIMESTAMP_NTZ(CONVERT_TIMEZONE('UTC', TIMESTAMP_TZ_COLUMN)) as "update_time_column"

SQL のトラブルシューティング

以下のセクションでは、インポートコネクタの設定に使用できる SQL クエリの例を示します。

必要なイベントプロパティ

Amplitudeのデータウェアハウスインポートコネクタ用に作成するSnowflake SQLクエリは、AmplitudeのイベントAPIスキーマと一致する特定の列を返す必要があります。 次の例を使用して、クエリを構造化できます。

SELECT
    event_type,             -- String: Name of the event
    user_id,                -- String: Unique identifier for the user
    EXTRACT(EPOCH_MILLISECOND FROM event_timestamp) as time  -- Timestamp in milliseconds
FROM your_events_table

プロパティを含む基本的なイベントクエリ

SELECT
    event_name as event_type,
    user_id,
    EXTRACT(EPOCH_MILLISECOND FROM event_timestamp) as time,
    device_id,
    -- Construct event properties from multiple columns
    OBJECT_CONSTRUCT(
        'page_name', page_name,
        'button_id', button_id,
        'interaction_type', interaction_type,
        'duration_ms', duration_ms
    ) as event_properties,
    -- Construct user properties
    OBJECT_CONSTRUCT(
        'account_type', account_type,
        'subscription_tier', subscription_tier,
        'last_login', TO_VARCHAR(last_login_date)
    ) as user_properties,
    platform,
    app_version
FROM app_events
WHERE event_timestamp >= DATEADD(day, -7, CURRENT_DATE())

Snowflake固有の機能とベストプラクティス

以下に示すのは、Snowflake固有の機能とベストプラクティスの例です。

JSONの操作

-- Combining multiple JSON objects
SELECT
    event_type,
    user_id,
    EXTRACT(EPOCH_MILLISECOND FROM event_timestamp) as time,
    OBJECT_CONSTRUCT(
        'base_properties', base_properties,  -- existing JSON column
        'additional_data', OBJECT_CONSTRUCT(
            'new_field1', value1,
            'new_field2', value2
        )
    ) as event_properties
FROM events
-- Parsing JSON fields
SELECT
    event_type,
    user_id,
    time,
    PARSE_JSON(raw_properties):field_name::string as extracted_value
FROM events

タイムスタンプの処理

-- Converting different timestamp formats
SELECT
    event_type,
    user_id,
    CASE
        WHEN TRY_TO_TIMESTAMP(timestamp_string) IS NOT NULL
            THEN EXTRACT(EPOCH_MILLISECOND FROM TRY_TO_TIMESTAMP(timestamp_string))
        WHEN TRY_TO_TIMESTAMP_NTZ(timestamp_string) IS NOT NULL
            THEN EXTRACT(EPOCH_MILLISECOND FROM TRY_TO_TIMESTAMP_NTZ(timestamp_string))
        ELSE NULL
    END as time
FROM events

データ検証クエリ

-- Validate required fields
SELECT COUNT(*)
FROM (
    YOUR_QUERY_HERE
) t
WHERE event_type IS NULL
    OR user_id IS NULL
    OR time IS NULL;
-- Validate JSON structure
SELECT COUNT(*)
FROM (
    YOUR_QUERY_HERE
) t
WHERE NOT (
    TRY_CAST(event_properties AS OBJECT) IS NOT NULL
    AND TRY_CAST(user_properties AS OBJECT) IS NOT NULL
);
-- Validate timestamp range
SELECT
    MIN(time) as min_time,
    MAX(time) as max_time,
    TIMEADD(millisecond, MIN(time), '1970-01-01'::timestamp) as min_readable_time,
    TIMEADD(millisecond, MAX(time), '1970-01-01'::timestamp) as max_readable_time
FROM (
    YOUR_QUERY_HERE
) t;

パフォーマンス最適化のヒント

次の例を使用して、連携のパフォーマンスを最適化できます。

クラスタリングキーを使用する

ソーステーブルで適切なクラスタリングキーを使用してください。

ALTER TABLE your_events_table CLUSTER BY (event_timestamp, user_id);

マテリアライズドビューを使用する

複雑な変換にはマテリアライズドビューを使用します。

CREATE MATERIALIZED VIEW amplitude_ready_events AS
SELECT
    -- Your transformed columns here
FROM source_events;

WHERE句での日付分割

WHERE event_timestamp >= DATEADD(day, -7, CURRENT_DATE())
  AND event_timestamp < CURRENT_DATE()

マイクロパーティション

SELECT ...
FROM your_table
WHERE TO_DATE(event_timestamp) BETWEEN '2024-01-01' AND '2024-01-31'

トラブルシューティング

1. エラー：SQL compilation error: Invalid identifier INFORMATION_SCHEMA.QUERY_HISTORY_BY_SESSION. Results not generated.

• 原因：この問題は、連携に使用された Snowflake ロールが指定された Snowflake データベースのINFORMATION_SCHEMA にアクセスする権限を有していない場合に発生します。Amplitudeは、顧客のSnowflakeインスタンス上で実行されているクエリのステータスを確認するためにこのスキーマへのアクセスを必要とします。
• 解決策：連携に使用されるロールに、ターゲットの Snowflake データベース内の INFORMATION_SCHEMAにアクセスするために必要な権限があることを確認してください。この問題は、適切なアクセス権を維持することなくロールが最近変更または更新された場合に頻繁に発生します。

1. エラー：JWT token is invalid.

• 原因: このエラーは、特定のユーザーに割り当てられた公開鍵と、Amplitudeがキーペア認証のために生成した秘密鍵との間に不一致がある場合に発生します。
• 解決策: 公開鍵が特定のユーザーに適切に設定されていること、および ORGNAME-ACCOUNTNAME 形式でアカウントを提供していることを確認してください。 アカウントの末尾にアカウントロケータがある場合、公開鍵が正しく設定されている場合でも、キーペアの認証は成功しない可能性があります。 アカウント識別子を正しい形式で取得するには、SnowflakeインスタンスでSELECT CURRENT_ORGANIZATION_NAME() || '-' || CURRENT_ACCOUNT_NAME();を実行してください。

よくある質問

Snowflakeとの連携で問題が発生した場合は、以下のトピックを参照してください。

Snowflakeクエリがタイムアウトになった場合はどうなりますか？

Snowflakeクエリがタイムアウトになると、Amplitudeは指数関数的バックオフ戦略を使用して自動的にクエリを再試行します。 各再試行の待ち時間は前回よりも徐々に長くなり、Snowflakeがリクエストを正常に処理するための時間がより多く確保されます。

Amplitudeは失敗したクエリを何回再試行しますか？

Amplitudeは、インポートジョブを失敗とマークする前に、最大8回の再試行を試みます（最初のクエリを含めて合計9回の試行回数）。 再試行は毎回新規に開始されるため、データの取得には一貫したアプローチが適用されます。

なぜ ABORT_DETACHED_QUERY が FALSE に設定されているのですか？

アカウントレベルでABORT_DETACHED_QUERY = FALSE設定すると、Snowflakeが5分以上実行されるインポートクエリを自動的にキャンセルすることを防げます。この設定がない場合：

• Snowflakeは、長時間実行されているクエリを通知なしにキャンセルします。
• Amplitudeはこれを一時的な障害と解釈し、再試行します。
• このため、イベントの重複やイベント数の増加につながる可能性があります。

この設定によりAmplitudeの再試行メカニズムは変更されませんが、Snowflakeの自動クエリキャンセルによる不要な再試行を防止します。

インポート時にデータの重複を防ぐにはどうすればよいですか？

再試行時のデータの重複や重複インポートを防ぐには:

• 含める insert_id: この固有の識別子により、Amplitudeは重複したイベントを検出して無視できます。
• 適切な再試行設定を行う：ABORT_DETACHED_QUERY = FALSE不要な再試行を防止するように構成します。

insert_idがなければ、Amplitudeはすべての受信イベントを新規イベントとして扱うため、データの重複やイベント数の増加、不正確なアナリティクスにつながる可能性があります。

クエリのタイムアウト時にデータが失われるリスクはありますか?

いいえ。Amplitudeのインポートジョブがタイムアウトした場合にデータが失われるリスクはありません。 その理由は次のとおりです。

• 読み取り専用操作：AmplitudeはSnowflakeインスタンスからのデータのみを読み取ります。 テーブルを変更・削除したり、テーブルに書き込んだりすることはありません。
• ソースデータの保護：タイムアウトが発生するのはデータ転送中であり、ソースデータに影響を与える可能性のある操作中ではありません。
• 自動再試行：Amplitudeは失敗したインポートを最大8回自動的に再試行するため、データにインポートを成功させる機会を複数回提供します。

インポートジョブの結果に関係なく、Snowflakeデータは安全に保たれ、変更されることはありません。