Google Bigquery Export Integration V2

Google BigQuery Connector V2は、Google BigQueryへの大量データアップロードプロセスを効率化するように設計されています。以下の主要機能を提供します：

• Big Queryにアップロードされる大量データを処理するためのParquetファイルへの効率的なデータセットのパッケージング
• BigQueryロードジョブを使用した最適化されたデータアップロード
• Truncate同期モードのサポートを追加した柔軟なデータ同期操作

• TD Toolbeltを含むTreasure AIの基本知識
• Google Cloud Platformアカウント

• ARRAYのようなネストされたデータ型または繰り返しデータ型は、宛先列としてサポートされていません。

このコネクターは「append、replace、replace backup、truncate」モードをサポートしています。

コネクターは、auth_methodパラメーターで選択される2つの方法のいずれかを使用してGoogle BigQueryに認証します：

• Service Account JSONキー（json_key、デフォルト） — Google Cloudからダウンロードする長期間有効なJSON認証情報です。設定は簡単ですが、キーは保存・ローテーション・保護が必要な静的なシークレットです。
• Workload Identity Federation（wif） — キーレス認証です。Treasure AIのワークロード（AWS上で実行）が自身のAWS IDを短命なGoogleアクセストークンと交換するため、保存または漏洩する長期間有効なキーがありません。Googleはセキュリティとコンプライアンスの観点からサービスアカウントキーよりもWIFを推奨しており、ダウンロード可能なサービスアカウントキーを禁止している組織に適した選択肢です。

最も簡単な設定にはjson_keyを使用します。セキュリティポリシーで静的な認証情報が許可されていない場合はwifを使用します。既存のjson_key設定は変更なくそのまま動作します。

認証方法にかかわらず、コネクターが認証するID（json_keyの場合はサービスアカウント、wifの場合はフェデレーションプリンシパルまたはなりすまし対象のサービスアカウント）には、以下の両方のロールが必要です。エクスポートジョブはテーブルデータの書き込みとロードジョブの実行の両方を行うため、片方のロールだけでは不十分です：

• 対象データセットに対するBigQuery Data Editor（roles/bigquery.dataEditor） — テーブルの作成と書き込みを行うため。これはコネクターが起動時に検証するデータセットメタデータの読み取りもカバーします。
• プロジェクトに対するBigQuery Job User（roles/bigquery.jobUser） — BigQueryロードジョブを実行するため。これはプロジェクトレベルで付与する必要があります。データセットレベルの付与にはbigquery.jobs.createが含まれません。

Google Cloud Consoleでこれらを付与するには：

1. BigQuery Data Editorの場合：BigQueryを開き、対象のデータセットを選択し、Sharing > Permissions > Add principalを選択して、IDを入力し、BigQuery Data Editorを割り当てます。
2. BigQuery Job Userの場合：IAM & Admin > IAMを開き（対象プロジェクトを選択した状態で）、Grant accessを選択し、同じIDを入力して、BigQuery Job Userを割り当てます。

いずれかのロールが欠けていると実行時エラーが発生します。具体的なメッセージについてはトラブルシューティングを参照してください。

Service Account JSONキー方式を使用するには、以下が必要です：

• Project ID
• JSON Credential

Google BigQueryとの統合は、サーバー間API認証に基づいています。

1. Google Developer Consoleに移動します。
2. APIs & auth > Credentialsを選択します。
3. Service accountを選択します。

4. GoogleがJSON形式のキータイプを推奨しており、これを選択します。キーはブラウザによって自動的にダウンロードされます。

1. Google Developer Consoleに移動します。
2. Homeを選択します。
3. Project IDを確認します。

Workload Identity Federation認証方式を使用する場合は、サービスアカウントキーをダウンロードする代わりに、このGCP側の設定を行います。これにより、コネクターのadc_keyfileフィールドに貼り付けるApplication Default Credentials（ADC）キーファイルが生成されます。

Treasure AIのエクスポートワークロードはAWS上で実行されます。WIFはそのAWS IDをGoogleが直接信頼できるようにするため、長期間有効なキーは交換されません。この設定では、Workload Identity Poolを通じてTreasure AIのAWSアカウントにBigQueryリソースへのアクセスを付与します。Treasure AIのAWSアカウントIDはコネクターの認証フォーム（読み取り専用のTD's AWS Account IDフィールド）に表示されます。以下でサンプルとして123456789012が使用されている箇所には、その実際の値を使用してください。

Google Cloudプロジェクトで以下のAPIを有効にします：IAM、Resource Manager、Service Account Credentials、Security Token Service。

1. Google Cloud ConsoleでIAM & Admin > Workload Identity Federationに移動します。新しいプールを作成します（または既存のプールを選択します）。
2. Treasure AIのAWSアカウントID（コネクターの認証フォームに表示されます。ここではサンプルとして123456789012を使用）を使用して、タイプAWSのプロバイダーを追加します。

2つの付与モードのいずれかを選択します。どちらもコネクターでサポートされており、違いはダウンロードするexternal_account JSONのみです。

• 直接フェデレーションID — BigQueryロールをプールのプリンシパルに直接付与します。コネクターはフェデレーションID自体として認証します。
• サービスアカウントのなりすまし（impersonation） — （BigQueryロールを持つ）既存のサービスアカウントになりすます権限をプールに付与します。ダウンロードされる設定にはservice_account_impersonation_urlフィールドが含まれます。これには対象のサービスアカウントに対するroles/iam.serviceAccountTokenCreatorロールが必要です。

いずれのモードを選択する場合でも、認証するIDに必要なBigQuery権限で説明されているBigQuery Data EditorおよびBigQuery Job Userロールを付与します。直接フェデレーションIDの場合、プリンシパルはprincipalSet://iam.googleapis.com/projects/PROJECT_NUMBER/locations/global/workloadIdentityPools/POOL_ID/*の形式になります。なりすましの場合は、なりすまし対象のサービスアカウントにロールを付与します。

Workload Identity Poolから、AWSプロバイダー（Treasure AIのAWSアカウント）のDownload configを選択し、生成されたexternal_account JSONを保存します。これがadc_keyfileとして指定する値です。期待される構造については、以下の実例 WIF adc_keyfileを参照してください。

BigQueryコンソールからDatasetとTableを作成します。

1. Treasure コンソールに移動します。
2. Integrations Hub > Catalogに移動します。
3. Google Big Query V2を選択します。

4. 次のようにすべての情報を入力します：

Treasure コンソールは、データをエクスポートする複数の方法をサポートしています。Data Workbenchからデータをエクスポートするには、次の手順に従います。

1. Data Workbench > Queriesに移動します。
2. New Queryを選択し、クエリを定義します。
3. Export Resultsを選択して、データのエクスポートを設定します。
4. 既存のSnapchat CAPI認証を選択するか、前述の手順で新しい認証を作成します
5. Doneを選択します。

コネクターは、複数のソースからのスキーマを調和させて最終的なテーブル構造またはスキーマを決定する、スキーマ調整または統合プロセスを実装します。このプロセスは、設定された同期モードとユーザー設定に基づいてスキーマ進化シナリオをサポートしながら、データの整合性を保証します。

コネクターは、階層順に3つの潜在的なスキーマソースを評価することで、最終的なテーブルスキーマを決定します。この階層は、データスキーマが宛先BigQueryテーブルでどのように具現化されるかを理解するために重要です。

1. 宛先テーブルスキーマ（最高優先順位）

宛先テーブルが存在する場合、そのスキーマはデータロード操作の主要な権限として機能します。

適用される場合：

• 既存のテーブルへのAPPEND操作中
• TRUNCATE操作中

実例：数値型の処理

• クエリ結果データ

宛先テーブルスキーマ

• user_id: INT64
• transaction_amount: NUMERIC(38,9)
• transaction_date: DATE

結果: クエリ結果データがtransaction_amountに対してFLOAT64として提供される場合でも、宛先スキーマに従ってNUMERICに変換されます。

2. ユーザー定義のJSONスキーマ（第2優先順位）

JSON Schema File設定を通じて提供される場合、このスキーマはクエリ結果からの型推論を上書きし、明示的な列定義を提供します。

適用される場合：

• テーブル作成中（新しいテーブル）
• 既存のテーブルに新しい列を追加する場合
• 明示的な型キャストと列プロパティ定義の場合​

実例：タイムスタンプと日付の変換

• ソースデータ

• ユーザー定義スキーマ

• event_time: TIMESTAMP

• registration_date: DATE

• 効果

event_time:

• ソースデータはTIMESTAMPとして解析されます
• マイクロ秒の精度を維持します
• UTCで保存されます

3. クエリ結果スキーマ（最低優先順位）

ソースデータ構造から派生したスキーマ。他のスキーマが指定されていない場合のベースラインスキーマとして機能します。

適用される場合：

• JSONスキーマ定義なしで新しいテーブルを作成する場合
• 明示的な型定義なしで新しい列を追加する場合
• オーバーライドが存在しない場合の型推論のソースとして

実例：整数から文字列への変換

• クエリ結果データ

• クエリ結果スキーマ

• product_id: STRING

• status_code: Int

• 可能な変換先

1. status_code as STRING:

• INT64 → STRING（互換性のある変換）
• 結果: "200", "404"

2. status_code as INT64:

• 元の型を維持
• 結果: 200, 404

以下は、同期モード、テーブルの状態、設定によってスキーマがどのように処理されるかを示す構造化された参照表です。

コネクターは、ソースデータ型とBigQueryネイティブ型の間の包括的な型システムマッピングを実装しており、3つのカテゴリーの変換があります：

1. デフォルトマッピング（ロスレス変換）

2. 互換性のあるマッピング（型強制）

3. 潜在的にロスのあるマッピング（検証が必要）

Scheduled Jobs と Result Export を使用して、指定したターゲット宛先に出力結果を定期的に書き込むことができます。

Treasure Data のスケジューラー機能は、高可用性を実現するために定期的なクエリ実行をサポートしています。

2 つの仕様が競合するスケジュール仕様を提供する場合、より頻繁に実行するよう要求する仕様が優先され、もう一方のスケジュール仕様は無視されます。

例えば、cron スケジュールが '0 0 1 * 1' の場合、「月の日」の仕様と「週の曜日」が矛盾します。前者の仕様は毎月 1 日の午前 0 時 (00:00) に実行することを要求し、後者の仕様は毎週月曜日の午前 0 時 (00:00) に実行することを要求するためです。後者の仕様が優先されます。

1. Data Workbench > Queries に移動します

2. 新しいクエリを作成するか、既存のクエリを選択します。

3. Schedule の横にある None を選択します。

4. ドロップダウンで、次のスケジュールオプションのいずれかを選択します:

   ドロップダウン値	説明
   Custom cron...	Custom cron... の詳細を参照してください。
   @daily (midnight)	指定されたタイムゾーンで 1 日 1 回午前 0 時 (00:00 am) に実行します。
   @hourly (:00)	毎時 00 分に実行します。
   None	スケジュールなし。

 *    *    *    *    *
 -    -    -    -    -
 |    |    |    |    |
 |    |    |    |    +----- day of week (0 - 6) (Sunday=0)
 |    |    |    +---------- month (1 - 12)
 |    |    +--------------- day of month (1 - 31)
 |    +-------------------- hour (0 - 23)
 +------------------------- min (0 - 59)

次の名前付きエントリを使用できます:

• Day of Week: sun, mon, tue, wed, thu, fri, sat.
• Month: jan, feb, mar, apr, may, jun, jul, aug, sep, oct, nov, dec.

各フィールド間には単一のスペースが必要です。各フィールドの値は、次のもので構成できます:

1. (オプション) Delay execution を有効にすることで、クエリの開始時刻を遅延させることができます。

クエリに名前を付けて保存して実行するか、単にクエリを実行します。クエリが正常に完了すると、クエリ結果は指定された宛先に自動的にエクスポートされます。

(オプション) Delay execution を有効にすることで、クエリの開始時刻を遅延させることができます。

Audience Studio で activation を作成することで、segment データをターゲットプラットフォームに送信することもできます。

1. Audience Studio に移動します。
2. parent segment を選択します。
3. ターゲット segment を開き、右クリックして、Create Activation を選択します。
4. Details パネルで、Activation 名を入力し、前述の Configuration Parameters のセクションに従って activation を設定します。
5. Output Mapping パネルで activation 出力をカスタマイズします。

• Attribute Columns
  • Export All Columns を選択すると、変更を加えずにすべての列をエクスポートできます。
  • + Add Columns を選択して、エクスポート用の特定の列を追加します。Output Column Name には、Source 列名と同じ名前があらかじめ入力されます。Output Column Name を更新できます。+ Add Columns を選択し続けて、activation 出力用の新しい列を追加します。
• String Builder
  • + Add string を選択して、エクスポート用の文字列を作成します。次の値から選択します:
    • String: 任意の値を選択します。テキストを使用してカスタム値を作成します。
    • Timestamp: エクスポートの日時。
    • Segment Id: segment ID 番号。
    • Segment Name: segment 名。
    • Audience Id: parent segment 番号。

6. Schedule を設定します。

• スケジュールを定義する値を選択し、オプションでメール通知を含めます。

7. Create を選択します。

CLI（Toolbelt）を使用してBigQueryに結果をエクスポートすることもできます。

td queryコマンドの*--resultオプションを使用して、Snapchatサーバーへのエクスポート情報を指定する必要があります。td query*コマンドの詳細については、この記事を参照してください。

オプションの形式はJSONで、一般的な構造は次のとおりです。

APPENDモード：

type: 'bigquery_v2'
json_keyfile: |
  {
    "type": "service_account",
    "private_key_id": "xxx",
    "private_key": "-----BEGIN PRIVATE KEY-----xxx-----END PRIVATE KEY-----\n",
    "client_email": "account@xxx.iam.gserviceaccount.com",
    "client_id": "xxx",
    "auth_uri": "https://accounts.google.com/o/oauth2/auth",
    "token_uri": "https://accounts.google.com/o/oauth2/token",
    "auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs",
    "client_x509_cert_url": "https://www.googleapis.com/robot/v1/metadata/x509/account%40xxx.iam.gserviceaccount.com"
  }
mode: APPEND
project: gcp project id
dataset: bigquery dataset
table: bigquery table
location: gcp location
auto_create_table: true
add_missing_columns: true
schema_file: |
  [
    {"name": "c1", "type": "STRING", "mode": "NULLABLE"},
    {"name": "c2", "type": "INTEGER", "mode": "REQUIRED"}
  ]
skip_invalid_records: true

REPLACEモード：

  type: 'bigquery_v2'
  json_keyfile: |
    {
      "type": "service_account",
      "private_key_id": "xxx",
      "private_key": "-----BEGIN PRIVATE KEY-----xxx-----END PRIVATE KEY-----\n",
      "client_email": "account@xxx.iam.gserviceaccount.com",
      "client_id": "xxx",
      "auth_uri": "https://accounts.google.com/o/oauth2/auth",
      "token_uri": "https://accounts.google.com/o/oauth2/token",
      "auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs",
      "client_x509_cert_url": "https://www.googleapis.com/robot/v1/metadata/x509/account%40xxx.iam.gserviceaccount.com"
    }
  mode: REPLACE
  project: gcp project id
  dataset: bigquery dataset
  table: bigquery table
  location: gcp location
  auto_create_table: true
  add_missing_columns: true
  schema_file: |
    [
      {"name": "c1", "type": "STRING", "mode": "NULLABLE"},
      {"name": "c2", "type": "INTEGER", "mode": "REQUIRED"}
    ]
  skip_invalid_records: true

REPLACE_BACKUPモード：

  type: 'bigquery_v2'
  json_keyfile: |
    {
      "type": "service_account",
      "private_key_id": "xxx",
      "private_key": "-----BEGIN PRIVATE KEY-----xxx-----END PRIVATE KEY-----\n",
      "client_email": "account@xxx.iam.gserviceaccount.com",
      "client_id": "xxx",
      "auth_uri": "https://accounts.google.com/o/oauth2/auth",
      "token_uri": "https://accounts.google.com/o/oauth2/token",
      "auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs",
      "client_x509_cert_url": "https://www.googleapis.com/robot/v1/metadata/x509/account%40xxx.iam.gserviceaccount.com"
    }
  mode: REPLACE_BACKUP
  backup_mode: TABLE_RENAME
  project: gcp project id
  dataset: bigquery dataset
  table: bigquery table
  location: gcp location
  auto_create_table: true
  add_missing_columns: true
  schema_file: |
    [
      {"name": "c1", "type": "STRING", "mode": "NULLABLE"},
      {"name": "c2", "type": "INTEGER", "mode": "REQUIRED"}
    ]
  skip_invalid_records: true

TRUNCATEモード：

  type: 'bigquery_v2'
  json_keyfile: |
    {
      "type": "service_account",
      "private_key_id": "xxx",
      "private_key": "-----BEGIN PRIVATE KEY-----xxx-----END PRIVATE KEY-----\n",
      "client_email": "account@xxx.iam.gserviceaccount.com",
      "client_id": "xxx",
      "auth_uri": "https://accounts.google.com/o/oauth2/auth",
      "token_uri": "https://accounts.google.com/o/oauth2/token",
      "auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs",
      "client_x509_cert_url": "https://www.googleapis.com/robot/v1/metadata/x509/account%40xxx.iam.gserviceaccount.com"
    }
  mode: TRUNCATE
  project: gcp project id
  dataset: bigquery dataset
  table: bigquery table
  location: gcp location
  auto_create_table: true
  add_missing_columns: true
  schema_file: |
    [
      {"name": "c1", "type": "STRING", "mode": "NULLABLE"},
      {"name": "c2", "type": "INTEGER", "mode": "REQUIRED"}
    ]
  skip_invalid_records: true

Workload Identity Federationを使用する場合は、auth_methodをwifに設定し、json_keyfileの代わりにexternal_account JSONをadc_keyfileとして指定します。コネクターはEC2インスタンスメタデータサービス（IMDSv2）からAWS IDを読み取るため、Workload Identity Poolが信頼するIAMロールを持つTreasure AIのAWSベースのエクスポートワークロード上で実行する必要があります。

プールからダウンロードされるexternal_account JSONは、次の構造になっています（直接フェデレーションID）：

{
  "universe_domain": "googleapis.com",
  "type": "external_account",
  "audience": "//iam.googleapis.com/projects/PROJECT_NUMBER/locations/global/workloadIdentityPools/POOL_ID/providers/PROVIDER_ID",
  "subject_token_type": "urn:ietf:params:aws:token-type:aws4_request",
  "token_url": "https://sts.googleapis.com/v1/token",
  "credential_source": {
    "environment_id": "aws1",
    "region_url": "http://169.254.169.254/latest/meta-data/placement/availability-zone",
    "url": "http://169.254.169.254/latest/meta-data/iam/security-credentials",
    "regional_cred_verification_url": "https://sts.{region}.amazonaws.com?Action=GetCallerIdentity&Version=2011-06-15"
  }
}

audienceはWorkload Identity Providerの完全なリソース名である必要があり、PROJECT_NUMBERは数値のプロジェクトID（プロジェクトID文字列ではありません）です。サービスアカウントのなりすましの場合、ダウンロードされるJSONには追加でservice_account_impersonation_urlフィールドが含まれます（例：https://iamcredentials.googleapis.com/v1/projects/-/serviceAccounts/SA_NAME@PROJECT_ID.iam.gserviceaccount.com:generateAccessToken）。

WIFを使用したコネクター設定の例（APPENDモード）：

type: 'bigquery_v2'
auth_method: wif
adc_keyfile: |
  {
    "universe_domain": "googleapis.com",
    "type": "external_account",
    "audience": "//iam.googleapis.com/projects/PROJECT_NUMBER/locations/global/workloadIdentityPools/POOL_ID/providers/PROVIDER_ID",
    "subject_token_type": "urn:ietf:params:aws:token-type:aws4_request",
    "token_url": "https://sts.googleapis.com/v1/token",
    "credential_source": {
      "environment_id": "aws1",
      "region_url": "http://169.254.169.254/latest/meta-data/placement/availability-zone",
      "url": "http://169.254.169.254/latest/meta-data/iam/security-credentials",
      "regional_cred_verification_url": "https://sts.{region}.amazonaws.com?Action=GetCallerIdentity&Version=2011-06-15"
    }
  }
mode: APPEND
project: gcp project id
dataset: bigquery dataset
table: bigquery table
location: gcp location
auto_create_table: true
add_missing_columns: true
skip_invalid_records: true

APPENDモード

$ td query --result '{"type": "bigquery_v2", "td_authentication_id": 123456, "mode": "APPEND", "project": "gcp_project_id", "dataset": "bg_dataset", "table": "bg_table", "location": "US", "auto_create_table": true, "add_missing_columns": true, "schema_file": "[{\"name\": \"c1\", \"type\": \"INTEGER\"}, {\"name\": \"c2\", \"type\": \"STRING\"}]", "skip_invalid_records":true}' -d sample_datasets "select ........ from ........" -T presto

REPLACEモード

$ td query --result '{"type": "bigquery_v2", "td_authentication_id": 123456, "mode": "REPLACE", "project": "gcp_project_id", "dataset": "bg_dataset", "table": "bg_table", "location": "US", "auto_create_table": true, "add_missing_columns": true, "schema_file": "[{\"name\": \"c1\", \"type\": \"INTEGER\"}, {\"name\": \"c2\", \"type\": \"STRING\"}]", "skip_invalid_records":true}' -d sample_datasets "select ........ from ........" -T presto

REPLACE_BACKUPモード

$ td query --result '{"type": "bigquery_v2", "td_authentication_id": 123456, "mode": "REPLACE_BACKUP", "backup_mode": "TABLE_RENAME", "project": "gcp_project_id", "dataset": "bg_dataset", "table": "bg_table", "location": "US", "auto_create_table": true, "add_missing_columns": true, "schema_file": "[{\"name\": \"c1\", \"type\": \"INTEGER\"}, {\"name\": \"c2\", \"type\": \"STRING\"}]", "skip_invalid_records":true}' -d sample_datasets "select ........ from ........" -T presto

TRUNCATEモード

$ td query --result '{"type": "bigquery_v2", "td_authentication_id": 123456, "mode": "TRUNCATE", "project": "gcp_project_id", "dataset": "bg_dataset", "table": "bg_table", "location": "US", "auto_create_table": true, "add_missing_columns": true, "schema_file": "[{\"name\": \"c1\", \"type\": \"INTEGER\"}, {\"name\": \"c2\", \"type\": \"STRING\"}]", "skip_invalid_records":true}' -d sample_datasets "select ........ from ........" -T presto

• Result Exportをスケジュールして、定期的にターゲット先にデータをアップロードできます。
• すべてのインポートおよびエクスポート統合は、Treasure ワークフローに追加できます。tdデータオペレーターは、クエリ結果を指定された統合にエクスポートできます。詳細については、Reference for Treasure Data Operatorsを参照してください。