BigQueryデータのインポート

AmplitudeのBigQuery連携を使用すると、BigQueryデータをAmplitudeプロジェクトに直接取り込むことができます。

前提条件

BigQuery からのインポートを開始するには、次の前提条件を満たしてください。

• データをインポートするには、BigQueryにテーブル（または複数のテーブル）が必要です。

• GCS バケットを作成します。 Amplitudeは、この目的専用のものを推奨しています。BigQuery のエクスポートオプションが限られているため、取り込みプロセスではデータをAmplitudeに取り込む前に GCS バケットにオフロードする必要があります。

• 取り込みたいバケットとテーブルに対して権限が付与されたサービスアカウントを作成し、サービスアカウントキーを取得します。 サービスアカウントに次の役割を付与します。

  • BigQuery：
    • プロジェクトレベルのBigQueryジョブユーザー。
    • データにアクセスするために必要なリソースレベルのBigQuery Data Viewer。データが 1 つのテーブルにある場合は、そのテーブル リソース用のサービス アカウント BigQuery Data Viewer を付与してください。 クエリに複数のテーブルまたはデータセットが必要な場合は、クエリ内のすべてのテーブルまたはデータセットに BigQuery Data Viewer を許可してください。
  • Cloud Storage:
    • 取り込みに使用している GCS バケットの Storage Admin。

• 会社のネットワークポリシーによっては、AmplitudeのサーバーがBigQueryインスタンスにアクセスできるようにするために、許可リストに次のIPアドレスを追加する必要がある場合があります。

  • Amplitude米国のIPアドレス:
    • 52.33.3.219
    • 35.162.216.242
    • 52.27.10.221
  • Amplitude EU の IP アドレス:
    • 3.124.22.25
    • 18.157.59.125
    • 18.192.47.195

BigQueryをソースとして追加する

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

1. Amplitudeデータで、[カタログ] をクリックし、[ソース] タブを選択します。
2. Warehouse Sources セクションで、BigQuery をクリックします。
3. サービスアカウントキーを追加し、GCS バケット名を指定します。
4. [次へ] をクリックして接続をテストします。
5. 資格情報を確認したら、[次へ] をクリックしてデータを選択します。 次の設定オプションから選択できます。
   • データのタイプ：Amplitudeは、イベントデータ、ユーザープロパティデータ、またはグループプロパティデータのいずれを取り込んでいるかを知ることができます。
   • インポートの種類：
     • 完全同期：Amplitudeはデータがすでにインポートされているかどうかに関係なく、データセット全体を定期的にインポートします。 これは、行データが時間の経過とともに変化するデータセットの場合に適していますが、どの行が変更されたかを簡単に判断する方法はありません。 それ以外の場合は、時間ベースのインポートがより効率的なオプションです。 このオプションはイベントデータの取り込みをサポートしていません。
     • 時間ベース: Amplitudeは、指定されたタイムスタンプ列によって決定されるように、データの最新の行を定期的に取り込みます。 最初のインポートでは使用可能なすべてのデータが取り込まれ、その後のインポートでは最新のインポート以降のタイムスタンプ付きのすべてのデータが取り込まれます。 このオプションを使用するには、SQLステートメントの出力にタイムスタンプ列を含める必要があります。
     • ミラー同期（早期アクセス）：Amplitudeは、BigQueryの変更履歴を使用して、BigQueryテーブルからINSERT、UPDATE、およびDELETEの操作を、Amplitudeにすでに存在するイベントデータに適用します。このオプションを使用して、Amplitudeを真実のソースと同期させます。 詳細については、「変更データキャプチャ（CDC）を使用したミラー同期」を参照してください。
   • 頻度: 5分から1か月までの範囲で、複数のスケジュールオプションから選択できます。 毎日の同期は、1 日の特定の時間帯に実行できます。 週単位と月単位の同期は、特定の曜日と時間帯に実行できます。
   • SQLクエリ：Amplitudeが適切なデータを取り込むために使用するクエリのコードです。
6. 設定オプションを設定したら、SQL をテスト をクリックして、BigQuery インスタンスからデータがどのように送信されるかを確認します。エラーはすべて「SQLのテスト」ボタンの下に表示されます。
7. エラーがない場合は、「完了」をクリックします。 新しいBigQueryソースを確認する通知が届きます。その後、Amplitude はソース一覧ページにリダイレクトし、そこで新しい BigQuery ソースを確認できます。

このフローに従う際に問題や質問がある場合は、Amplitudeチームまでお問い合わせください。

時間ベースのインポート

Amplitudeの時間ベースのインポートオプションの場合、ベストプラクティスとして単調に増加するタイムスタンプ値を使用してください。 この値は、SQL 設定がクエリを実行しているソーステーブルにレコードがロードされた時点を示す必要があります（「サーバーのアップロード時刻」と呼ばれることもあります）。ウェアハウスインポートツールは、その後のインポートごとに、インポート設定UI内のタイムスタンプ列名入力で参照されている列の最大値を継続的に更新することで、Amplitudeにデータを取り込みます。

変更データキャプチャ（CDC）を使用したミラー同期

Mirror Sync は、BigQuery テーブルを信頼できる情報源（source of truth）として、Amplitude をそのテーブルと同期させ続けます。Amplitude は新しい行を追加するだけでなく、すでに Amplitude 内にあるイベントデータに対して INSERT、UPDATE、および DELETE の操作を適用します。Amplitude は BigQuery のCHANGES()テーブル値関数を使用して変更を検出します。この関数はテーブルの変更履歴を読み取ります。

Mirror 同期はソースの信頼できる情報源をミラーリングするため、Amplitude はこのソースに対するエンリッチメントサービス（ID 解決とユーザーマージ、プロパティとアトリビューションの同期、ロケーション解決、タクソノミー検証）を無効にします。そのため、お客様のデータは BigQuery 内の状態とまったく同じままに保たれます。これは、SnowflakeおよびDatabricks用のMirror Syncと一致します。 ミューテーションがAmplitude全体にどのように適用されるかについての詳細は、「データのミューテーション性」を参照してください。

前提条件

ミラー同期ソースを作成する前に、BigQuery テーブルを準備してください。

• 変更履歴を有効にする（ソーステーブル、またはクエリが読み取るすべてのテーブルに対して）:

  sql

  コピー

  ALTER TABLE your_dataset.your_table
  SET OPTIONS (enable_change_history = TRUE);
  

  変更履歴が有効になっていない場合、接続テストは Change history is not enabled for table ... などのエラーで失敗します。

• タイムトラベル履歴を十分に保持してください。 CHANGES() は、テーブルの タイムトラベルウィンドウ 内でのみ読み取ることができます（デフォルトは 7 日間ですが、2～7 日間で設定可能です）。Amplitudeはデフォルトの7日間を推奨しています。 ミラー同期ソースが接続解除されたままだったり、タイムトラベルウィンドウよりも遅れたりした場合、BigQuery には追いつくための履歴がなくなります。そのため、ソースを再作成する必要があります。

• サポートされているテーブルタイプを使用してください。 Mirror 同期は、標準の BigQuery テーブルから変更履歴を読み取ります。ビュー、マテリアライズドビュー、外部テーブルまたはフェデレーションテーブル、または複数ステートメントトランザクションで作成されたテーブルはサポートされていません。

データをマッピングする

BigQuery の他のインポート戦略と同様に、Mirror Sync は SQL SELECTステートメントを使用してソース列を Amplitude が想定するフィールドにマッピングします。イベント データ型の各ソース列を送信先フィールド名にエイリアス化します。「必須データフィールド」を参照してください。

例えば：

SELECT
    source_row_id AS insert_id,
    user_id       AS user_id,
    event_name    AS event_type,
    event_time_ms AS time
FROM `your_project.your_dataset.your_table`

すべてのイベントにはユーザーIDが含まれている必要があります。ミラー同期では、一意の識別子がない行はドロップされます。つまり、匿名イベントはサポートされていません。

同期する操作を選択する

Mirror Sync のデータ可変性設定では、適用する操作を選択できます。INSERT は常にオンです。BigQuery の変更を Amplitude にどのように伝播させるかに応じて、UPDATE 、DELETE 、または両方を有効にします。

同期をスケジュールする

BigQuery のCHANGES()機能は、同期ごとに最大で 1 日のウィンドウを読み込み、変更データは短い遅延時間 (約 10 分) 後に利用可能になります。 このため、Mirror Sync ソースは一日に一度以上同期する必要があります。Amplitude は、5 分ごとから 12 時間ごとの頻度を提供しています。 ソースが常に追いつくことができるよう、タイムトラベルウィンドウの範囲内で余裕を持った頻度を選択してください。

トラブルシューティング

BigQuery エクスポートステートメントの制限事項

BigQuery のエクスポート文は、クエリ内のメタテーブルを参照できません。 これには、INFORMATION_SCHEMAビュー、システムテーブル、またはワイルドカードテーブルが含まれます。クエリがこれらのメタテーブルのいずれかを参照している場合、Amplitudeは次のようなエラーを報告しますEXPORT DATA statement cannot reference meta tables in the queries。

この制限を回避するには：

• SQLクエリが標準のテーブルとビューのみを参照していることを確認してください。
• INFORMATION_SCHEMAビューへの参照を含めないでください。
• クエリにシステムテーブルを使用することは避けてください。
• インポートクエリでワイルドカードテーブルを使用しないでください。

BigQuery のエクスポートステートメントの制限事項の詳細については、Google の記事「Export statements in GoogleSQ」を参照してください。

必須データフィールド

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

イベント

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

ユーザープロパティ

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

グループプロパティ

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

アクティブなインポートをモニター

インジェスト ジョブ ページでは、BigQuery のインポートをアクティブなインポートとインポート履歴に分割しています。アクティブなインポートはそれぞれ、ジョブの実行中にライブで更新されるカードとして表示されるため、ジョブの詳細ドローアを開かなくても進捗状況を追跡できます。

各カードには以下が表示されます：

• ジョブのステータス。ステータスインジケーターとラベルが表示されます。
• 実行間隔と開始時刻 (Started の後にジョブが開始された時刻が続きます)。
• 取り込まれたレコードと予想されるレコードを比較する進捗バー。 アップロードフェーズ中またはAmplitudeが予想されるレコード数を認識するまで、バーは不確定と表示されます。
• これまでのレコード、およびAmplitudeが取り込んだ予想レコードの割合。
• ジョブが失敗した場合の短いエラーテキスト。

カードは自動的に更新されます。最初の20秒間は2秒ごとで、その後は5秒ごとです。Amplitudeの管理者は、カード上のジョブIDも確認できます。

ジョブの詳細を表示

ジョブ詳細ドロワーは、BigQuery インポートでのみ使用できます。

ジョブ状態

各ジョブは次の4つの状態のいずれかで表示されます。

パイプラインステージ

ドローワーは、ジョブの進捗状況を次の 3 つの段階で視覚化します：

1. アップロード: AmplitudeはBigQueryからデータを抽出し、データをCloud Storageにオフロードします。
2. 処理：Amplitudeは取り込み用にデータを準備し、バッチ処理します。
3. 完了：Amplitudeはバッチをプロジェクトに取り込みます。

各ステージには、保留中、実行中、完了済み、または失敗済みの4つのステータスのいずれかが表示されます。Amplitude はアクティブなステージを強調表示します。ジョブの状態に関係なく、3つのステージすべてが常に表示されます。

取り込み統計情報

完了済みまたは一部完了済みのジョブについては、引き出しに次の情報が表示されます。

• アップロード済み: BigQuery から抽出されたレコードの合計数です。
• 取り込まれました: Amplitudeがプロジェクトに追加したレコード。
• 取り込まれていません: Amplitudeが取り込まなかったレコード。
• 重複: Amplitudeが重複排除したレコード。
• 非準拠：検証に失敗したレコード。
• 取り込まれたレコードの割合：取り込まれたレコードとアップロードされたレコードの比率。

ドロワー内のジョブを移動する

ドロワー内の矢印コントロールを使用してジョブ間を移動したり、ページ番号を直接入力して特定のページにジャンプしたりできます。このドロワーは、ジョブリストのどのページでもTemporalジョブに使用できます。フィルタを変更するか、[空のジョブを表示] オプションを切り替えると、ページネーションがリセットされます。

BigQueryサービスアカウントキーの更新

BigQuery ソースに使用されているサービスアカウントを更新するには、[データ] の [ソース] セクションで既存の BigQuery ソースを選択し、歯車アイコンをクリックします。 このモーダルで、Amplitudeが今後使用する新しいサービスアカウントキーをアップロードします。

BigQuery SQLヘルパー

プロパティフィールド

Amplitudeの多くの機能は、「プロパティ」フィールドに依存しています。このフィールドはプロパティキーとプロパティ値で構成されています。 これらのプロパティフィールドの中で最も一般的なのは、event_properties と user_propertiesです。

Amplitudeがこれらのキーと値のセットを正しく取り込むためには、BigQueryはそれらをJSON文字列ではなく、生のJSONとしてエクスポートする必要があります。BigQuery は JSON を十分にサポートしていませんが、以下ではデータが BigQuery からエクスポートされ、Amplitude にエラーなくインポートされることを確認する方法について説明します。

プロパティフィールドは、STRUCTタイプの列から取得されます。構造体タイプはキーと値の構造を表し、BigQueryから生のJSON形式でエクスポートされます。

ソーステーブルにイベントまたはユーザープロパティが構造体タイプの列に整理されていない場合は、SELECT SQLで作成できます。たとえば、イベントプロパティがフラット化され、それぞれ独自の列に分割されている場合、event_propertiesを次のような構造体に構築できます。

SELECT STRUCT(
    event_property_column_1 AS event_property_name_1,
    event_property_column_2 AS event_property_name_2
) as event_properties
FROM your_table;

RECORDタイプとREPEATEDモードフィールドからイベントまたはユーザープロパティを再構築する

RECORDタイプとREPEATEDモードを持つevent_propertiesまたはuser_propertiesフィールドがある場合は、Amplitudeに取り込む前に有効なフォーマットに変換する必要があります。

この変換を実現するためのアプローチは、次の2つです。

• PARSE_JSONを使用してプロパティをJSONオブジェクトに再構築し、すべての値が適切にフォーマットされていることを確認します。

PARSE_JSON(CONCAT('{',
    (
      SELECT STRING_AGG(
          CONCAT('"', key, '":"',
            COALESCE(
              NULLIF(CODE_POINTS_TO_STRING(
                ARRAY((
                  SELECT * FROM UNNEST((
                    SELECT TO_CODE_POINTS(CAST(value.string_value AS STRING))
                  )) AS code_points
                  WHERE code_points > 31 AND code_points != 34 AND code_points != 92
                ))
              ), ''),
              NULLIF(CAST(value.int_value AS STRING), ''),
              NULLIF(CAST(value.float_value AS STRING), ''),
              NULLIF(CAST(value.double_value AS STRING), '')
            ),
            '"'
          )
      ) FROM UNNEST(event_properties)
    ),
    '}'
  )) AS event_properties

• 個々のキーと値のペアを直接抽出します:

(SELECT value.string_value FROM UNNEST(event_params) WHERE key = 'key1') AS key1,
(SELECT value.string_value FROM UNNEST(event_params) WHERE key = 'key2') AS key2

イベントプロパティを変換した後、Amplitudeが取り込むためのSTRUCTとしてフォーマットします：

STRUCT
(
  action AS action,
  field AS field
) AS event_properties

JSON文字列フィールドからのプロパティ

イベントまたはユーザープロパティを文字列フィールドにJSON形式でフォーマットしている場合でも、選択SQL内のプロパティフィールドをSTRUCTとして再構築する必要があります。BigQuery は、コンテンツが JSON であっても、文字列フィールドを String としてエクスポートします。 Amplitudeのイベント検証はこれらを拒否します。

JSON 文字列フィールドから値を抽出し、プロパティー STRUCT で使用できます。 JSON_EXTRACT_SCALAR関数を使用して、次のように文字列内の値にアクセスします。テーブルの EVENT_PROPERTIES 列に次のような JSON 文字列が含まれている場合:

"{\"record count\":\"50\",\"region\":\"eu-central-1\"}" これは BigQuery UI に {"record count":"50","region":"eu-central-1"}) のように表示されます。この場合、JSON 文字列から値を抽出できます。次のようにします。

SELECT STRUCT(
    JSON_EXTRACT_SCALAR(EVENT_PROPERTIES, "$.record count") AS record_count,
   JSON_EXTRACT_SCALAR(EVENT_PROPERTIES, "$.region") AS region
) as event_properties
FROM your_table;

文字列リテラル

他のデータウェアハウス製品とは異なり、BigQueryは「二重引用符付き文字列」を文字列リテラルとして扱います。列名やテーブル名などの識別子を引用符で囲むことはできません。そうしないと、SQLがBigQueryで実行されません。