Snowflake Streaming コネクタ構成ガイド

API情報

このコネクタは、以下のベンダーSDKを使用します。

• 現行アクション（HPA）：com.snowflake:snowpipe-streaming バージョン1.8.1
• クラシックアクション：Snowflake Ingest SDKバージョン4.4.2

仕組み

Snowflake Streamingコネクタは、強力なSnowflake Snowpipe Streaming機能を活用して、イベントデータと訪問データをSnowflakeのステージングテーブルにほぼリアルタイムで直接インポートします。これにより、データ遅延を10秒未満に抑え、処理や分析にすぐ利用できるようになります。データセット全体または特定の属性をステージングテーブルに送信でき、ビジネスニーズに応じたカスタマイズ可能なデータ統合を実現します。

Snowpipe Streamingを使用すると、Snowflake内のデータアクセシビリティと分析機能をエンリッチメントしながら、遅延とコストを削減する効率的な経路を構築できます。サードパーティ統合を排除し、Snowpipe Streamingのサーバーレスコンピューティングモデルを使用することで、Snowflake Streamingコネクタは、コスト効率に優れた安全な接続を提供します。

Snowflake Streamingコネクタにより、以下を含むさまざまなユースケースが可能になります。

• ID：クレンジングおよび正規化されたリアルタイムのIDデータを使用して、IDグラフを構築し、アクティブ化します。
• インサイト：豊富で高品質な統合データを使用して、差別化されたインテリジェンスを改善し、アクティブ化します。
• AI：リアルタイムデータレイヤーを使用して、AIイニシアチブへの情報提供とアクティブ化を行います。
• 同意：インサイトとアクティベーションが、常に顧客のプライバシーと構成を尊重するようにします。

アーキテクチャ

このコネクタは、2つのSnowpipe Streamingアーキテクチャをサポートしています。

• 高パフォーマンスアーキテクチャ（HPA）：パイプベースです。HPAは現行のアーキテクチャであり、新しいアクション構成ではデフォルトです。
• クラシック：テーブルベースです。既存の構成ではクラシックアクションが維持され、コネクタ内で**（レガシー）**と表示されます。Snowflakeはクラシックアーキテクチャを非推奨化しています。詳細については、Snowflake：Snowpipe Streamingクラシックアーキテクチャの概要を参照してください。

チャネルとは、Snowflakeターゲットに行を追加するストリーミング接続です。HPAではSnowflake Elastic Channelsを使用します。Snowflakeはパイプごとに1つの暗黙的なチャネルを作成し、複数のライターがそのチャネルに並行して追加できます。Elastic Channelsでは、取り込み順序や重複排除は保証されません。詳細については、Snowflake：Elastic Channelsの制限事項を参照してください。

クラシックアクションでは、Elastic Channelsを使用する代わりに、コネクタアクションごとに個別のチャネルを開きます。クラシックアーキテクチャのチャネルを削除する管理タスクを実行すると、これらのチャネルが削除されます。

HPAでは、行レベルのエラーが非同期で報告されます。Snowflakeがバッチを永続的に受け入れた後、コネクタは送信成功を報告します。後からSnowflakeが行を拒否した場合（無効なタイムスタンプや型の不一致など）でも、Traceとアクションには送信成功と表示され、その行は書き込まれません。拒否された行を確認するには、Snowflakeの行レベルのエラーレポートを確認してください。詳細については、Snowflake：Elastic Channelsの行レベルエラー処理を参照してください。クラシックアクションでは、このような場合に同期的に失敗します。

ステージングテーブル

Tealiumからデータを受信するSnowflakeのステージングテーブルには、選択したコネクタアクションに応じた特定の列が必要です。HPAアクションの場合、ステージングテーブルはパイプのターゲットテーブルであり、コネクタはそのターゲットテーブルから列のリストを読み取ります。

イベントデータ全体を送信または訪問データ全体を送信アクションを使用する場合、ステージングテーブルには、データセットを受信するためのVARIANT列と、タイムスタンプを受信するための列が必要です。タイムスタンプ列は、送信されるタイムスタンプのデータ形式をサポートしている必要があります。

カスタムイベントデータを送信またはカスタム訪問データを送信を使用する場合、ステージングテーブルには各データ型に対応する列が必要です。各データ属性を列に割り当てる必要があります。それぞれの属性がSnowflakeで受け入れられるデータ型であり、列がそのデータ型を受信できるよう適切にフォーマットされていることを確認してください。タイムスタンプをマッピングするには、別のタイムスタンプ列を追加する必要があります。

詳細については、Snowflake：サポートされるJavaデータ型を参照してください。

データ型

以下の表は、Snowflake Streamingコネクタでサポートされるデータ型をまとめたものです。

テーブルおよび列の構成

このコネクタは、以下のテーブルまたは列の構成をサポートしていません。

• 照合順序が構成された列
• TRANSIENTまたはTEMPORARYテーブル
• AUTOINCREMENTまたはIDENTITY列構成を持つテーブル
• NULL以外の列のデフォルト値

許可するIPアドレス

Snowflakeでは、リクエストを受け入れるシステムについて厳格なルールが設けられています。TealiumのIPアドレスをSnowflakeの許可リストに追加してください。

us-west-1およびサーバーサイドプロファイルのリージョンアドレスを許可リストに追加する必要があります。これらのアドレスを許可リストに追加しないと、データを取得しようとしたときにエラーが発生します。Tealiumはコネクタの構成中にus-west-1のIPアドレスを使用します。

ベストプラクティス

クラシックアクションでは、以下のSnowflakeテーブル構成を推奨します。

• テーブルごとに1つのイベントフィード
• テーブルごとに1つのオーディエンス

クラシックアクションでは、複数のフィードまたはオーディエンスから同じテーブルに同時に書き込むと、パフォーマンスエラーが発生する可能性があります。

現行のHPAアクションでは、同じパイプへの複数ライターによる同時書き込みをサポートするSnowflake Elastic Channelsを使用します。テーブル単位の同時実行に関する推奨事項は、HPAアクションには適用されません。

HPAによる配信では、行が重複する可能性があります。Snowflakeが10秒以内にバッチを確認しない場合、コネクタはそのバッチを再試行します。その後、Snowflakeが最初の試行をコミットすると、同じ行が2回書き込まれます。イベントIDを使用して、後続の処理で重複を排除することを推奨します。

Snowflakeネイティブアプリ

Tealiumは、Snowflake Marketplaceで無料のネイティブアプリを提供しています：Tealium Snowflake Connector。

このアプリには、TealiumからSnowflakeにストリーミングされるデータを構成および管理するための補助ユーティリティが含まれています。スキーマ、テーブル、ビューの作成など、初期ランディングテーブルの構成タスクを自動化します。

ネイティブアプリの使用は任意であり、このガイドの構成手順に代わるものではありません。構成を手動で実行するには、以下の手順に従ってください。

現在、Tealium Snowflake Appは訪問ペイロードデータのみを変換します。イベントペイロードデータはサポートされていません。

構成

コネクタマーケットプレイスに移動し、新しいコネクタを追加します。詳細については、コネクタについてを参照してください。

コネクタを追加した後、次の構成を構成します。

• 認証タイプ：接続の認証タイプを選択します。指定できる値は、Private keyまたはPublic keyです。
• ユーザー名：Snowflakeアカウントのユーザー名です。データを受信するSnowflakeテーブルに対するOWNERSHIP権限またはINSERT権限が必要です。
• ロール：（任意）接続後にユーザーがロールを切り替える必要がある場合は、アカウントロールを指定します。Tealium Snowflake Connectorネイティブアプリを使用しており、アプリケーションロールがユーザーに直接付与されている場合は、空白のままにします。
• アカウント識別子：アカウント識別子は、ACCOUNT_IDENTIFIER.snowflakecomputing.com.という形式でアカウントURLを構成するために使用されます。アカウント識別子にアンダースコアが含まれている場合は、代わりにハイフン形式を使用します。
  • アンダースコアを使用したURL：https://acme-marketing_test_account.snowflakecomputing.com
  • ハイフンを使用したURL：https://acme-marketing-test-account.snowflakecomputing.com
• 秘密鍵認証の場合：

  • 秘密鍵：お客様が生成した秘密鍵です。暗号化された秘密鍵と暗号化されていない秘密鍵の両方をサポートします。Snowflakeの秘密鍵を生成する手順については、Snowflake：キーペア認証とキーペアのローテーションを参照してください。

    秘密鍵が暗号化されている場合は、秘密鍵のパスフレーズを指定する必要があります。

  • 秘密鍵のパスフレーズ：暗号化された秘密鍵で使用するパスフレーズです。秘密鍵が暗号化されていない場合は、値を構成しないでください。

  • 秘密鍵を使用するには：

    1. Snowflakeで公開鍵を生成します。詳細については、公開鍵の生成を参照してください。
    2. SnowflakeでALTER USERコマンドを使用して、上記のユーザーに公開鍵を割り当てます。ユーザーを変更できるのは、そのユーザーの所有者、またはSECURITYADMIN以上のロールを持つユーザーのみです。詳細については、Snowflakeユーザーへの公開鍵の割り当てを参照してください。
       ユーザーに公開鍵を正常に割り当てるには、次の点を確認します。
       • Snowflakeのユーザー名を二重引用符（"）で囲んで入力します。例："SNOWFLAKE.USER"。
       • 改行を含めずに公開鍵をコピーして貼り付けます。
    3. クエリを実行し、新しい公開鍵でユーザーを更新します。
• 公開鍵認証の場合：
  • 鍵アルゴリズム：キーペアを生成するアルゴリズムを選択し、生成をクリックします。鍵のフィンガープリントが下に表示されます。新しい鍵を生成するには、再生成をクリックします。RSA-2048、RSA-3072、またはRSA-4096がサポートされています。
  • 公開鍵を使用するには：
    1. 公開鍵をダウンロードをクリックして、下に表示されている公開鍵をダウンロードします。
    2. ファイルから鍵をコピーします。
    3. Snowflakeインスタンスで次のクエリを実行します。
       ALTER USER [your username] SET RSA_PUBLIC_KEY = [your public key from the file].
    4. 完了をクリックして、新しい接続を作成します。
• IcebergテーブルでSnowpipe Streamingを有効化：SnowflakeのターゲットテーブルがマネージドIcebergテーブルである場合は、このオプションを有効にします。この接続レベルのオプションは、現行（HPA）アクションとClassicアクションの両方に適用されます。

Snowflakeは公開鍵と秘密鍵のローテーションをサポートしています。詳細については、キーペアのローテーションの構成を参照してください。

アクション

このコネクタでは、新しい構成向けに現行（HPA）アクションを、後方互換性のためにClassicアクションを提供しています。

現行アクション（HPA）

各現行アクションは、Elastic Channelsを備えたSnowflake Snowpipe Streamingの高性能アーキテクチャ（HPA）を使用し、選択したパイプにデータをストリーミングします。

Classicアクション

次のアクションはClassic（テーブルベース）アーキテクチャを使用しており、非推奨です。既存の構成は引き続き動作します。新しい構成では、現行（HPA）アクションを使用してください。

アクションの選択

新しい構成では、現行（HPA）アクションを使用してください。既存のClassic構成は、Classicアクションで引き続き動作します。2つのアーキテクチャの比較については、アーキテクチャを参照してください。

以降のセクションでは、各アクションでサポートされているパラメータについて説明します。

カスタムイベントデータを送信

パラメータ

イベントデータ全体を送信

パラメータ

カスタムビジターデータを送信

パラメータ

訪問データ全体を送信

パラメータ

ログイベント全体を送信

パラメータ

ログイベントを送信

パラメータ

カスタムイベントデータを送信（レガシー）

このアクションは非推奨です。現在のアクションについては、カスタムイベントデータを送信を参照してください。

パラメータ

イベントデータ全体を送信（レガシー）

このアクションは非推奨です。現在のアクションについては、イベントデータ全体を送信を参照してください。

パラメータ

カスタム訪問データを送信（レガシー）

このアクションは非推奨です。現在のアクションについては、カスタム訪問データを送信を参照してください。

パラメータ

訪問データ全体を送信（レガシー）

このアクションは非推奨です。現在のアクションについては、訪問データ全体を送信を参照してください。

パラメータ

ログイベント全体を送信（レガシー）

このアクションは非推奨です。現在のアクションについては、ログイベント全体を送信を参照してください。

パラメータ

ログイベントを送信（レガシー）

このアクションは非推奨です。現在のアクションについては、ログイベントを送信を参照してください。

パラメータ

ClassicアクションをHPAに移行する

既存のClassicアクションを現在の（HPA）バージョンに移行するには、次の手順を実行します。

1. HPAアクション用のパイプを作成するか、デフォルトの<TABLE_NAME>-STREAMINGパイプを使用するように計画します。権限については、必要なSnowflake権限を参照してください。名前付きパイプを作成するには、パイプを作成するを参照してください。
2. Classicアクションと同じマッピングを使用して、現在の（HPA）アクションをコネクタに追加します。
3. 同じ保存操作で、Classicアクションを無効化または削除します。
4. HPAアクションを介して新しいデータが到着し始めたら、Classicアーキテクチャのチャネルを削除するを使用して、Classicアクションによって作成された残りのチャネルを削除します。

現在の（HPA）アクションとClassicアクションを、同じターゲットテーブルに対して同時に実行しないでください。両方のアクションが重複して実行されている間は、すべてのイベントが2回書き込まれます。最初にHPAアクションをテストする場合は、別のターゲットテーブルを指定してください。

コネクタが送信するのは、アクションの保存後に処理されたイベントのみです。この移行ではデータは再処理されず、バックフィルの仕組みもありません。

Classicアクションを削除する保存内容を公開すると、アクションが停止する前に、コネクタはバッファリング中または再試行中のすべてのバッチをSnowflakeにフラッシュします。これらの行は保持されます。

VARIANT列またはARRAY列を持つテーブルでは、HPAアクションはClassicアクションと同じ値を書き込むため、移行後も既存データが維持されます。

管理タスク

パイプを作成する

HPAアクション用のSnowflakeパイプを作成するには、パイプを作成管理タスクを使用します。このタスクは、現在の（HPA）アクションにあるパイプ名フィールド、またはコネクタのリモート構成から利用できます。

必要なSnowflake権限

コネクタが使用するロールは、作成したパイプを所有し、Snowflakeはその所有ロールの権限を使用してデータを読み込みます。

パイプを作成するには、そのロールに次の権限を付与します。

GRANT USAGE ON DATABASE <database_name> TO ROLE <role>;
GRANT USAGE ON SCHEMA <database_name>.<schema_name> TO ROLE <role>;
GRANT CREATE PIPE ON SCHEMA <database_name>.<schema_name> TO ROLE <role>;
GRANT INSERT ON TABLE <database_name>.<schema_name>.<target_table> TO ROLE <role>;

既存のパイプ（別のロールによって作成されたパイプ）にデータをストリーミングするには、そのロールに次の権限を付与します。

GRANT USAGE ON DATABASE <database_name> TO ROLE <role>;
GRANT USAGE ON SCHEMA <database_name>.<schema_name> TO ROLE <role>;
GRANT OPERATE ON PIPE <database_name>.<schema_name>.<pipe_name> TO ROLE <role>;
GRANT INSERT ON TABLE <database_name>.<schema_name>.<target_table> TO ROLE <role>;

パイプ名フィールドは、ドロップダウンの項目を構成するためにSHOW PIPESを使用します。このコマンドで一覧表示されるのは、ロールが所有しているか、ロールがMONITORまたはOPERATE権限を持つパイプのみです。マッピング用の列リストではDESCRIBE PIPEとDESCRIBE TABLEを使用するため、ロールにはパイプに対するMONITOR権限とターゲットテーブルに対するSELECT権限も必要です。コネクタのロールがパイプを作成した場合は、パイプの所有権によってパイプ側の要件が満たされます。

Snowflakeによって自動的にプロビジョニングされるデフォルトの<TABLE_NAME>-STREAMINGパイプでは、ターゲットテーブルに対するINSERT権限があれば十分です。テーブルの所有権は必要ありません。

パイプを作成する

1. 接続 > イベントコネクタまたは接続 > オーディエンスコネクタに移動します。
2. Snowflakeコネクタを選択します。
3. リモート構成で、パイプを作成をクリックします。
4. 次の情報を入力します。
   • データベース名
     （必須）ターゲットテーブルを含むSnowflakeデータベース。
   • スキーマ名
     （必須）ターゲットテーブルで使用されるスキーマの名前。
   • パイプ名
     （必須）作成するパイプの名前。
   • ターゲットテーブル
     （必須）パイプの書き込み先となるSnowflakeテーブルの名前。
5. 作成をクリックします。

このタスクはSnowflakeで次のステートメントを実行します。このステートメントは、大文字と小文字を区別せず、名前に基づいてターゲットテーブルのすべての列をマッピングします。

CREATE PIPE "<database_name>"."<schema_name>"."<pipe_name>"
AS
COPY INTO "<database_name>"."<schema_name>"."<target_table>"
FROM TABLE(DATA_SOURCE(TYPE=>'STREAMING'))
MATCH_BY_COLUMN_NAME = CASE_INSENSITIVE

このステートメントはCREATE OR REPLACE PIPEではなく、CREATE PIPEです。指定した名前のパイプがスキーマ内にすでに存在する場合、Snowflakeはエラーを返し、管理タスクにそのエラーが表示されます。

クラシックアーキテクチャのチャネルを削除する

この管理タスクは、クラシックアクションにのみ適用されます。現行の（HPA）アクションでは、コネクタごとのチャネルを作成せず、削除タスクも必要としない Snowflake Elastic Channels が使用されます。

Tealium が作成したチャネルを削除するには、チャネルのリストを CSV 形式で指定します。Snowflake インスタンスで SHOW CHANNELS を実行すると、チャネルのリストを取得できます。Tealium が作成したすべてのチャネルを削除するには、リストを空欄のままにします。

削除するチャネルを構成するには、次の手順に従います。

1. Connect > Event Connectors または Connect > Audience Connectors に移動します。
2. Snowflake コネクタを選択します。
3. Remote Configuration で、Drop classic architecture channels をクリックします。
4. 次の情報を入力します。
   • Database Name
     （必須）テーブルが格納されている Snowflake データベースの名前。
   • Schema Name
     （必須）テーブルで使用されているスキーマの名前。
   • Table Name
     （必須）Snowflake テーブルの名前。
   • Channel Name
     削除する Snowflake チャネルの名前を CSV 形式で指定します。Iceberg テーブルでは Channel Name が必須です。チャネル名を指定しない場合は、すべてのチャネルが削除されます。
5. Drop をクリックします。

テーブルのリストが大きく、削除に 30 秒以上かかる場合、チャネルが削除されなかったという誤ったエラーが表示されることがあります。Snowflake インスタンスで SHOW CHANNELS を実行し、チャネルが正常に削除されたことを確認することをお勧めします。