Google BigQuery Import Integration V2

Google BigQuery Export Integration V2の詳細はこちら

Google BigQuery V2のインテグレーションにより、BigQueryテーブルまたはクエリ結果からTreasure Dataにデータをインポートすることができます。

• Treasure Dataの基本的な知識
• Google Cloud Platform(BigQuery、Cloud Storage、IAM)の基本的な知識

• このコネクタではOAuthはサポートされなくなりました。サポートされている認証方式は、サービスアカウントJSON keyfileとWorkload Identity Federation(WIF)です。
• データセットがUSまたはEUマルチリージョン以外の場所にある場合は、ロケーションを指定する必要があります。指定しない場合、TDのジョブは「Cannot find job_id xxxxx」というエラーで失敗します。
• このコネクタは外部テーブルのインポートをサポートしていません。
• ビューからのインポートは、tableではなくqueryのインポートタイプを使用する必要があります。
• テーブルまたはクエリにRange型のフィールドが含まれている場合、GCSへのエクスポートは使用できません。使用すると「Unsupported type for JSON export: RANGE」というエラーが発生します。
• Require partition filterオプションが有効になっているパーティションテーブルからqueryタイプでインポートする場合、SQL文のWHERE句にパーティションフィールドを含める必要があります。含めないと、「Cannot query over table 'xxxxx' without a filter over column(s) '{partition field}' that can be used for partition elimination」というエラーが返されます。
• queryタイプのインポートでlegacyを使用する場合、INTERVAL、TIMESTAMP(12)、JSON、またはRANGEのデータ型のフィールドを含むソーステーブルはサポートされず、「Querying tables with INTERVAL, TIMESTAMP(12), JSON, or RANGE type is not supported in Legacy SQL」というエラーが返されます。
• Require partition filterオプションが有効になっているPartition Tableは、インポートモードtableでは増分インポートをサポートしていませんが、インポートモードqueryではサポートしています。

このコネクタは、auth_methodパラメータで選択する以下の2つの方式のいずれかでGoogle BigQueryに認証します。

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

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

このデータコネクタを使用するには、コネクタが認証するアイデンティティ(json_keyの場合はサービスアカウント、wifの場合はフェデレーテッドプリンシパルまたは権限借用されるサービスアカウント)に以下の権限またはIAMロールが必要です。

IAMの権限とロールの詳細については、Google Cloudのドキュメントを参照してください:BigQueryおよびCloud Storage

Workload Identity Federationを使用するには、サービスアカウントキーをダウンロードする代わりに、以下のGCP側のセットアップを完了してください。これにより、コネクタのadc_keyfileフィールドに貼り付けるApplication Default Credentials(ADC)キーファイルが生成されます。

このセットアップでは、Treasure AIのAWSアカウントを信頼するWorkload Identity Poolを作成します。これにより、インポートワークロードがそのAWSアイデンティティを短期有効なGoogleアクセストークンと交換できるようになり、長期キーは一切使用されません。Treasure AIのAWSアカウントIDはコネクタの認証フォーム(読み取り専用のTreasure AI'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の内容のみです。

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

どちらのモードを選択した場合も、使用するインポート機能(テーブルロード、クエリロード、Import Large Dataset)に応じて、権限とロールに記載されているロールを認証するアイデンティティに付与してください。直接フェデレーテッドアイデンティティの場合、プリンシパルはprincipalSet://iam.googleapis.com/projects/PROJECT_NUMBER/locations/global/workloadIdentityPools/POOL_ID/*の形式になります。権限借用の場合は、権限借用されるサービスアカウントにロールを付与します。ロールの不足やプールの設定ミスは実行時エラーの原因になります。具体的なエラーメッセージについては、Workload Identity Federationのトラブルシューティングを参照してください。

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

1. Integrations Hub > Catalogに移動します。
2. Google BigQueryを検索して選択します。
3. ダイアログが開きます。Authentication MethodとしてJSON keyfileまたはWorkload Identity Federationを選択します。
4. JSON keyfileの場合:サービスアカウントキーのJSON文字列を「JSON keyfile」セクションに入力します。新しいサービスアカウントキーの作成については、Google Cloudのドキュメントを参照してください。
5. Workload Identity Federationの場合:Workload Identity Poolからダウンロードしたexternal_account JSONを「ADC keyfile」セクションに入力します。取得方法については、Workload Identity Federation(WIF)のセットアップを参照してください。

SQL結果をロードする場合は、「Query statement」を選択し、「SQL statement」にSQLクエリを入力します。 転送を作成する前に、BigQuery Web UIでクエリが有効であることを確認してください。https://docs.cloud.google.com/bigquery/docs/bigquery-web-ui

接続を作成すると、自動的にAuthenticationsタブに移動します。作成した接続を探してSourceを選択します。

インポートするデータソースを設定します。 Google Cloud PlatformプロジェクトのIDを「Project ID」に入力します。 テーブルを保存するデータセット名を入力します。

インポートのタイプを選択します。テーブル全体をロードする(table loading)か、SQL結果をロードする(query loading)かを選択します。

テーブル全体をロードする場合は、「Table」を選択し、エクスポートしたい「Table name」を入力します。

マテリアライズドビューをロードする場合は、代わりに「Query Statement」を選択してください。

SQL結果をロードする場合は、「Query statement」を選択し、「SQL statement」にSQLクエリを入力します。

デフォルトのSQL方言はStandard SQL (GoogleSQL)です。Legacy SQLを使用する場合は、Use Legacy SQLをチェックしてください。

デフォルトでは、このコネクタは特定の条件下でキャッシュされた結果を使用します。キャッシュを無効にする場合は、Use Cached Resultsのチェックを外してください。

User Defined Function List: クエリで使用する関数

Legacy SQLの場合は以下に従ってください:https://cloud.google.com/bigquery/docs/user-defined-functions-legacy#register

Standard SQLの場合は以下に従ってください:https://cloud.google.com/bigquery/docs/user-defined-functions#sql-udf-structure

UDFがすでに存在する場合、クエリはエラーになる可能性があります。

データがUSまたはEUマルチリージョン以外の場所にある場合は、ロケーションを指定する必要があります。

データがasia-northeast1リージョンにある場合は、ロケーションを指定する必要があります。

ロケーションの詳細については、Google Cloudのドキュメントを参照してください。

インクリメンタルローディングは、自動増分ID列や作成日のタイムスタンプ列など、増加する一意の列を使用して、前回の実行後の新しいレコードのみをロードできます。

これを有効にするには、Incremental Loading をチェックし、「Incremental Column Name」に増分する列名を指定します。 インクリメンタル列としてサポートされているのは、INTEGER型とTIMESTAMP型のみです。

このコネクタは、インクリメンタル列で順序付けられた最新のレコードである「最終レコード」を記録します。次回の実行時には、最終レコードを使用して以下のルールで構築されたクエリを実行してレコードをロードします。

テーブルローディングの場合、すべてのフィールドがWHERE句で選択されます。

SELECT   * FROM   `${dataset}.${table}` WHERE   ${incremental_column} > ${value_of_last_record}

クエリローディングの場合、生のクエリがWHERE句でラップされます。

SELECT   * FROM   (${query}) embulk_incremental_ WHERE   ${incremental_column} > ${value_of_last_record}

大規模なデータセット(ベンチマークとして500MB以上)をロードする場合は、この「Import Large Dataset」オプションの使用を推奨します。このオプションは、データをGCS(Google Cloud Storage)オブジェクトとしてエクスポートし、複数のタスクでデータをロードします。そのため、ロードが高速化されます。

このオプションを有効にするには、Import Large Dataset をチェックし、「GCS bucket」と「GCS path prefix」を指定します。

• クエリ(クエリローディングまたはインクリメンタルローディング付きのテーブルローディング)を実行すると、クエリ結果はデータセット設定内の一時的なBigQueryテーブルにエクスポートされます。
• その後、一時テーブルは、パスプレフィックスを使用してGoogle Cloud Storageバケットに「gs://my-bucket/data-connector/result-[12桁の数字].jsonl.gz」としてgzip圧縮されたJSON Linesファイルとしてエクスポートされます。ファイルの数は結果データのサイズによって異なります。
• インクリメンタルローディングなしのテーブルローディングの場合、ソーステーブル内のすべてのデータが直接GCSにエクスポートされます。
• 完了後、一時テーブルとGCSオブジェクトは削除されます。

• データセットが「US」に設定されていない場合、GCSバケットもテーブルと同じロケーションにある必要があります。米国ベースのデータセットから別のリージョンのCloud Storageバケットにデータをエクスポートできます。詳細については、Google Cloudのエクスポート制限ドキュメントを参照してください。

インポートを実行する前に、Generate Preview を選択してデータのプレビューを表示できます。Data preview はオプションであり、選択した場合はダイアログの次のページに安全にスキップできます。

1. Next を選択します。Data Preview ページが開きます。
2. データをプレビューする場合は、Generate Preview を選択します。
3. データを確認します。

データの配置について、データを配置したいターゲット database と table を選択し、インポートを実行する頻度を指定します。

1. Next を選択します。Storage の下で、インポートされたデータを配置する新しい database を作成するか、既存の database を選択し、新しい table を作成するか、既存の table を選択します。

2. Database を選択 > Select an existing または Create New Database を選択します。

3. オプションで、database 名を入力します。

4. Table を選択 > Select an existing または Create New Table を選択します。

5. オプションで、table 名を入力します。

6. データをインポートする方法を選択します。

   • Append (デフォルト) - データインポートの結果は table に追加されます。 table が存在しない場合は作成されます。
   • Always Replace - 既存の table の全体の内容をクエリの結果出力で置き換えます。table が存在しない場合は、新しい table が作成されます。
   • Replace on New Data - 新しいデータがある場合のみ、既存の table の全体の内容をクエリの結果出力で置き換えます。

7. Timestamp-based Partition Key 列を選択します。 デフォルトキーとは異なるパーティションキーシードを設定したい場合は、long または timestamp 列をパーティショニング時刻として指定できます。デフォルトの時刻列として、add_time フィルターで upload_time を使用します。

8. データストレージの Timezone を選択します。

9. Schedule の下で、このクエリを実行するタイミングと頻度を選択できます。

1. Off を選択します。
2. Scheduling Timezone を選択します。
3. Create & Run Now を選択します。

1. On を選択します。
2. Schedule を選択します。UI では、@hourly、@daily、@monthly、またはカスタム cron の 4 つのオプションが提供されます。
3. Delay Transfer を選択して、実行時間の遅延を追加することもできます。
4. Scheduling Timezone を選択します。
5. Create & Run Now を選択します。

転送が実行された後、Data Workbench > Databases で転送の結果を確認できます。

BigQueryのデータ型は、次の表に示すように、対応するTreasure Dataの型に自動的に変換されます。テーブルまたはクエリ結果のスキーマにサポートされていない型が含まれている場合、エラーが発生します。

BigQueryとCloud Storageのすべてのクォータと制限は、GCPプロジェクトに適用されます。

必要に応じて、TD Toolbelt経由でコネクタを使用できます。

CLI上でTDツールベルトをセットアップします。

ここでは「config.yml」と呼ばれる設定YAMLファイルを作成します。

• GCSへのエクスポートとインクリメンタルを使用したテーブルからのインポート例

in:
  type: bigquery_v2
  json_keyfile:
    content: |
      {
      xxxxxxxx
      }
  project_id: xxxx
  dataset: xxx
  import_type: table
  table: xxxx
  location: US
  export_to_gcs: true
  gcs_bucket: xxxx
  gcs_path_prefix: xxxx
  incremental: true
  incremental_column: xxx

in:
  type: bigquery_v2
  json_keyfile:
    content: |
      {
      xxxxxxxx
      }
  project_id: xxxx
  dataset: xxx
  import_type: query
  query: xxxxxxx
  location: US
  udf:
    - function: |
        CREATE TEMP FUNCTION addTwo(x INT64)
        RETURNS INT64
        AS ( x + 2 )
    - function: |
        CREATE TEMP FUNCTION addTwo(x INT64)
        RETURNS INT64
        AS ( x + 2 )
  location: US
  export_to_gcs: true
  gcs_bucket: xxxx
  gcs_path_prefix: xxxx
  incremental: true
  incremental_column: xxx

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

プールからダウンロードしたexternal_account JSONは以下の構造です(直接フェデレーテッドアイデンティティの場合):

{
  "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の文字列ではありません)。サービスアカウントの権限借用の場合、ダウンロードされるJSONにはservice_account_impersonation_urlフィールドが追加で含まれます(例:https://iamcredentials.googleapis.com/v1/projects/-/serviceAccounts/SA_NAME@PROJECT_ID.iam.gserviceaccount.com:generateAccessToken)。

WIFを使用したコネクタ設定の例(テーブルからのインポート):

in:
  type: bigquery_v2
  auth_method: wif
  adc_keyfile:
    content: |
      {
        "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"
        }
      }
  project_id: xxxx
  dataset: xxx
  import_type: table
  table: xxxx
  location: US

(オプション) プレビュー

td td connector:preview コマンドを実行して、設定ファイルを検証します

td connector:preview config.yml

td connector:create を実行します。

次の例では、BigQuery コネクタを使用した日次インポートセッションが作成されます。

$ td connector:create daily_bigquery_import \     "10 0 * * *" td_sample_db td_sample_table config.yml Name     : daily_bigquery_import Cron     : 10 0 * * * Timezone : UTC Delay    : 0 Database : td_sample_db Table    : td_sample_table Config --- in:   ...

コネクタセッションでは、結果データ内に少なくとも 1 つのタイムスタンプカラムがデータパーティションキーとして使用される必要があり、デフォルトでは最初のタイムスタンプカラムがキーとして選択されます。カラムを明示的に指定する場合は、"--time-column" オプションを使用します。

$ td connector:create --time-column created_at \     daily_bigquery_import ...

結果データにタイムスタンプカラムがない場合は、次のようにフィルタ設定を追加して "time" カラムを追加します。

in:
  type: bigquery
  ...
filters:
  - type: add_time
    from_value:
      mode: upload_time
    to_column:
      name: time
out:
  type: td