Content Analytics ヘルプサイト
⏩

タグイベントデータをBigQueryに取り込む

ContentAnalyticsで取得しているユーザーのコンテンツ単位タグイベントデータは、データエクスポート機能を使うことで、CDPやMAツールなどに取り込んでいただくことが可能です。

本ページでは、BigQueryに取り込む手順を共有します！

設定の流れ

設定の流れは、大きく7STEPあります。順番に解説しますね。

image
  1. サービスアカウント作成
  2. お客様環境のCloud Storageにオブジェクト作成権限を付与
  3. コンテンツ名登録
  4. データエクスポート（AM2:00-6:00の間で自動実行）
  5. エクスポートファイルをお客様環境のCloud Storageにコピー
  6. BigQueryにデータ転送
  7. 分析者のためにデータを事前加工

STEP1: サービスアカウント作成

ContentAnalyticsの 組織管理 > データエクスポート から、サービスアカウントを作成します。

▼サービスアカウント作成方法はこちら

  • サービスアカウント作成後にダウンロードするキーファイルは、外部に流出しないように大切に管理してください。
  • サービスアカウントが作成されたら、アカウントの情報が画面に表示されます。 後続の手順で使うため、サービスアカウントのメールアドレスと、GCSバケット名をコピーして控えておきましょう。

次に、タグイベントデータをエクスポートしたいプロジェクトを指定してください。

image

これで、ContentAnalytics側の設定は完了です。

STEP2: お客様環境のCloud Storageにオブジェクト作成権限を付与

今回の手順では、BigQueryへの取り込みに「BigQuery Data Transfer」を使用するため、お客様環境のCloud Storageを使用します。

まず、Cloud Storageにバケットを作成しましょう。 今回の手順では、バケット名を「ca-tagdata-1234abcd」とします。（GCSバケット名は全世界でユニークな名称にする必要があります）

image

作成したバケットに、Storage管理者権限を追加します。 STEP1で作成したサービスアカウントのメールアドレスを追加しましょう。

入力項目
入力値
新しいプリンシパル
STEP1で作成したサービスアカウントのメールアドレス
ロール
Storage オブジェクト管理者 （再取り込み時をしたい場合に、上書き(削除)するため、管理者としている）

image

STEP3: 分析対象のコンテンツ名を登録

ContentAnalyticsで認識済みのコンテンツは、コンテンツを識別するIDが採番されていますが、ランダムな32桁の文字列のため、視認性がよくありません。

分析者がコンテンツを把握しやすくするために、分析対象のコンテンツには、Content Analytics上で名称を登録しておくことを推奨します。

▼コンテンツ名登録方法はこちら

STEP4: データエクスポート（AM2:00-6:00の間で自動実行）

タグイベントのデータ、コンテンツマスタのデータをCloud Storageにエクスポートします。

STEP5: エクスポートファイルをお客様環境のCloud Storageにコピー

①　Google CloudのSDKを使用し、エクスポートされたファイルを、お客様のCloud Storageにコピーします。

SDKのインストールは以下のページを参考にしてください。

②　SDKのインストールが完了したら、実際にファイルをコピーしてみましょう。

まず、ターミナル（コマンドプロンプトなど）を開き、以下のコマンドでサービスアカウントを認証します。下記の青字を書き換えてご利用ください。

gcloud auth activate-service-account {STEP1で作成したサービスアカウントのメールアドレス} --key-file {サービスアカウント作成時にダウンロードしたキーファイルのパス}

以下のメッセージが表示されると、認証成功です。

Activated service account credentials for: [{STEP1で作成したサービスアカウントのメールアドレス}]

③　次に、ファイルコピーのコマンドを実行します。 タグイベントデータは膨大な量になる可能性があるため、対象日付を指定する方法を共有します。

gsutil -m cp -r gs://{STEP1で作成されたGCSバケット名}/projects/{対象プロジェクトID}/{対象日付} gs://{お客様環境のGCSバケット名}

※ エクスポート対象プロジェクトのGCSバケット名やプロジェクトIDは下記からもご確認いただけます。

image
image

以下のメッセージが表示されると、コピー成功です。

Operation completed over xxx objects/xxx KiB.

※権限エラー（AccessDeniedException 403）を表示された場合は、権限が設定されているかどうかご確認ください。

ちなみに、対象日付を絞らずに連携する場合は、以下のコマンドを実行してください。

gsutil -m cp -r  gs://{STEP1で作成されたGCSバケット名}/projects/{対象プロジェクトID}  gs://{お客様環境のGCSバケット名}

コマンドが成功したら、お客様環境のCloud Storageにファイルが存在するか確認しましょう。

STEP6: BigQueryにデータ転送

Cloud Storageから取り込むBigQueryのデータセット・テーブルを準備しましょう。

まずは、データセットを作成します。 今回の手順では、データセット名を「ca_tagdata」とします。 下記SQLの「{project_id}」は、お客様のGCPのプロジェクトIDに書き換えてください。

CREATE SCHEMA `{project_id}.ca_tagdata`;

以降の手順でSQLのサンプルを載せていますが、プロジェクトIDを指定していません。 GCPコンソール上の選択プロジェクトが追加対象のプロジェクトであるかご確認の上、SQLを実行してください。もしくは、各SQLにプロジェクトIDを追加していただいても構いません。

次に、データセット内に転送用のテーブルを3つ作成します。

CREATE TABLE ca_tagdata.tagdata (
    event_id STRING NOT NULL OPTIONS(description="タグイベントID"),
    timestamp TIMESTAMP NOT NULL OPTIONS(description="イベント発火日時"),
    project_id STRING NOT NULL OPTIONS(description="プロジェクトID"),
    session_id STRING NOT NULL OPTIONS(description="セッションID"),
    view_id STRING NOT NULL OPTIONS(description="ページビューID"),
    client_id STRING OPTIONS(description="クライアントID"),
    user_id STRING OPTIONS(description="ユーザID"),
    normalized_url STRING OPTIONS(description="URL(正規化済)"),
    original_url STRING OPTIONS(description="URL(正規化前)"),
    title STRING OPTIONS(description="タイトル"),
    device STRING OPTIONS(description="デバイス"),
    event_type STRING NOT NULL OPTIONS(description="イベントタイプ"),
    content_id STRING OPTIONS(description="コンテンツID"),
    second INT64 OPTIONS(description="閲覧秒数"),
    clicks_count INT64 OPTIONS(description="クリック数"),
    dimension1 STRING OPTIONS(description="カスタムディメンション1"),
    dimension2 STRING OPTIONS(description="カスタムディメンション2"),
    dimension3 STRING OPTIONS(description="カスタムディメンション3"),
    dimension4 STRING OPTIONS(description="カスタムディメンション4"),
    dimension5 STRING OPTIONS(description="カスタムディメンション5"),
    dimension6 STRING OPTIONS(description="カスタムディメンション6"),
    dimension7 STRING OPTIONS(description="カスタムディメンション7"),
    dimension8 STRING OPTIONS(description="カスタムディメンション8"),
    dimension9 STRING OPTIONS(description="カスタムディメンション9"),
    dimension10 STRING OPTIONS(description="カスタムディメンション10"),
    dimension11 STRING OPTIONS(description="カスタムディメンション11"),
    dimension12 STRING OPTIONS(description="カスタムディメンション12"),
    dimension13 STRING OPTIONS(description="カスタムディメンション13"),
    dimension14 STRING OPTIONS(description="カスタムディメンション14"),
    dimension15 STRING OPTIONS(description="カスタムディメンション15"),
    dimension16 STRING OPTIONS(description="カスタムディメンション16"),
    dimension17 STRING OPTIONS(description="カスタムディメンション17"),
    dimension18 STRING OPTIONS(description="カスタムディメンション18"),
    dimension19 STRING OPTIONS(description="カスタムディメンション19"),
    dimension20 STRING OPTIONS(description="カスタムディメンション20"),
    event_category STRING OPTIONS(description="イベントカテゴリ"),
    event_action STRING OPTIONS(description="イベントアクション"),
    event_label STRING OPTIONS(description="イベントラベル"),
    event_value FLOAT64 OPTIONS(description="イベント値"),
    created_at TIMESTAMP NOT NULL OPTIONS(description="レコード登録日時"),
    updated_at TIMESTAMP NOT NULL OPTIONS(description="レコード更新日時"),
    ssc_id STRING OPTIONS(description="サーバサイドCookieで生成したクライアントID"),
) PARTITION BY DATE_TRUNC(timestamp, DAY) CLUSTER BY event_type
;

CREATE TABLE ca_tagdata.content (
    content_id STRING NOT NULL OPTIONS(description="コンテンツID"),
    project_id INT64 NOT NULL OPTIONS(description="プロジェクトID"),
    url STRING NOT NULL OPTIONS(description="URL"),
    name STRING NOT NULL OPTIONS(description="コンテンツ名"),
    created_at TIMESTAMP NOT NULL OPTIONS(description="レコード登録日時"),
    updated_at TIMESTAMP NOT NULL OPTIONS(description="レコード更新日時"),
);

CREATE TABLE ca_tagdata.content_tags (
    content_id STRING NOT NULL OPTIONS(description="コンテンツID"),
    project_id INT64 NOT NULL OPTIONS(description="プロジェクトID"),
    name STRING NOT NULL OPTIONS(description="タグ名"),
);

次に、Cloud StorageのファイルをBigQueryに取り込むための「データ転送」を作成します。

image

まずは、タグイベントデータ用の設定を作成します。

入力項目
入力値
ソースタイプ
Google Cloud Storage
転送構成名
ca_tagdata（管理しやすい名前を入力）
スケジュールオプション
-
　繰り返しの頻度
オンデマンド（手動実行）
転送先の設定
-
　データセット
ca_tagdata
データソースの詳細
-
　Destination table
tagdata
　Cloud Storage URI
{お客様環境のGCSバケット名}/*/tagdata_*.csv.gz
　Write preference
APPEND（テーブルにデータを追加する）
　Delete source files after transfer
チェックする（データ追加後はCloud Storageからファイルを削除する）
　File format
CSV
Transfer Options
（以下の項目を変更する）
　Ignore unknown values
チェックする
　Field delimiter
,
　Quote character　
“
　Header rows to skip
1（1行目はヘッダーのためスキップする）
　Allow quoted newlines
チェックする
　Allow jagged rows
チェックする

データ転送情報を入力し、保存すると、データ転送実行ユーザの選択が求められます。 ここでは、サービスアカウントではなく、お客様自身のユーザを選択します。

image

次に、コンテンツマスタ用の設定を作成します。

入力項目
入力値
ソースタイプ
Google Cloud Storage
転送構成名
ca_content（管理しやすい名前を入力）
スケジュールオプション
-
　繰り返しの頻度
オンデマンド（手動実行）
転送先の設定
-
　データセット
ca_tagdata
データソースの詳細
-
　Destination table
content
　Cloud Storage URI
{お客様環境のGCSバケット名}/*/{yyyymmdd}/contents_*.csv.gz （最新のファイル1件でOK、yyyymmddの部分を書き換えてください）
　Write preference
MIRROR（テーブルのデータを置き換える）
　Delete source files after transfer
チェックする（データ追加後はCloud Storageからファイルを削除する）
　File format
CSV
Transfer Options
（以下の項目を変更する）
　Ignore unknown values
チェックする
　Field delimiter
　Quote character
“
　Header rows to skip
1（1行目はヘッダーのためスキップする）
　Allow quoted newlines
チェックする
　Allow jagged rows
チェックする

次に、コンテンツタグデータ用の設定を作成します。

入力項目
入力値
ソースタイプ
Google Cloud Storage
転送構成名
ca_content_tags（管理しやすい名前を入力）
スケジュールオプション
-
　繰り返しの頻度
オンデマンド（手動実行）
転送先の設定
-
　データセット
ca_tagdata
データソースの詳細
-
　Destination table
content_tags
　Cloud Storage URI
{お客様環境のGCSバケット名}/*/{yyyymmdd}/content_tags_*.csv.gz （最新のファイル1件でOK、yyyymmddの部分を書き換えてください）
　Write preference
MIRROR（テーブルのデータを置き換える）
　Delete source files after transfer
チェックする（データ追加後はCloud Storageからファイルを削除する）
　File format
CSV
Transfer Options
（以下の項目を変更する）
　Ignore unknown values
チェックする
　Field delimiter
,
　Quote character
“
　Header rows to skip
1（1行目はヘッダーのためスキップする）
　Allow quoted newlines
チェックする
　Allow jagged rows
チェックする

データ転送設定が作成できたら、転送を実行してみましょう。

image

image

しばらくたつと、転送が完了します。

image

実行ログを見て、成功したか確認しましょう。

image

データ転送の実行ログに以下のメッセージが表示されると、成功です。

Summary: succeeded xxx jobs, failed 0 jobs.

次に、BigQueryにデータが登録されたか、確認してみましょう。

image

以上で、BigQueryにデータを取り込む手順は完了しました。 このまま分析に進んでもいいですが、分析者のために次のSTEPで事前データ加工しておくことを推奨します。

STEP7: 分析者のためにデータを事前加工

tagdataには、以下3つのイベントが1つのテーブルに格納されています。

　1. ページビューイベント 　2. コンテンツイベント 　3. カスタムイベント

また、データ転送のコストを削減するため、ページビュー単位のデータはコンテンツイベントには保持していません。（例: カスタムディメンション、ユーザIDなど）

分析者の利便性を考えると、以下のように事前データ加工しておくことを、推奨します。

image

データセット内にイベント単位のテーブルを作成します。

CREATE TABLE ca_tagdata.page_view (
    event_id STRING NOT NULL OPTIONS(description="タグイベントID"),
    access_date DATE NOT NULL OPTIONS(description="イベント発火日付"),
    timestamp TIMESTAMP NOT NULL OPTIONS(description="イベント発火日時"),
    project_id STRING NOT NULL OPTIONS(description="プロジェクトID"),
    session_id STRING NOT NULL OPTIONS(description="セッションID"),
    view_id STRING NOT NULL OPTIONS(description="ページビューID"),
    client_id STRING OPTIONS(description="クライアントID"),
    ssc_id STRING OPTIONS(description="サーバサイドCookieで生成したクライアントID"),
    user_id STRING OPTIONS(description="ユーザID"),
    normalized_url STRING OPTIONS(description="URL(正規化済)"),
    original_url STRING OPTIONS(description="URL(正規化前)"),
    title STRING OPTIONS(description="タイトル"),
    device STRING OPTIONS(description="デバイス"),
    dimension1 STRING OPTIONS(description="カスタムディメンション1"),
    dimension2 STRING OPTIONS(description="カスタムディメンション2"),
    dimension3 STRING OPTIONS(description="カスタムディメンション3"),
    dimension4 STRING OPTIONS(description="カスタムディメンション4"),
    dimension5 STRING OPTIONS(description="カスタムディメンション5"),
    dimension6 STRING OPTIONS(description="カスタムディメンション6"),
    dimension7 STRING OPTIONS(description="カスタムディメンション7"),
    dimension8 STRING OPTIONS(description="カスタムディメンション8"),
    dimension9 STRING OPTIONS(description="カスタムディメンション9"),
    dimension10 STRING OPTIONS(description="カスタムディメンション10"),
    dimension11 STRING OPTIONS(description="カスタムディメンション11"),
    dimension12 STRING OPTIONS(description="カスタムディメンション12"),
    dimension13 STRING OPTIONS(description="カスタムディメンション13"),
    dimension14 STRING OPTIONS(description="カスタムディメンション14"),
    dimension15 STRING OPTIONS(description="カスタムディメンション15"),
    dimension16 STRING OPTIONS(description="カスタムディメンション16"),
    dimension17 STRING OPTIONS(description="カスタムディメンション17"),
    dimension18 STRING OPTIONS(description="カスタムディメンション18"),
    dimension19 STRING OPTIONS(description="カスタムディメンション19"),
    dimension20 STRING OPTIONS(description="カスタムディメンション20"),
    created_at TIMESTAMP NOT NULL OPTIONS(description="レコード登録日時"),
    updated_at TIMESTAMP NOT NULL OPTIONS(description="レコード更新日時"),
) PARTITION BY access_date
;

CREATE TABLE ca_tagdata.content_event (
    event_id STRING NOT NULL OPTIONS(description="タグイベントID"),
    access_date DATE NOT NULL OPTIONS(description="イベント発火日付"),
    timestamp TIMESTAMP NOT NULL OPTIONS(description="イベント発火日時"),
    project_id STRING NOT NULL OPTIONS(description="プロジェクトID"),
    session_id STRING NOT NULL OPTIONS(description="セッションID"),
    view_id STRING NOT NULL OPTIONS(description="ページビューID"),
    client_id STRING OPTIONS(description="クライアントID"),
    ssc_id STRING OPTIONS(description="サーバサイドCookieで生成したクライアントID"),
    user_id STRING OPTIONS(description="ユーザID"),
    normalized_url STRING OPTIONS(description="URL(正規化済)"),
    original_url STRING OPTIONS(description="URL(正規化前)"),
    title STRING OPTIONS(description="タイトル"),
    device STRING OPTIONS(description="デバイス"),
    content_id STRING OPTIONS(description="表示用コンテンツID"),
    second INT64 OPTIONS(description="閲覧秒数"),
    clicks_count INT64 OPTIONS(description="クリック数"),
    dimension1 STRING OPTIONS(description="カスタムディメンション1"),
    dimension2 STRING OPTIONS(description="カスタムディメンション2"),
    dimension3 STRING OPTIONS(description="カスタムディメンション3"),
    dimension4 STRING OPTIONS(description="カスタムディメンション4"),
    dimension5 STRING OPTIONS(description="カスタムディメンション5"),
    dimension6 STRING OPTIONS(description="カスタムディメンション6"),
    dimension7 STRING OPTIONS(description="カスタムディメンション7"),
    dimension8 STRING OPTIONS(description="カスタムディメンション8"),
    dimension9 STRING OPTIONS(description="カスタムディメンション9"),
    dimension10 STRING OPTIONS(description="カスタムディメンション10"),
    dimension11 STRING OPTIONS(description="カスタムディメンション11"),
    dimension12 STRING OPTIONS(description="カスタムディメンション12"),
    dimension13 STRING OPTIONS(description="カスタムディメンション13"),
    dimension14 STRING OPTIONS(description="カスタムディメンション14"),
    dimension15 STRING OPTIONS(description="カスタムディメンション15"),
    dimension16 STRING OPTIONS(description="カスタムディメンション16"),
    dimension17 STRING OPTIONS(description="カスタムディメンション17"),
    dimension18 STRING OPTIONS(description="カスタムディメンション18"),
    dimension19 STRING OPTIONS(description="カスタムディメンション19"),
    dimension20 STRING OPTIONS(description="カスタムディメンション20"),
    created_at TIMESTAMP NOT NULL OPTIONS(description="レコード登録日時"),
    updated_at TIMESTAMP NOT NULL OPTIONS(description="レコード更新日時"),
) PARTITION BY access_date
;

CREATE TABLE ca_tagdata.custom_event (
    event_id STRING NOT NULL OPTIONS(description="タグイベントID"),
    access_date DATE NOT NULL OPTIONS(description="イベント発火日付"),
    timestamp TIMESTAMP NOT NULL OPTIONS(description="イベント発火日時"),
    project_id STRING NOT NULL OPTIONS(description="プロジェクトID"),
    session_id STRING NOT NULL OPTIONS(description="セッションID"),
    view_id STRING NOT NULL OPTIONS(description="ページビューID"),
    client_id STRING OPTIONS(description="クライアントID"),
    ssc_id STRING OPTIONS(description="サーバサイドCookieで生成したクライアントID"),
    user_id STRING OPTIONS(description="ユーザID"),
    normalized_url STRING OPTIONS(description="URL(正規化済)"),
    original_url STRING OPTIONS(description="URL(正規化前)"),
    title STRING OPTIONS(description="タイトル"),
    device STRING OPTIONS(description="デバイス"),
    dimension1 STRING OPTIONS(description="カスタムディメンション1"),
    dimension2 STRING OPTIONS(description="カスタムディメンション2"),
    dimension3 STRING OPTIONS(description="カスタムディメンション3"),
    dimension4 STRING OPTIONS(description="カスタムディメンション4"),
    dimension5 STRING OPTIONS(description="カスタムディメンション5"),
    dimension6 STRING OPTIONS(description="カスタムディメンション6"),
    dimension7 STRING OPTIONS(description="カスタムディメンション7"),
    dimension8 STRING OPTIONS(description="カスタムディメンション8"),
    dimension9 STRING OPTIONS(description="カスタムディメンション9"),
    dimension10 STRING OPTIONS(description="カスタムディメンション10"),
    dimension11 STRING OPTIONS(description="カスタムディメンション11"),
    dimension12 STRING OPTIONS(description="カスタムディメンション12"),
    dimension13 STRING OPTIONS(description="カスタムディメンション13"),
    dimension14 STRING OPTIONS(description="カスタムディメンション14"),
    dimension15 STRING OPTIONS(description="カスタムディメンション15"),
    dimension16 STRING OPTIONS(description="カスタムディメンション16"),
    dimension17 STRING OPTIONS(description="カスタムディメンション17"),
    dimension18 STRING OPTIONS(description="カスタムディメンション18"),
    dimension19 STRING OPTIONS(description="カスタムディメンション19"),
    dimension20 STRING OPTIONS(description="カスタムディメンション20"),
    event_category STRING OPTIONS(description="イベントカテゴリ"),
    event_action STRING OPTIONS(description="イベントアクション"),
    event_label STRING OPTIONS(description="イベントラベル"),
    event_value FLOAT64 OPTIONS(description="イベント値"),
    created_at TIMESTAMP NOT NULL OPTIONS(description="レコード登録日時"),
    updated_at TIMESTAMP NOT NULL OPTIONS(description="レコード更新日時"),
) PARTITION BY access_date
;

テーブルの作成が完了したら、各テーブルにデータをコピーします。

MERGE ca_tagdata.page_view T
USING (
  SELECT * except(row_num) FROM (
    SELECT
      event_id,
      DATE(timestamp, 'Asia/Tokyo') AS access_date,
      timestamp,
      project_id,
      session_id,
      view_id,
      client_id,
      ssc_id,
      user_id,
      normalized_url,
      original_url,
      title,
      device,
      dimension1,
      dimension2,
      dimension3,
      dimension4,
      dimension5,
      dimension6,
      dimension7,
      dimension8,
      dimension9,
      dimension10,
      dimension11,
      dimension12,
      dimension13,
      dimension14,
      dimension15,
      dimension16,
      dimension17,
      dimension18,
      dimension19,
      dimension20,
      created_at,
      updated_at,
      -- 更新データが連携されるパターンがあるので、最終更新データのみ取り込む用にROW_NUMBERを振る
      -- （カスタムディメンションのスコープ"セッション"のデータがセットされた、など）
      ROW_NUMBER() OVER (
        PARTITION BY event_id
        ORDER BY updated_at DESC
      ) AS row_num,
    FROM ca_tagdata.tagdata WHERE event_type = 'page_view'
  )
  WHERE row_num = 1
) S ON
  T.event_id = S.event_id
WHEN MATCHED THEN UPDATE SET
  ssc_id = S.ssc_id,
  user_id = S.user_id,
  dimension1 = S.dimension1,
  dimension2 = S.dimension2,
  dimension3 = S.dimension3,
  dimension4 = S.dimension4,
  dimension5 = S.dimension5,
  dimension6 = S.dimension6,
  dimension7 = S.dimension7,
  dimension8 = S.dimension8,
  dimension9 = S.dimension9,
  dimension10 = S.dimension10,
  dimension11 = S.dimension11,
  dimension12 = S.dimension12,
  dimension13 = S.dimension13,
  dimension14 = S.dimension14,
  dimension15 = S.dimension15,
  dimension16 = S.dimension16,
  dimension17 = S.dimension17,
  dimension18 = S.dimension18,
  dimension19 = S.dimension19,
  dimension20 = S.dimension20,
  updated_at = S.updated_at
WHEN NOT MATCHED THEN INSERT ROW;


MERGE ca_tagdata.content_event T
USING (
  WITH page_views AS (
    SELECT
      project_id,
      session_id,
      view_id,
      client_id,
      ssc_id,
      user_id,
      normalized_url,
      original_url,
      title,
      device,
      dimension1,
      dimension2,
      dimension3,
      dimension4,
      dimension5,
      dimension6,
      dimension7,
      dimension8,
      dimension9,
      dimension10,
      dimension11,
      dimension12,
      dimension13,
      dimension14,
      dimension15,
      dimension16,
      dimension17,
      dimension18,
      dimension19,
      dimension20,
    FROM ca_tagdata.page_view
  )
  SELECT * except(row_num) FROM (
    SELECT
      t.event_id,
      DATE(t.timestamp, 'Asia/Tokyo') AS access_date,
      t.timestamp,
      t.project_id,
      t.session_id,
      t.view_id,
      pv.client_id,
      pv.ssc_id,
      pv.user_id,
      pv.normalized_url,
      pv.original_url,
      pv.title,
      pv.device,
      t.content_id,
      t.second,
      t.clicks_count,
      pv.dimension1,
      pv.dimension2,
      pv.dimension3,
      pv.dimension4,
      pv.dimension5,
      pv.dimension6,
      pv.dimension7,
      pv.dimension8,
      pv.dimension9,
      pv.dimension10,
      pv.dimension11,
      pv.dimension12,
      pv.dimension13,
      pv.dimension14,
      pv.dimension15,
      pv.dimension16,
      pv.dimension17,
      pv.dimension18,
      pv.dimension19,
      pv.dimension20,
      t.created_at,
      t.updated_at,
      -- 更新データが連携されるパターンがあるので、最終更新データのみ取り込む用にROW_NUMBERを振る
      -- （カスタムディメンションのスコープ"セッション"のデータがセットされた、など）
      ROW_NUMBER() OVER (
        PARTITION BY t.event_id
        ORDER BY t.updated_at DESC
      ) AS row_num,
    FROM ca_tagdata.tagdata t
    INNER JOIN page_views pv ON
      pv.project_id = t.project_id AND
      pv.session_id = t.session_id AND
      pv.view_id = t.view_id
    WHERE t.event_type = 'content_event'
  )
  WHERE row_num = 1
) S ON
  T.event_id = S.event_id
WHEN MATCHED THEN UPDATE SET
  ssc_id = S.ssc_id,
  user_id = S.user_id,
  second = CASE WHEN S.second > T.second THEN S.second ELSE T.second END,
  clicks_count = CASE WHEN S.clicks_count > T.clicks_count THEN S.clicks_count ELSE T.clicks_count END,
  dimension1 = S.dimension1,
  dimension2 = S.dimension2,
  dimension3 = S.dimension3,
  dimension4 = S.dimension4,
  dimension5 = S.dimension5,
  dimension6 = S.dimension6,
  dimension7 = S.dimension7,
  dimension8 = S.dimension8,
  dimension9 = S.dimension9,
  dimension10 = S.dimension10,
  dimension11 = S.dimension11,
  dimension12 = S.dimension12,
  dimension13 = S.dimension13,
  dimension14 = S.dimension14,
  dimension15 = S.dimension15,
  dimension16 = S.dimension16,
  dimension17 = S.dimension17,
  dimension18 = S.dimension18,
  dimension19 = S.dimension19,
  dimension20 = S.dimension20,
  updated_at = S.updated_at
WHEN NOT MATCHED THEN INSERT ROW;


MERGE ca_tagdata.custom_event T
USING (
  WITH page_views AS (
	  SELECT
      project_id,
      session_id,
      view_id,
      client_id,
      ssc_id,
      user_id,
      normalized_url,
      original_url,
      title,
      device,
      dimension1,
      dimension2,
      dimension3,
      dimension4,
      dimension5,
      dimension6,
      dimension7,
      dimension8,
      dimension9,
      dimension10,
      dimension11,
      dimension12,
      dimension13,
      dimension14,
      dimension15,
      dimension16,
      dimension17,
      dimension18,
      dimension19,
      dimension20,
    FROM ca_tagdata.page_view
	)
  SELECT * except(row_num) FROM (
    SELECT
      t.event_id,
      DATE(t.timestamp, 'Asia/Tokyo') AS access_date,
      t.timestamp,
      t.project_id,
      t.session_id,
      t.view_id,
      pv.client_id,
      pv.ssc_id,
      pv.user_id,
      pv.normalized_url,
      pv.original_url,
      pv.title,
      pv.device,
      pv.dimension1,
      pv.dimension2,
      pv.dimension3,
      pv.dimension4,
      pv.dimension5,
      pv.dimension6,
      pv.dimension7,
      pv.dimension8,
      pv.dimension9,
      pv.dimension10,
      pv.dimension11,
      pv.dimension12,
      pv.dimension13,
      pv.dimension14,
      pv.dimension15,
      pv.dimension16,
      pv.dimension17,
      pv.dimension18,
      pv.dimension19,
      pv.dimension20,
      t.event_category,
      t.event_action,
      t.event_label,
      t.event_value,
      t.created_at,
      t.updated_at,
      -- 更新データが連携されるパターンがあるので、最終更新データのみ取り込む用にROW_NUMBERを振る
      -- （カスタムディメンションのスコープ"セッション"のデータがセットされた、など）
      ROW_NUMBER() OVER (
        PARTITION BY t.event_id
        ORDER BY t.updated_at DESC
      ) AS row_num,
    FROM ca_tagdata.tagdata t
    INNER JOIN page_views pv ON
      pv.project_id = t.project_id AND
      pv.session_id = t.session_id AND
      pv.view_id = t.view_id
    WHERE t.event_type = 'custom_event'
  )
  WHERE row_num = 1
) S ON
  T.event_id = S.event_id
WHEN MATCHED THEN UPDATE SET
  ssc_id = S.ssc_id,
  user_id = S.user_id,
  dimension1 = S.dimension1,
  dimension2 = S.dimension2,
  dimension3 = S.dimension3,
  dimension4 = S.dimension4,
  dimension5 = S.dimension5,
  dimension6 = S.dimension6,
  dimension7 = S.dimension7,
  dimension8 = S.dimension8,
  dimension9 = S.dimension9,
  dimension10 = S.dimension10,
  dimension11 = S.dimension11,
  dimension12 = S.dimension12,
  dimension13 = S.dimension13,
  dimension14 = S.dimension14,
  dimension15 = S.dimension15,
  dimension16 = S.dimension16,
  dimension17 = S.dimension17,
  dimension18 = S.dimension18,
  dimension19 = S.dimension19,
  dimension20 = S.dimension20,
  updated_at = S.updated_at
WHEN NOT MATCHED THEN INSERT ROW;

各テーブルへのSQLが正常終了したら、実際にデータが登録できたか確認しましょう。

以上で、すべての手順は完了です。 お疲れ様でした！