初心者のためのBigQuery APIの使い方

Google BigQuery は、組織が大規模なデータ分析を扱う方法を革新しました。サーバーレスアーキテクチャ、スケーラビリティ、馴染みのある SQL インターフェースにより、大規模なデータセットから洞察を導き出すための強力なツールとなっています。 Google Cloud Console や bq コマンドラインツールを通じて BigQuery と対話することは一般的ですが、自動化、統合、およびカスタムアプリケーション開発の真の力は、その包括的なアプリケーションプログラミングインターフェース（API）のセットを通じて解放されます。

このガイドでは、BigQuery API の使用方法を段階的に探求し、プログラムを介してデータウェアハウスと対話し、データパイプラインを構築し、アプリケーションに BigQuery を統合し、リソースを効率的に管理できるようにします。利用可能な API のさまざまな種類、環境のセットアップ方法、Python クライアントライブラリを使用した実用的な例、および高度なユースケース向けの専門的な API を紹介します。

💡

素晴らしい API テストツールを探していますか？美しい API ドキュメントを生成しますか？

開発チームの生産性を最大限に引き出すための、統合されたオールインワンプラットフォームが必要ですか？

ApiDog は、すべての要求に応え、非常に手ごろな価格で Postman を置き換えます！

ボタン

BigQuery API の理解

コードに入る前に、コアコンセプトと BigQuery とプログラム的に対話する方法を理解することが重要です。

コア BigQuery コンセプト:

プロジェクト: Google Cloud の最上位コンテナで、BigQuery データセット、テーブル、ジョブを含むすべてのリソースを保持します。請求と権限は通常、プロジェクトレベルで管理されます。
データセット: 特定のプロジェクト内のテーブルとビューのコンテナです。データセットはデータを整理し、アクセスを制御するのに役立ちます。従来のシステムのスキーマやデータベースのように考えてください。
テーブル: 定義されたスキーマで行と列に構造化されたデータを保存します。 BigQuery は、さまざまなデータ型やネストされた/繰り返しフィールドをサポートしています。
ジョブ: BigQuery で実行される非同期アクションを表し、クエリの実行、データのロード、データのエクスポート、テーブルのコピーなどがあります。 API を使用すると、これらのジョブを開始、監視、および管理できます。

BigQuery API の種類:

BigQuery は、サービスとプログラム的に対話するためのいくつかの方法を提供しています。

REST API: これは、HTTP と JSON に基づいて構築された基盤となる API です。これを使用すると、BigQuery リソースと操作に直接アクセスできます。標準 HTTP リクエスト (GET、POST、PUT、DELETE) を使用して、特定のエンドポイント (例: https://bigquery.googleapis.com/bigquery/v2/projects/{projectId}/datasets) をターゲットに対話できます。強力で詳細な制御を提供しますが、REST API を直接使用する場合は、認証、リクエストフォーマット、応答解析、およびエラー処理を手動で処理する必要があります。認証には通常、OAuth 2.0 アクセストークンが含まれます。

クライアントライブラリ: Google は、さまざまな人気のあるプログラミング言語 (Python、Java、Go、Node.js、C#、PHP、Ruby など) 用の高レベルのクライアントライブラリを提供しています。これらのライブラリは、基盤となる REST API をラップし、より慣用的で開発者に優しいエクスペリエンスを提供します。一般的なタスクを簡素化し、認証を管理 (アプリケーションデフォルト認証情報を介して自動的に行うことがよくあります)、リトライを管理し、記述する必要のあるボイラープレートコードの量を削減します。これは、ほとんどのアプリケーション開発に推奨されるアプローチです。

専門の API: 特定の高性能または専門的なタスクに対して、BigQuery は専用の API を提供しています：

BigQuery ストレージ読み取り API: BigQuery 管理ストレージから直接高スループットデータを読み取るために設計されています。標準のクエリ結果エクスポートよりも大幅に高速で、特に大きなデータセットに対して、Apache Spark、Apache Beam、および Pandas などのデータ処理フレームワークとうまく統合されます。複数のストリームを介して並列読み取りが可能です。
BigQuery 接続 API: Cloud SQL、Spanner、Cloud Storage などの外部データソースへの接続を作成および管理することができます。これにより、データを最初に BigQuery にロードすることなく、EXTERNAL_QUERY 関数を使用して、これらの外部システムのデータを BigQuery から直接クエリできます。
Analytics Hub API: 異なる組織や事業部門間で、安全でスケーラブルなデータ共有を促進します。 "データ交換" 内で共有可能なデータセットへの参照である「リスティング」を作成できます。消費者は、これらのリスティングにサブスクライブして、各自のプロジェクト内で共有データにアクセスできます。
BigLake API: BigLake テーブルを管理するために使用されます。BigLake を使用すると、データレイク (主に Google Cloud Storage) に格納されているオープンフォーマット (Parquet、ORC、Avro など) のデータ上で動作する BigQuery のテーブルを作成できます。これにより、管理された BigQuery ストレージとデータレイクストレージの両方を一元的にクエリでき、セキュリティとガバナンスを一貫して保つことができます。

環境のセットアップ

API を呼び出す準備をするには、ローカルまたはサーバー環境を設定する必要があります。

前提条件:

Google Cloud アカウント: アクティブな Google Cloud アカウントが必要です。
Google Cloud プロジェクト: Google Cloud Console で新しいプロジェクトを作成するか、既存のプロジェクトを選択します。プロジェクト ID をメモしてください。
BigQuery API の有効化: プロジェクトに対して BigQuery API が有効になっていることを確認します。これを Cloud Console (API & サービス > ライブラリ > "BigQuery API" を検索 > 有効化) を介して行うことができます。ユースケースに応じて、BigQuery ストレージ読み取り API や BigQuery 接続 API など、他の API を有効にする必要がある場合もあります。
請求: プロジェクトに対して請求が有効になっていることを確認します。BigQuery の操作は、ストレージされたデータ、処理された分析、ストリーミング挿入に基づいてコストが発生します。

認証:

アプリケーションが Google Cloud に認証し、そのアイデンティティおよび BigQuery リソースへのアクセス権限を確認する必要があります。ほとんどのシナリオに推奨される方法は、アプリケーションデフォルトの認証情報 (ADC) です。

アプリケーションデフォルトの認証情報 (ADC): この戦略により、アプリケーションは、実行されている環境に基づいて認証情報を自動的に見つけることができ、コードにキーを直接埋め込む必要がなくなります。
ローカル開発: Google Cloud CLI (gcloud) をインストールします。コマンド gcloud auth application-default login を実行します。これにより、Google アカウントでログインするためのブラウザウィンドウが開き、SDK に認証情報へのアクセスが許可されます。クライアントライブラリはこれらの認証情報を自動的に検出します。
Compute Engine、Cloud Functions、App Engine など: ADC は、自分のコードが実行されているリソースに関連付けられたサービスアカウントを自動的に使用します。
サービスアカウント: Google Cloud の外部で実行されるアプリケーションや特定の権限を必要とするアプリケーションでは、サービスアカウントを使用できます。

Cloud Console (IAM & 管理 > サービスアカウント) でサービスアカウントを作成します。
サービスアカウントに必要な BigQuery 権限 (例: BigQuery データエディタ、BigQuery ジョブユーザー、BigQuery ユーザー) を付与します。
サービスアカウントのキーをダウンロードします (JSON 形式)。
環境変数 GOOGLE_APPLICATION_CREDENTIALS をダウンロードした JSON キーファイルの絶対パスに設定します。環境変数が設定されると、クライアントライブラリは、このキーファイルを認証に自動的に使用します。

クライアントライブラリのインストール (Python の例):

例では Python に焦点を当てます。以下のコマンドを使用して、必要なライブラリをインストールできます：

pip install google-cloud-bigquery
# オプション: より高速な読み取りのためにストレージ API ライブラリをインストール
pip install google-cloud-bigquery-storage
# オプション: より良い型操作のために pandas 統合と db-dtypes をインストール
pip install pandas db-dtypes pyarrow

Python がインストールされていることを確認してください (最新のライブラリ機能にはバージョン 3.7 以上を推奨)。

BigQuery クライアントライブラリの使用 (Python の例)

次に、google-cloud-bigquery Python ライブラリを使用して、一般的な BigQuery 操作を探ります。

1. クライアントのインポートと初期化:

最初に、ライブラリをインポートします。次に、クライアントインスタンスを作成します。ADC が正しく構成されている場合、クライアントは自動的に認証されます。

from google.cloud import bigquery
import pandas as pd

# BigQuery クライアントオブジェクトを構築します。
# GOOGLE_APPLICATION_CREDENTIALS が設定されている場合、サービスアカウントを使用します。
# gcloud auth application-default login が実行された場合、それらの認証情報を使用します。
# GCP インフラ上で実行している場合、それはインスタンスのサービス アカウントを使用します。
client = bigquery.Client()

# 必要に応じてプロジェクト ID を明示的に指定できますが、
# さもなければ環境/ADC の認証情報から自動的に推測されます。
# client = bigquery.Client(project='your-project-id')

print("クライアントが正常に作成されました。")

2. クエリの実行:

最も一般的な操作は、SQL クエリを実行することです。

シンプルな同期クエリ: 短時間で完了することが期待されるクエリのためのものです。 query_and_wait() はクエリを送信し、結果を待機します。

# SQL クエリを定義します
query = """
    SELECT name, SUM(number) as total_people
    FROM `bigquery-public-data.usa_names.usa_1910_2013`
    WHERE state = 'TX'
    GROUP BY name, state
    ORDER BY total_people DESC
    LIMIT 10
"""

# API リクエストを行い、ジョブが完了するのを待ちます。
query_job = client.query(query)  # API リクエスト
print(f"ジョブを開始しました: {query_job.job_id}")

# ジョブが完了するのを待ち、結果を取得します。
rows = query_job.result()  # ジョブが完了するまで待機します。

print("\nTXでの上位10名（1910-2013）:")
for row in rows:
    # 行の値にはフィールド名またはインデックスでアクセスできます。
    print(f"名前: {row.name}, カウント: {row['total_people']}") # 属性もしくはキーでアクセス

# 結果を Pandas DataFrame に変換します
df = rows.to_dataframe()
print("\nPandas DataFrame としての結果:")
print(df.head())

非同期クエリ: 長時間実行されるクエリのために、ジョブを開始し、後でそのステータスを確認します。

query = """
    SELECT corpus, COUNT(word) as distinct_words
    FROM `bigquery-public-data.samples.shakespeare`
    GROUP BY corpus
    ORDER BY distinct_words DESC;
"""
job_config = bigquery.QueryJobConfig(
    # クエリには標準SQL構文を使用します。
    use_legacy_sql=False
)

# 追加の構成を渡してクエリを開始します。
query_job = client.query(query, job_config=job_config) # 待たない
print(f"非同期ジョブを開始しました: {query_job.job_id}")

# --- アプリケーション内の後の部分 ---
# ジョブステータスを確認します (オプション)
# from google.cloud.exceptions import NotFound
# try:
#     job = client.get_job(query_job.job_id, location=query_job.location)
#     print(f"ジョブ {job.job_id} ステータス: {job.state}")
#     if job.state == "DONE":
#         if job.error_result:
#             print(f"ジョブが失敗しました: {job.errors}")
#         else:
#             results = job.result() # 結果を取得します
#             print("結果が取得されました。")
#             # 結果を処理...
# except NotFound:
#     print(f"ジョブ {query_job.job_id} が見つかりません。")

# 必要に応じて、単に完了を待機します
results = query_job.result() # ジョブが終了するまでブロックします
print("非同期ジョブが完了しました。")
for row in results:
    print(f"コーパス: {row.corpus}, 異なる単語: {row.distinct_words}")

パラメータ化クエリ: セキュリティのために重要です (ユーザー入力をクエリに組み込むときに SQL インジェクションを防ぐため)。

from google.cloud.bigquery import ScalarQueryParameter, ArrayQueryParameter, StructQueryParameter, QueryJobConfig

# 例: 特定の状態で特定の接頭辞から始まる名前を見つける
state_param = "NY"
prefix_param = "Ma"
min_count_param = 1000

query = """
    SELECT name, SUM(number) as total_people
    FROM `bigquery-public-data.usa_names.usa_1910_2013`
    WHERE state = @state_abbr AND name LIKE @name_prefix
    GROUP BY name
    HAVING total_people >= @min_count
    ORDER BY total_people DESC;
"""

job_config = QueryJobConfig(
    query_parameters=[
        ScalarQueryParameter("state_abbr", "STRING", state_param),
        # LIKE 演算子用の 'val%'
        ScalarQueryParameter("name_prefix", "STRING", f"{prefix_param}%"),
        ScalarQueryParameter("min_count", "INT64", min_count_param),
    ]
)

query_job = client.query(query, job_config=job_config)
print(f"パラメータ化されたクエリジョブを開始しました: {query_job.job_id}")

rows = query_job.result()

print(f"\n'{prefix_param}' で始まる {state_param} の名前 (人数>= {min_count_param}):")
for row in rows:
    print(f"名前: {row.name}, カウント: {row.total_people}")

3. データセットの管理:

データセットを作成、リスト、詳細取得、および削除できます。

# データセット ID とロケーションを定義します
project_id = client.project
dataset_id = f"{project_id}.my_new_dataset"
dataset_location = "US" # 例: "US", "EU", "asia-northeast1"

# API に送信するフルデータセットオブジェクトを構築します。
dataset = bigquery.Dataset(dataset_id)
dataset.location = dataset_location
dataset.description = "Python クライアントライブラリを介して作成されたデータセット"

try:
    # データセットを作成する API リクエストを実行します。
    dataset = client.create_dataset(dataset, timeout=30)  # API リクエストを実行します。
    print(f"{client.project}.{dataset.dataset_id} のデータセットを作成しました。")

    # プロジェクト内のデータセットのリストを表示します
    print("\nプロジェクト内のデータセット:")
    datasets = list(client.list_datasets()) # API リクエスト
    if datasets:
        for ds in datasets:
            print(f"\t{ds.dataset_id}")
    else:
        print(f"\t{client.project} プロジェクトにはデータセットが含まれていません。")

    # データセット情報を取得します
    retrieved_dataset = client.get_dataset(dataset_id) # API リクエスト
    print(f"\n{dataset_id} のデータセット情報を取得しました:")
    print(f"\t説明: {retrieved_dataset.description}")
    print(f"\tロケーション: {retrieved_dataset.location}")

except Exception as e:
    print(f"データセット操作中のエラー: {e}")

finally:
    # クリーンアップ: データセットを削除します
    try:
        client.delete_dataset(
            dataset_id, delete_contents=True, not_found_ok=True
        )  # API リクエスト
        print(f"\nデータセット '{dataset_id}' を正常に削除しました。")
    except Exception as e:
         print(f"データセット {dataset_id} の削除中にエラーが発生しました: {e}")

4. テーブルの管理:

テーブルに対しても、データセットと同様の操作が存在します: スキーマを定義してテーブルを作成、データをロード、メタデータを取得、テーブルを削除します。

# 以前に作成したデータセット ID を使用します (存在することを確認するか、上記の削除ステップを削除します)
dataset_id_for_table = "my_new_dataset" # 有効なデータセット ID を使用します
table_id = f"{client.project}.{dataset_id_for_table}.my_new_table"

# スキーマを定義します
schema = [
    bigquery.SchemaField("full_name", "STRING", mode="REQUIRED"),
    bigquery.SchemaField("age", "INTEGER", mode="REQUIRED"),
    bigquery.SchemaField("email", "STRING", mode="NULLABLE"),
]

# テーブルを作成します
table = bigquery.Table(table_id, schema=schema)
try:
    # まずデータセットが存在することを確認します
    client.create_dataset(dataset_id_for_table, exists_ok=True)

    table = client.create_table(table)  # API リクエスト
    print(
        f"{table.project}.{table.dataset_id}.{table.table_id} のテーブルを作成しました。"
    )

    # --- データのロード (例: Pandas DataFrame から) ---
    data = {'full_name': ['Alice Smith', 'Bob Johnson'],
            'age': [30, 45],
            'email': ['alice@example.com', None]}
    dataframe = pd.DataFrame(data)

    job_config = bigquery.LoadJobConfig(
        # スキーマを指定することを推奨します。適切な型を保証します。
        schema=schema,
        # オプション: テーブルデータを上書きします
        write_disposition="WRITE_TRUNCATE",
        # または追加: write_disposition="WRITE_APPEND",
    )

    load_job = client.load_table_from_dataframe(
        dataframe, table_id, job_config=job_config
    )  # API リクエスト
    print(f"DataFrame からデータをロードするためのジョブ {load_job.job_id} を開始しています。")

    load_job.result()  # ジョブが完了するまで待機します。
    print("DataFrame のロードジョブが終了しました。")

    destination_table = client.get_table(table_id) # API リクエスト
    print(f"テーブル {table_id} に {destination_table.num_rows} 行をロードしました。")

    # --- データのロード (例: Google Cloud Storage URI から) ---
    # gs://your-bucket/data.csv に互換性のあるデータを持つ CSV ファイルが存在することを想定します
    # uri = "gs://your-bucket/data.csv"
    # job_config_gcs = bigquery.LoadJobConfig(
    #     schema=schema,
    #     skip_leading_rows=1, # ヘッダー行をスキップ
    #     source_format=bigquery.SourceFormat.CSV,
    #     write_disposition="WRITE_APPEND", # 既存データに追加
    # )
    # load_job_gcs = client.load_table_from_uri(
    #     uri, table_id, job_config=job_config_gcs
    # )
    # print(f"GCS からデータをロードするためのジョブ {load_job_gcs.job_id} を開始しています。")
    # load_job_gcs.result()
    # print("GCS のロードジョブが終了しました。")
    # destination_table = client.get_table(table_id)
    # print(f"GCS ロード後の合計行数: {destination_table.num_rows}")

except Exception as e:
    print(f"テーブル操作中のエラー: {e}")

finally:
    # クリーンアップ: テーブルを削除します
    try:
        client.delete_table(table_id, not_found_ok=True)  # API リクエスト
        print(f"テーブル '{table_id}' を正常に削除しました。")
        # この例のためだけにデータセットを再度削除することもできます
        # client.delete_dataset(dataset_id_for_table, delete_contents=True, not_found_ok=True)
        # print(f"データセット '{dataset_id_for_table}' を正常に削除しました。")
    except Exception as e:
        print(f"テーブル {table_id} の削除中にエラーが発生しました: {e}")

5. ジョブの操作:

すべての非同期操作 (クエリ、ロード、エクスポート、コピー) はジョブリソースを作成します。これらのジョブをリストおよび管理できます。

# 最近のジョブをリスト表示します
print("\n最近の BigQuery ジョブ:")
for job in client.list_jobs(max_results=10): # API リクエスト
    print(f"ジョブ ID: {job.job_id}, タイプ: {job.job_type}, ステータス: {job.state}, 作成日: {job.created}")

# 特定のジョブを取得します (以前の実行から有効なジョブ ID に置き換えます)
# try:
#     job_id_to_get = "..." # 実際のジョブ ID に置き換えます
#     location = "US"      # デフォルトでない場合はそのジョブのロケーションに置き換えます
#     retrieved_job = client.get_job(job_id_to_get, location=location) # API リクエスト
#     print(f"\nジョブ {retrieved_job.job_id} の詳細:")
#     print(f"\tステータス: {retrieved_job.state}")
#     if retrieved_job.error_result:
#         print(f"\tエラー: {retrieved_job.errors}")
# except NotFound:
#     print(f"ジョブ {job_id_to_get} が見つかりません。")
# except Exception as e:
#     print(f"ジョブの取得中にエラーが発生しました: {e}")

専門化された API の活用 (コンセプトとユースケース)

コアクライアントライブラリは多くのユースケースをカバーしていますが、専門の API は特定のタスクのためのパフォーマンスや機能を向上させます。

1. BigQuery ストレージ読み取り API (Python):

目的: BigQuery テーブルからの非常に迅速なデータ取得を実現し、標準のクエリ実行エンジンをバイパスします。大規模なデータセットをエクスポートしたり、ML トレーニングパイプラインやデータ処理フレームワークにデータを供給したりするのに最適です。
動作原理: テーブル、カラム、および行フィルタを指定して ReadSession を作成します。API は 1 つ以上の Streams を返します。これらのストリームからデータを読み取ります (通常は Apache Arrow または Avro フォーマット) し、通常は並行して行います。
クライアントライブラリ: google-cloud-bigquery-storage
概念的な Python 使用例 (Pandas 統合):

# 必要なライブラリ: pip install google-cloud-bigquery-storage pyarrow pandas db-dtypes

from google.cloud import bigquery_storage_v1
from google.cloud.bigquery_storage_v1 import types, GetDataStreamRequest

# --- Pandas read_gbq を使用する (最も簡単な統合) ---
# インストールされている場合、ストレージ API を自動的に使用します。
# table_id_read = "bigquery-public-data.usa_names.usa_1910_2013"
# cols_to_read = ["name", "number", "state"]
# row_filter = "state = 'CA' AND number > 5000"
#
# try:
#      df_storage = pd.read_gbq(
#          table_id_read,
#          project_id=client.project,
#          columns=cols_to_read,
#          row_filter=row_filter,
#          use_bqstorage_api=True, # 明示的にストレージ API を要求します
#          progress_bar_type='tqdm' # オプションのプログレスバー
#      )
#      print("\nStorage API を介して pandas.read_gbq を使用してデータを読み取りました:")
#      print(df_storage.head())
#      print(f"{len(df_storage)} 行を読み取りました。")
# except Exception as e:
#      print(f"Storage API による read_gbq でエラーが発生しました: {e}")


# --- 手動ストレージ API の使用 (より多くの制御) ---
# bqstorageclient = bigquery_storage_v1.BigQueryReadClient()
# table = f"projects/{project_id}/datasets/{dataset_id}/tables/{table_name}" # あなたのテーブル詳細に置き換えます

# requested_session = types.ReadSession(
#     table=table,
#     data_format=types.DataFormat.ARROW,
#     read_options=types.ReadSession.TableReadOptions(
#         selected_fields=["col1", "col2"], # 列を指定します
#         row_restriction="col1 > 100"     # フィルタを指定します
#     ),
# )
# parent = f"projects/{project_id}"

# read_session = bqstorageclient.create_read_session(
#     parent=parent,
#     read_session=requested_session,
#     max_stream_count=1, # 並列ストリームの数を要求します
# )

# stream = read_session.streams[0]
# reader = bqstorageclient.read_rows(stream.name)
# frames = [message.arrow_record_batch for message in reader.messages()]
# if frames:
#     arrow_table = pa.Table.from_batches(frames)
#     df_manual = arrow_table.to_pandas()
#     print("\nStorage API を使用して手動でデータを読み取りました:")
#     print(df_manual.head())
# else:
#     print("手動ストレージ API を使用して読み取るデータはありませんでした。")

2. BigQuery 接続 API:

目的: データの移動なしに、BigQuery から外部データソース (Cloud SQL、Spanner、Cloud Storage) を直接クエリします。
動作原理:

API (または Cloud Console/bq ツール) を使用して、外部ソースタイプと詳細を指定して Connection リソースを作成します。
接続のサービスアカウントに外部リソースに対する適切な権限を付与します (例: Cloud SQL ユーザーロール)。
BigQuery SQL 内で EXTERNAL_QUERY("connection_id", "external_sql_query") 関数を使用します。

クライアントライブラリ: google-cloud-bigquery-connection は、create_connection、get_connection、list_connections、delete_connection のようなメソッドを提供しています。

3. Analytics Hub API:

目的: 精選されたデータセットを組織やチーム全体で安全に共有します。
主要コンセプト:
データ交換: データを共有するためのコンテナ。パブリックまたはプライベートにできます。
リスティング: 交換内で共有される特定の BigQuery データセット (または将来的にはビュー/テーブルの可能性もあります) への参照です。発行者がリスティングを作成し、購読者がそれにアクセスします。
API: 交換、リスティング、サブスクリプションをプログラム的に管理するために、REST およびクライアントライブラリ (例: google-cloud-analyticshub) を介して使用できます。

4. BigLake API:

目的: BigLake テーブルを管理し、BigQuery インターフェースとガバナンスを使用して Cloud Storage 内のデータをクエリできるようにします。
動作原理: GCS バケット内のデータファイル (Parquet、ORC など) を参照する BigQuery 内にテーブルを定義します。 BigLake API (現在は主に REST) は、これらのテーブルのメタデータを作成、更新、および管理するために使用されます。クエリは標準 BigQuery SQL を介して行われます。

REST API を直接使用する

クライアントライブラリが一般的に推奨されますが、次の理由から REST API を直接使用する場合があります：

公式の Google クライアントライブラリがない言語を使用している場合。
ライブラリでまだ公開されていない先端機能へのアクセスが必要な場合。
HTTP リクエスト/レスポンスサイクルを絶対に制御する必要がある場合。

リクエストの作成:

通常は、HTTP クライアント (例: curl や Python の requests ライブラリ) を使用します。必要なステップは：

OAuth 2.0 アクセストークンを取得します (例: gcloud auth print-access-token を使用)。
正しい API エンドポイント URL を構築します。
API メソッドの仕様に従って JSON リクエスト本文を作成します。
アクセストークンを Authorization: Bearer <token> ヘッダーに含めます。
HTTP レスポンスを処理します (ステータスコード、JSON 解析、エラーメッセージ)。

例: REST を介してクエリを実行する (jobs.query)

# 1. アクセストークンを取得します
TOKEN=$(gcloud auth print-access-token)

# 2. プロジェクト ID とリクエストボディを定義します
PROJECT_ID="your-project-id" # プロジェクト ID に置き換えます
REQUEST_BODY=$(cat <<EOF
{
  "query": "SELECT name, SUM(number) as total_people FROM \`bigquery-public-data.usa_names.usa_1910_2013\` WHERE state = 'CA' GROUP BY name ORDER BY total_people DESC LIMIT 5;",
  "useLegacySql": false
}
EOF
)

# 3. curl を使用して API コールを行います
curl -X POST \
  "https://bigquery.googleapis.com/bigquery/v2/projects/${PROJECT_ID}/jobs" \
  -H "Authorization: Bearer ${TOKEN}" \
  -H "Content-Type: application/json; charset=utf-8" \
  -d "${REQUEST_BODY}"

# レスポンスには、ジョブ情報が含まれ、ジョブ ID も表示されます。
# ジョブが完了した後、実際のデータ行を取得するために jobs.getQueryResults を使用して、ジョブ ID への次の呼び出しを行う必要があります。

この例は、クエリジョブを開始するだけです。結果を取得するには、ジョブのステータスをポーリングし、その後 jobs.getQueryResults エンドポイントを呼び出す必要があります。これは、クライアントライブラリと比較して、追加のステップが関与することを強調しています。

ベストプラクティスとヒント

クライアントライブラリを優先: シンプルさ、堅牢性、メンテナンスの容易さのために、可能な限り Google Cloud クライアントライブラリを使用してください。
ADC を使用: 認証にアプリケーションデフォルトの認証情報を使用することで、異なる環境間での展開が簡素化されます。コードに認証情報を埋め込まないようにしてください。
パラメータ化クエリ: 外部入力を組み込むときは、常にパラメータ化クエリを使用して SQL インジェクションの脆弱性を防ぎます。
クエリの最適化: 効率的な SQL を記述します。SELECT * を避け、WHERE 句を効果的に使用し、大きなテーブルのパフォーマンスのためのパーティショニングとクラスターの理解を持ち、クエリプランの説明機能を使用します。
エラーハンドリング: ネットワークの問題や API 特有のエラー (例: クォータ超過、未発見) のために、堅牢なエラーハンドリングを実装します。クライアントライブラリには一時的なエラーのためのビルトインのリトライメカニズムが含まれていることがよくあります。
リソース管理: テストまたは一時的な処理中に作成されたデータセットやテーブルを明示的に削除して、不必要なストレージコストを回避します。
コストとクォータを監視: BigQuery の価格設定 (クエリの処理されたバイト数、ストレージコスト) に注意してください。使用状況を監視し、API クォータ (例: 同時クエリ制限、API リクエストレート制限) を把握してください。最大バイト請求 ( maximum_bytes_billed ) を使用してクエリコストを制御します。
専門の API を使用: 大規模データのエクスポート/読み取りにはストレージ読み取り API を活用し、外部ソースの効率的なクエリには接続 API を活用します。

結論

BigQuery API は、データウェアハウスとプログラム的に対話するための強力で柔軟な方法を提供します。データの自動化、アプリケーションの一部としての複雑な分析クエリの実行、ダッシュボードへの BigQuery のインサイトの統合、またはリソースの動的管理が必要な場合、API は必要なツールを提供します。

さまざまな API タイプを理解し、環境を正しくセットアップし、クライアントライブラリ (特に Python において) の便利さを活用することにより、インタラクティブコンソールを超えて BigQuery のフルポテンシャルを引き出すことができます。一般的なタスクにはコア API を使用し、パフォーマンス上重要な操作のためにストレージ読み取り API などの専門 API を探索し、REST API は常に利用可能であることを忘れないでください。BigQuery を使用してアプリケーションを構築する際、これらの API はデータアーキテクチャの重要なコンポーネントになります。