BigQuery

Foundry を Google BigQuery に接続して、BigQuery テーブルと Foundry データセットの間でデータを読み取り、同期します。

対応している機能

セットアップ

1. Data Connection アプリケーションを開き、画面右上の + 新規ソースを選択します。
2. 利用可能なコネクターの種類から BigQuery を選択します。
3. 表示される追加の設定プロンプトに従い、以下のセクションの情報を使ってコネクターのセットアップを進めます。

Foundry でのコネクターのセットアップについて詳しくはこちらをご覧ください。

認証

BigQuery コネクターを使用するには、以下の Identity and Access Management（IAM）ロールが必要です。

BigQuery データの読み取り：

• BigQuery 読み取りセッション ユーザー：BigQuery プロジェクトに対して付与します
• BigQuery データ閲覧者：データとメタデータを読み取るために、BigQuery データに対して付与します
• BigQuery ジョブユーザー（任意）：ビューの取り込みとカスタムクエリの実行のために付与します

Foundry から BigQuery へのデータのエクスポート：

• BigQuery データ編集者：BigQuery データセットまたはプロジェクトに対して付与します
• BigQuery ジョブユーザー：BigQuery プロジェクトに対して付与します
• ストレージオブジェクト管理者：Google Cloud Storage を使用してデータをエクスポートする場合、バケットに対して付与します

一時テーブルの使用：

• BigQuery データ編集者：コネクターがデータセットを自動的に作成する場合、BigQuery プロジェクトに対して付与します
• BigQuery データ編集者：一時テーブルの保存先として指定したデータセットに対して付与します

必要なロールについて詳しくはこちらの Google Cloud のアクセス制御に関するドキュメント ↗ をご覧ください。

利用可能な認証方法から1つを選択します。

• GCP インスタンス アカウント： インスタンスベースの認証のセットアップ方法については、Google Cloud ドキュメント ↗ を参照してください。

  • GCP インスタンス認証は、GCP 内の適切に設定されたインスタンスで実行される Agent を介して動作するコネクターでのみ機能します。
  • GCP インスタンス認証の資格情報は、仮想テーブルでは利用できません。

• サービスアカウントキーファイル： サービスアカウントキーファイル認証のセットアップ方法については、Google Cloud ドキュメント ↗ を参照してください。キーファイルは JSON または PKCS8 の資格情報として指定できます。

• ワークロード ID フェデレーション (OIDC)： 表示されるソースシステムの設定手順に従って、OIDC をセットアップします。ワークロード ID フェデレーションの詳細については Google Cloud ドキュメント ↗ を、Foundry での OIDC の動作の詳細については Foundry のドキュメントを参照してください。

ネットワーク

BigQuery コネクターには、ポート443で以下のドメインへのネットワークアクセスが必要です。

• bigquery.googleapis.com
• bigquerystorage.googleapis.com
• storage.googleapis.com
• www.googleapis.com

以下のドメインへの追加のアクセスが必要になる場合があります。

• oauth2.googleapis.com
• accounts.google.com

Google Cloud Platform (GCP) 上の Foundry と GCP 上の BigQuery の間で直接接続を確立する場合、関連する VPC Service Controls でも接続を有効にする必要があります。この接続がセットアップに必要な場合は、追加のガイダンスについて Palantir サポートにお問い合わせください。

接続情報

BigQuery コネクターでは、以下の設定オプションを利用できます。

[1] 仮想テーブルを介して BigQuery ビュー ↗を登録する場合は、一時テーブルを有効にする必要があります。

[2] パラメーター化されていない BIGNUMERIC 列を読み取るには、Spark に対応した BIGNUMERIC のスケールと精度を有効にする必要があります。Spark が対応する精度の最大値は38、スケールの最大値は18であるため、BigQuery 内のデータの精度またはスケールがこれらの上限を超えると、データが失われる可能性があります。精度またはスケールが Spark の上限を超える、パラメーター化された BIGNUMERIC 列には対応していません。

BigQuery からデータを同期する

BigQuery コネクターでは、大規模なデータ同期やカスタムクエリ向けに高度な同期設定を利用できます。より複雑なシナリオでは、仮想テーブルまたはコードを使用した代替アプリケーションを使用します。

BigQuery の同期を設定するには、ソースの概要画面の右上にあるソースを探索して同期を作成を選択します。次に、Foundry に同期するテーブルを選択します。同期の準備ができたら、x個のデータセットの同期を作成を選択します。

Foundry のソース探索について詳しくはこちら。

利用可能な同期を探索してコネクターに追加したら、同期を編集に移動します。左側の同期パネルで、設定する同期を見つけ、その右側にある > を選択します。

データモデル

BigQuery のテーブルは、データが Avro 形式で保存された状態で Foundry にインポートされます。インポート時には、BIGNUMERIC 型および TIME 型の列には対応していません。

Foundry から BigQuery にデータをエクスポートする場合、MAPS、STRUCTS、ARRAYS を除くすべての列のデータ型に対応しています。

インポート設定

BigQuery から Foundry に同期するデータを選択します。

テーブル全体を同期する

テーブル全体を Foundry に同期するには、次の情報を入力します。

カスタム SQL

任意のクエリを実行でき、その結果は Foundry に保存されます。クエリの出力は20 GB（BigQuery テーブルの最大サイズ）未満である必要があります。それを超える場合は、一時テーブルの使用を有効にする必要があります。クエリはキーワード select または with で始まる必要があります。例：SELECT * from table_name limit 100;。

インクリメンタル（差分）同期

通常、同期では、同期の間にデータが変更されたかどうかにかかわらず、対象テーブルから条件に一致するすべての行をインポートします。一方、インクリメンタル同期では、直近の同期に関する状態を保持し、対象から条件に一致する新しい行のみを取り込みます。

インクリメンタル同期は、BigQuery から大規模なテーブルを取り込む際に使用できます。インクリメンタル同期を使用するには、テーブルに厳密に単調増加する列が含まれている必要があります。また、読み取り元のテーブルまたはクエリには、次のいずれかのデータ型の列が含まれている必要があります。

• INT64
• FLOAT64
• NUMERIC
• BIGNUMERIC
• STRING
• TIMESTAMP
• DATE
• TIME
• DATETIME

例： 5 TB のテーブルに、BigQuery に同期したい数十億行のデータが含まれています。このテーブルには、id という名前の単調増加する列があります。id 列をインクリメンタル列として使用し、初期値を -1、上限を5000万行に設定することで、1回につき5000万行を取り込むように同期を設定できます。

同期を初めて実行すると、id の値が -1より大きい最初の5000万行（id に基づく昇順）が Foundry に取り込まれます。たとえば、この同期を数回実行し、前回の実行で取り込まれた最大の id が 19384004822 だった場合、次の同期では 19384004822 より大きい最初の id から始まる次の5000万行が取り込まれ、以降も同様に続きます。

インクリメンタル同期には、次の設定が必要です。

カスタムクエリのインクリメンタル同期

カスタムクエリの同期でインクリメンタルクエリを有効にするには、クエリを次の形式に合わせて更新する必要があります。

SELECT * from table_name where incremental_column_name > @value order by incremental_column_name asc limit @limit

一時テーブル

標準以外の BigQuery テーブルから20 GB を超えるデータを同期する場合は、一時テーブルを有効にする必要があります。一時テーブルを使用すると、BigQuery で大規模なクエリ出力の結果や、ビューおよびその他の標準以外のテーブルのデータをインポートでき、大規模なインクリメンタルインポートおよび取り込みも可能になります。

1回の同期で、ディスクの空き容量までのデータをインポートできます。Foundry ワーカーで実行される同期では、通常、この上限は600 GB です。より大きなテーブルをインポートするには、インクリメンタル同期を使用します。

一時テーブルを使用するには、コネクターの設定で一時テーブルの設定を有効にし、必要に応じて次の設定を行います。

• データセットを自動的に作成： この選択肢を選択すると、一時テーブルを保存するための Palantir_temporary_tables データセットが作成されます。この選択肢を使用するには、BigQuery アカウントがプロジェクトに対する BigQuery データ編集者ロールを持っている必要があります。
• 使用するデータセットを指定： この選択肢を選択して、一時テーブルの保存に使用するデータセットのプロジェクト ID とデータセット名を手動で追加します。この選択肢を使用するには、BigQuery アカウントが指定したデータセットに対する BigQuery データ編集者ロールを持っている必要があります。

BigQuery ビュー ↗を仮想テーブルで使用する場合は、一時テーブルを有効にする必要があります。

データを BigQuery にエクスポートする

BigQuery へのエクスポートは、エクスポートタスクを介して行います。より複雑なシナリオでは、プロコードの代替アプリケーションを使用します。

コネクターは、次の2つの方法で BigQuery にエクスポートできます。

• Google Cloud Storage を中間ストアとして使用する方法（推奨）。詳しくは、以下の Cloud Storage 経由のエクスポートのセクションを参照してください。
• BigQuery APIs を使用する方法。使用する API の詳細については、ローカルファイルからデータを読み込む ↗を参照してください。

API 経由の書き込みは、数百万行規模のデータに適しています。このモードで期待されるパフォーマンスは、約2分で100万行をエクスポートすることです。データが数十億行規模に達する場合は、代わりに Google Cloud Storage を使用してください。

タスクの設定

データのエクスポートを開始するには、エクスポートタスクを設定する必要があります。エクスポート先のコネクターが含まれるプロジェクトフォルダーに移動します。コネクター名を右クリックし、Data Connection タスクを作成を選択します。

Data Connection の表示の左側パネルで、ソース名が使用するコネクターと一致していることを確認します。次に、入力項目に入力データセットを追加します。入力データセットには inputDataset という名前を付ける必要があり、これはエクスポートする Foundry データセットです。エクスポートタスクには、出力を1個設定する必要もあります。出力データセットには outputDataset という名前を付け、Foundry データセットを指定する必要があります。出力データセットは、タスクの実行、スケジュール設定、および監視に使用されます。

Data Connection の表示の左側パネルで、次の操作を行います。

1. ソース名が使用するコネクターと一致していることを確認します。
2. inputDataset という名前の入力を追加します。入力データセットは、エクスポートする Foundry データセットです。
3. outputDataset という名前の出力を追加します。出力データセットは、タスクの実行、スケジュール設定、および監視に使用されます。
4. 最後に、テキスト項目に YAML ブロックを追加して、タスクの設定を定義します。

左側パネルに表示されるコネクターと入力データセットのラベルには、YAML で定義した名前は反映されません。

エクスポートタスクの YAML を作成するときは、次のオプションを使用します。

タスクの設定例：

Copied!
1
2
3
4
5
6
7
type: magritte-bigquery-export-task
config:
    table:
      dataset: datasetInBigQuery
      table: tableInBigQuery
      project: projectId #(Optional: Do not specify unless the Project for export differs from the Project configured for the connector.)
    incrementalType: SNAPSHOT | REQUIRE_INCREMENTAL

エクスポートタスクを設定したら、右上隅の保存を選択します。

出力テーブルに行を追加する

宛先テーブルに行を追加する必要がある場合は、データセットを置き換えるのではなく、REQUIRE_INCREMENTAL を使用できます。

Cloud Storage 経由でエクスポートする（推奨）

Cloud Storage 経由でエクスポートするには、コネクターの設定で Cloud Storage バケットを設定する必要があります。また、バケットに一時的に書き込まれたデータにアクセスできるユーザーを可能な限り少なくするため、Cloud Storage バケットはコネクターの一時テーブル専用にする必要があります。

BigQuery APIs ではなく、Cloud Storage 経由で BigQuery にエクスポートすることを推奨します。Cloud Storage は大規模なデータをより効率的に処理でき、BigQuery に一時テーブルを作成しません。

BigQuery APIs 経由でエクスポートする

エクスポートジョブは、宛先テーブルとともに一時テーブルを作成します。この一時テーブルには、追加のアクセス制限は適用されません。また、SNAPSHOT エクスポートではテーブルを削除して再作成するため、追加のアクセス制限も削除されます。

BigQuery APIs 経由のエクスポートでは、データは一時テーブル datasetName.tableName_temp_$timestamp にエクスポートされます。エクスポートが完了すると、行は一時テーブルから宛先テーブルに自動的に転送されます。

エクスポートを REQUIRE_INCREMENTAL モードで実行する場合は、宛先テーブルを共有できます。SNAPSHOT モードで実行すると、実行のたびにテーブルが再作成されるため、共有設定を再適用する必要があります。

コードで BigQuery ソースを使用する

より複雑なシナリオでは、プロコードの代替アプリケーションを使用して BigQuery ソースに接続できます。

以下の例では、外部トランスフォームで Google BigQuery の Python クライアント ↗ google-cloud-bigquery を使用して、BigQuery ソースに接続する方法を示します。

これらの例は、BigQuery に保存された Shipments テーブルに基づいています。このテーブルは、荷物のシップメントと、その重量、運送業者、追跡番号の一覧を表します。

外部トランスフォームで BigQuery から読み込む

この例では、指定された運送業者の一覧に含まれ、かつ一定の重量を超えるシップメントを読み込みます。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
from pandas import DataFrame
from transforms.api import lightweight, Output, Input, transform_pandas
from transforms.external.systems import ResolvedSource, external_systems, Source
from google.cloud import bigquery
import json

@lightweight
@external_systems(
    bigquery_source=Source("<source_rid>")
)
@transform_pandas(
    Output("<dataset_rid>"),
    carriers_df=Input("<dataset_rid>") # スキーマ [carrier_name: String, carrier_details: String] を持つデータフレーム
)
def read(bigquery_source: ResolvedSource, carriers_df: DataFrame) -> DataFrame:
    # BigQuery クライアントを認証します
    account_info = json.loads(bigquery_source.get_secret("jsonCredentials"))
    client = bigquery.Client.from_service_account_info(account_info)

    # テーブル参照
    bigquery_project = "<bigquery_project_id>"
    bigquery_dataset = "<bigquery_dataset_id>"
    bigquery_table = "<bigquery_table_id>"
    table_ref = f"{bigquery_project}.{bigquery_dataset}.{bigquery_table}"

    # フィルターのパラメーター
    carrier_list = carriers_df['carrier_name'].drop_duplicates().tolist()
    min_weight = 10

    job_config = bigquery.QueryJobConfig(
        query_parameters=[
            bigquery.ArrayQueryParameter("carrier_list", "STRING", carrier_list),
            bigquery.ScalarQueryParameter("min_weight", "FLOAT64", min_weight)
        ]
    )

    query = f"""
        SELECT shipment_id, carrier, weight, tracking_number
        FROM `{table_ref}`
        WHERE carrier IN UNNEST(@carrier_list) AND weight >= @min_weight
    """

    query_job = client.query(query, job_config=job_config)
    return query_job.to_dataframe()

外部トランスフォームで BigQuery に書き込む

この関数は、指定されたシップメント情報に基づいてシップメントの追跡番号を更新し、更新されたシップメントの詳細を返します。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
from pandas import DataFrame
from transforms.api import lightweight, Output, Input, transform_pandas
from transforms.external.systems import ResolvedSource, external_systems, Source
from google.cloud import bigquery
import json

@lightweight
@external_systems(
    bigquery_source=Source("<source_rid>")
)
@transform_pandas(
    Output("<dataset_rid>"),
    shipments_df=Input("<dataset_rid>"), # スキーマ [shipment_id: Long, tracking_number: String] を持つ DataFrame
)
def update_tracking_numbers(bigquery_source: ResolvedSource, shipments_df: DataFrame) -> DataFrame:
    # BigQuery クライアントを認証します
    account_info = json.loads(bigquery_source.get_secret("jsonCredentials"))
    client = bigquery.Client.from_service_account_info(account_info)

    # テーブル参照
    bigquery_project = "<bigquery_project_id>"
    bigquery_dataset = "<bigquery_dataset_id>"
    bigquery_table = "<bigquery_table_id>"
    table_ref = f"{bigquery_project}.{bigquery_dataset}.{bigquery_table}"

    # 1. 一時テーブルが存在する場合はクリアします
    temp_table_id = f"{bigquery_project}.{bigquery_dataset}.temp_update_tracking"
    client.delete_table(temp_table_id, not_found_ok=True)


    # 2. shipments_df を一時テーブルにアップロードします
    load_job_config = bigquery.LoadJobConfig(
        schema=[
            bigquery.SchemaField("shipment_id", "INT64"),
            bigquery.SchemaField("tracking_number", "STRING"),
        ]
    )
    job = client.load_table_from_dataframe(shipments_df, temp_table_id, job_config=load_job_config)
    job.result()  # 読み込みが完了するまで待ちます

    # 3. MERGE でメインテーブルの tracking_number を更新します
    merge_query = f"""
    MERGE `{table_ref}` T
    USING `{temp_table_id}` S
    ON T.shipment_id = S.shipment_id
    WHEN MATCHED THEN
      UPDATE SET tracking_number = S.tracking_number
    """

    client.query(merge_query).result()

    # 3. （任意）一時テーブルを削除します
    client.delete_table(temp_table_id, not_found_ok=True)

    # 必要に応じて、確認用に更新された行を DataFrame として返します
    updated_ids = shipments_df["shipment_id"].tolist()
    query = f"""
        SELECT shipment_id, tracking_number
        FROM `{table_ref}`
        WHERE shipment_id IN UNNEST(@updated_ids)
    """
    job_config = bigquery.QueryJobConfig(
        query_parameters=[
            bigquery.ArrayQueryParameter("updated_ids", "INT64", [int(x) for x in updated_ids])
        ]
    )
    updated_rows = client.query(query, job_config=job_config).to_dataframe()
    return updated_rows

仮想テーブル

このセクションでは、BigQuery ソースで仮想テーブルを使用する際の詳細を説明します。Foundry データセットに同期する場合、このセクションは適用されません。

次のテーブルは、BigQuery でサポートされている仮想テーブルの機能を示しています。

BigQuery テーブルを入力または出力として使用できる、サポート対象の Foundry ワークフローについては、仮想テーブルのドキュメントを参照してください。

[1] BigQuery 仮想テーブルを元にしているパイプラインでインクリメンタル処理を有効にするには、適切な保持期間で Time Travel ↗ が有効になっていることを確認してください。この機能は Change History ↗ に依存しており、現在は APPEND のみに対応しています。Python トランスフォームの current および added 読み取りモードに対応しています。Python トランスフォームでは、_CHANGE_TYPE 列と _CHANGE_TIMESTAMP 列が利用可能になります。

ソース設定の要件

仮想テーブルを使用する場合は、次のソース設定の要件に留意してください。

• Foundry ワーカーのソースを使用する必要があります。仮想テーブルは Agent ワーカーの接続の使用に対応していません。
• このドキュメントのネットワークのセクションで説明されているとおり、双方向の接続と許可リスト登録が確立されていることを確認してください。
• Code Repositories で仮想テーブルを使用する場合、必要な追加のソース設定については、仮想テーブルのドキュメントを参照してください。
• サービスアカウントのキーファイルの資格情報、またはワークロード ID フェデレーション (OIDC) を使用します。仮想テーブルはインスタンスベースの認証に対応していません。
• BigQuery ビュー ↗を登録するには、一時テーブルを有効にし、関連するロールを認証用の資格情報に追加する必要があります。
• Spark でパラメーター化されていない BIGNUMERIC 列を読み取るには、Spark の設定を行います。

詳細については、上記の接続情報と一時テーブルのセクションを参照してください。

コンピュートプッシュダウン

Foundry では、仮想テーブルを使用する際に、コンピュートを BigQuery にプッシュダウンできます。仮想テーブルの入力には、述語プッシュダウンのサポートが組み込まれた BigQuery Spark コネクター ↗が使用されます。

同じソースに登録された BigQuery 仮想テーブルをパイプラインの入力および出力として使用する場合、コンピュートを完全に BigQuery に委譲できます。この機能は現在、Python トランスフォームで利用可能です。コンピュートを BigQuery にプッシュダウンする方法の詳細については、Python のドキュメントを参照してください。

トラブルシューティング

見つかりません：データセット <dataset> が場所 <location> に見つかりませんでした

BigQuery は、クエリで使用される入力、またはクエリの結果が保存される場所に基づいて、そのクエリを実行する場所を決定します。一時テーブルを使用する場合、出力は一時テーブル用のデータセット内の一時テーブルに設定されます。このデータセットの場所によって、クエリの実行場所が決まります。同期のすべての入力と一時テーブル用のデータセットが同じリージョンにあることを確認してください。データセットを自動的に作成する。 設定が有効になっている場合は、Google Cloud コンソールまたは Google の SDK/API を使用して、Palantir_temporary_tables という名前のデータセットの場所を特定します。

gRPC メッセージが最大サイズを超える

JSON 列などの大きなコンテンツを含むデータを同期する場合、上記のエラーで転送に失敗することがあります。ソースの gRPC 設定で BigQuery の受信メッセージの最大サイズを調整し、1回の API 呼び出しで転送するデータ量を増やします。1回の呼び出しで複数行を取得するため、最大の行サイズに設定するだけでは不十分な場合があることに留意してください。

この設定オプションは、接続設定 > 接続情報に移動し、追加オプションまでスクロールして gRPC 設定を選択すると表示されます。

BigQuery ビューに基づくデータセットまたは仮想テーブルのプレビューの読み込みに失敗する

プレビューの読み込みに失敗し、対象の BigQuery テーブルがビューである場合、スキーマの不一致が原因である可能性があります。ビューのスキーマは、作成時に定義された SQL クエリの結果によって決まります。その後、基となるテーブルの構造が変更されても、ビューのスキーマには自動的に適用されません。そのため、ビューを使用しようとした際（たとえば、Palantir プラットフォームで同期や仮想テーブルを介して使用する際）に、予期しない動作が発生する可能性があります。

BigQuery コンソールでビューと基となるテーブルのスキーマを比較し、不一致がないか調べることで、これを確認できます。不一致には、次のようなものがあります。

• ビューには存在するが、基となるテーブルには存在しない列（またはその逆）
• 同じ列に対して2つの異なるデータ型があること

不一致が見つかった場合は、CREATE OR REPLACE VIEW または ALTER VIEW を使用してビューを再作成または更新することで対処できます。詳細については、Google Cloud コミュニティの投稿 ↗を参照してください。期待される動作の詳細と、ベストプラクティスに関する追加のガイダンスが説明されています。