BigQuery Storage API を使用して BigQuery データを pandas にダウンロードする

BigQuery Storage API を使用すると、BigQuery に保存されたデータにすばやくアクセスできます。BigQuery に保存されたデータを Python 用 pandas ライブラリなどの分析ツールで使用するには、BigQuery Storage API を使用してこれらのデータをダウンロードします。

目標

このチュートリアルでは、次のことができます。

• Jupyter ノートブック内の BigQuery 用 IPython マジックから BigQuery Storage API を使用して、クエリ結果を pandas DataFrame にダウンロードする。
• Python 用 BigQuery クライアント ライブラリを使用して、クエリ結果を pandas DataFrame にダウンロードする。
• Python 用 BigQuery クライアント ライブラリを使用して、BigQuery テーブルのデータを pandas DataFrame にダウンロードする。
• Python 用 BigQuery Storage API クライアント ライブラリを使用して、BigQuery テーブルのデータを pandas DataFrame にダウンロードする。

料金

BigQuery は有料プロダクトとなっているため、クエリを実行すると BigQuery の使用料金が発生します。処理されるクエリデータは毎月 1 TB まで無料です。詳細については、BigQuery の料金ページをご覧ください。

BigQuery Storage API は有料プロダクトとなっているため、DataFrame をダウンロードするときにスキャンするテーブルデータに対して使用料金が発生します。詳細については、BigQuery の料金ページをご覧ください。

始める前に

このチュートリアルを始める前に、Google Cloud Console を使用して、プロジェクトを作成または選択し、課金を有効にします。

1. Google アカウントにログインします。

   Google アカウントをまだお持ちでない場合は、新しいアカウントを登録します。

2. GCP Console のプロジェクト セレクタのページで、GCP プロジェクトを選択または作成します。

   注: この手順で作成するリソースをそのまま保持する予定でない場合、既存のプロジェクトを選択するのではなく、プロジェクトを作成してください。チュートリアルの完了後にそのプロジェクトを削除すれば、プロジェクトに関連するすべてのリソースを削除できます。

   プロジェクト セレクタのページに移動

3. Google Cloud Platform プロジェクトに対して課金が有効になっていることを確認します。 プロジェクトに対して課金が有効になっていることを確認する方法を学習する。

4. 新しいプロジェクトでは、BigQuery が自動的に有効になります。既存のプロジェクトで BigQuery を有効にするには、 BigQuery, BigQuery Storage API必要な API を有効にします。

   API を有効にする

5. Python 開発環境を設定します。
   Python を設定する
6. 開発環境に対する認証を設定します。
   認証を設定する

このチュートリアルを始める前に、BigQuery 用 IPython マジック コマンド、BigQuery クライアント ライブラリ、pandas でのクライアント ライブラリの使用方法についても理解しておく必要があります。

クライアント ライブラリのインストール

BigQuery Python クライアント ライブラリ バージョン 1.9.0 以降と BigQuery Storage API Python クライアント ライブラリをインストールします。

PIP

google-cloud-bigquery パッケージと google-cloud-bigquery-storage パッケージをインストールします。

pip install --upgrade google-cloud-bigquery[bqstorage,pandas]

Conda

コミュニティが運営する conda-forge チャンネルから BigQuery パッケージと BigQuery Storage API Conda パッケージをインストールします。

conda install -c conda-forge google-cloud-bigquery \
  google-cloud-bigquery-storage \
  pandas \
  pyarrow

BigQuery 用 IPython マジック コマンドを使用してクエリ結果をダウンロードする

Jupyter ノートブック サーバーを起動し、新しい Jupyter ノートブックを作成します。%load_ext マジック コマンドを使用して、BigQuery 用 IPython マジック コマンドを読み込みます。

%load_ext google.cloud.bigquery

BigQuery Storage API でサイズの大きいクエリ結果をダウンロードするには、%%bigquery マジック コマンドに --use_bq_storage_api 引数を追加します。

%%bigquery tax_forms --use_bqstorage_api
SELECT * FROM `bigquery-public-data.irs_990.irs_990_2012`

小さいサイズのクエリ結果にこの引数を使用すると、BigQuery API を使用して結果がダウンロードされます。

%%bigquery stackoverflow --use_bqstorage_api
SELECT
  CONCAT(
    'https://stackoverflow.com/questions/',
    CAST(id as STRING)) as url,
  view_count
FROM `bigquery-public-data.stackoverflow.posts_questions`
WHERE tags like '%google-bigquery%'
ORDER BY view_count DESC
LIMIT 10

BigQuery Storage API がデフォルトで使用されるようにするには、context.use_bqstorage_api プロパティを True に設定します。

import google.cloud.bigquery.magics

google.cloud.bigquery.magics.context.use_bqstorage_api = True

context.use_bqstorage_api プロパティを設定した後、引数を追加せずに %%bigquery マジックを実行すると、BigQuery Storage API を使用してサイズの大きい結果がダウンロードされます。

%%bigquery tax_forms
SELECT * FROM `bigquery-public-data.irs_990.irs_990_2012`

Python クライアント ライブラリを使用する

Python クライアントを作成する

次のコードを使用して、BigQuery Client オブジェクトと BigQueryStorageClient を作成します。

GitHub で表示 フィードバック

import google.auth
from google.cloud import bigquery
from google.cloud import bigquery_storage_v1beta1

# Explicitly create a credentials object. This allows you to use the same
# credentials for both the BigQuery and BigQuery Storage clients, avoiding
# unnecessary API calls to fetch duplicate authentication tokens.
credentials, your_project_id = google.auth.default(
    scopes=["https://www.googleapis.com/auth/cloud-platform"]
)

# Make clients.
bqclient = bigquery.Client(
    credentials=credentials,
    project=your_project_id,
)
bqstorageclient = bigquery_storage_v1beta1.BigQueryStorageClient(
    credentials=credentials
)

google-auth Python ライブラリを使用して、両方の API に十分に対応できる認証情報を作成します。認証が 2 回行われないよう、それぞれのコンストラクタに認証情報オブジェクトを渡します。

BigQuery クライアント ライブラリを使用してクエリ結果をダウンロードする

query メソッドを使用してクエリを実行します。to_dataframe メソッドを呼び出し、クエリが完了して結果が BigQuery Storage API によってダウンロードされるまで待ちます。

GitHub で表示 フィードバック

# Download query results.
query_string = """
SELECT
CONCAT(
    'https://stackoverflow.com/questions/',
    CAST(id as STRING)) as url,
view_count
FROM `bigquery-public-data.stackoverflow.posts_questions`
WHERE tags like '%google-bigquery%'
ORDER BY view_count DESC
"""

dataframe = (
    bqclient.query(query_string)
    .result()
    .to_dataframe(bqstorage_client=bqstorageclient)
)
print(dataframe.head())

BigQuery クライアント ライブラリを使用してテーブルデータをダウンロードする

list_rows メソッドを使用してテーブル内の行をすべてダウンロードします。このメソッドは RowIterator オブジェクトを返します。bqstorage_client 引数を指定して to_dataframe メソッドを呼び出すことで、BigQuery Storage API を使用して行をダウンロードします。

GitHub で表示 フィードバック

# Download a table.
table = bigquery.TableReference.from_string(
    "bigquery-public-data.utility_us.country_code_iso"
)
rows = bqclient.list_rows(
    table,
    selected_fields=[
        bigquery.SchemaField("country_name", "STRING"),
        bigquery.SchemaField("fips_code", "STRING"),
    ],
)
dataframe = rows.to_dataframe(bqstorage_client=bqstorageclient)
print(dataframe.head())

BigQuery Storage API クライアント ライブラリを使用してテーブルデータをダウンロードする

フィルタや並列処理のきめ細かな制御を直接行うには、BigQuery Storage API クライアント ライブラリを使用します。シンプルな行フィルタのみが必要な場合は、クエリの代わりに BigQuery Storage API 読み取りセッションを使用できます。

読み取り対象のテーブルに関する TableReference オブジェクトを作成します。列の選択や行のフィルタ処理を行う TableReadOptions オブジェクトを作成します。create_read_session メソッドを使用して読み取りセッションを作成します。

セッションにストリームが存在する場合は、read_rows メソッドを使用して、そのストリームから行の読み取りを開始します。その読み取りに対して to_dataframe メソッドを呼び出し、ストリーム全体を pandas DataFrame に書き込みます。パフォーマンスを向上させるには、複数のストリームから並列に読み取りますが、このコード例では単純にするために単一のストリームから読み取ります。

GitHub で表示 フィードバック

table = bigquery_storage_v1beta1.types.TableReference()
table.project_id = "bigquery-public-data"
table.dataset_id = "new_york_trees"
table.table_id = "tree_species"

# Select columns to read with read options. If no read options are
# specified, the whole table is read.
read_options = bigquery_storage_v1beta1.types.TableReadOptions()
read_options.selected_fields.append("species_common_name")
read_options.selected_fields.append("fall_color")

parent = "projects/{}".format(your_project_id)
session = bqstorageclient.create_read_session(
    table,
    parent,
    read_options=read_options,
    # This API can also deliver data serialized in Apache Avro format.
    # This example leverages Apache Arrow.
    format_=bigquery_storage_v1beta1.enums.DataFormat.ARROW,
    # We use a LIQUID strategy in this example because we only read from a
    # single stream. Consider BALANCED if you're consuming multiple streams
    # concurrently and want more consistent stream sizes.
    sharding_strategy=(
        bigquery_storage_v1beta1.enums.ShardingStrategy.LIQUID
    ),
)

# This example reads from only a single stream. Read from multiple streams
# to fetch data faster. Note that the session may not contain any streams
# if there are no rows to read.
stream = session.streams[0]
position = bigquery_storage_v1beta1.types.StreamPosition(stream=stream)
reader = bqstorageclient.read_rows(position)

# Parse all Avro blocks and create a dataframe. This call requires a
# session, because the session contains the schema for the row blocks.
dataframe = reader.to_dataframe(session)
print(dataframe.head())

すべての例のソースコード

クライアント ライブラリのすべての例の完全なソースコードをご覧ください。

GitHub で表示 フィードバック

import google.auth
from google.cloud import bigquery
from google.cloud import bigquery_storage_v1beta1

# Explicitly create a credentials object. This allows you to use the same
# credentials for both the BigQuery and BigQuery Storage clients, avoiding
# unnecessary API calls to fetch duplicate authentication tokens.
credentials, your_project_id = google.auth.default(
    scopes=["https://www.googleapis.com/auth/cloud-platform"]
)

# Make clients.
bqclient = bigquery.Client(
    credentials=credentials,
    project=your_project_id,
)
bqstorageclient = bigquery_storage_v1beta1.BigQueryStorageClient(
    credentials=credentials
)
# Download a table.
table = bigquery.TableReference.from_string(
    "bigquery-public-data.utility_us.country_code_iso"
)
rows = bqclient.list_rows(
    table,
    selected_fields=[
        bigquery.SchemaField("country_name", "STRING"),
        bigquery.SchemaField("fips_code", "STRING"),
    ],
)
dataframe = rows.to_dataframe(bqstorage_client=bqstorageclient)
print(dataframe.head())
# Download query results.
query_string = """
SELECT
CONCAT(
    'https://stackoverflow.com/questions/',
    CAST(id as STRING)) as url,
view_count
FROM `bigquery-public-data.stackoverflow.posts_questions`
WHERE tags like '%google-bigquery%'
ORDER BY view_count DESC
"""

dataframe = (
    bqclient.query(query_string)
    .result()
    .to_dataframe(bqstorage_client=bqstorageclient)
)
print(dataframe.head())
table = bigquery_storage_v1beta1.types.TableReference()
table.project_id = "bigquery-public-data"
table.dataset_id = "new_york_trees"
table.table_id = "tree_species"

# Select columns to read with read options. If no read options are
# specified, the whole table is read.
read_options = bigquery_storage_v1beta1.types.TableReadOptions()
read_options.selected_fields.append("species_common_name")
read_options.selected_fields.append("fall_color")

parent = "projects/{}".format(your_project_id)
session = bqstorageclient.create_read_session(
    table,
    parent,
    read_options=read_options,
    # This API can also deliver data serialized in Apache Avro format.
    # This example leverages Apache Arrow.
    format_=bigquery_storage_v1beta1.enums.DataFormat.ARROW,
    # We use a LIQUID strategy in this example because we only read from a
    # single stream. Consider BALANCED if you're consuming multiple streams
    # concurrently and want more consistent stream sizes.
    sharding_strategy=(
        bigquery_storage_v1beta1.enums.ShardingStrategy.LIQUID
    ),
)

# This example reads from only a single stream. Read from multiple streams
# to fetch data faster. Note that the session may not contain any streams
# if there are no rows to read.
stream = session.streams[0]
position = bigquery_storage_v1beta1.types.StreamPosition(stream=stream)
reader = bqstorageclient.read_rows(position)

# Parse all Avro blocks and create a dataframe. This call requires a
# session, because the session contains the schema for the row blocks.
dataframe = reader.to_dataframe(session)
print(dataframe.head())

クリーンアップ

このチュートリアルで使用したリソースについて、Google Cloud Platform アカウントに課金されないようにする手順は次のとおりです。

プロジェクトを削除します。このチュートリアルでは BigQuery リソースを作成しませんでしたが、プロジェクトを削除すると、作成した他のすべてのリソースも削除されます。

注意: プロジェクトを削除すると、次のような影響があります。

• プロジェクト内のすべてのものが削除されます。このチュートリアルで既存のプロジェクトを使用した場合、それを削除すると、そのプロジェクトで行った他の作業もすべて削除されます。
• カスタム プロジェクト ID が失われます。このプロジェクトを作成したときに、将来使用するカスタム プロジェクト ID を作成した可能性があります。そのプロジェクト ID を使用した URL（たとえば、appspot.com）を保持するには、プロジェクト全体ではなくプロジェクト内の選択したリソースだけを削除します。

複数のチュートリアルとクイックスタートを検討する予定がある場合は、プロジェクトを再利用すると、プロジェクトの割り当て制限を超えないようにすることができます。

1. GCP Console で [リソースの管理] ページに移動します。

   [リソースの管理] ページに移動

2. プロジェクト リストで、削除するプロジェクトを選択し、[削除] をクリックします。
3. ダイアログでプロジェクト ID を入力し、[シャットダウン] をクリックしてプロジェクトを削除します。

次のステップ

• Python クライアント ライブラリのリファレンスを参照する -
  • Python 用 BigQuery クライアント ライブラリのリファレンス
  • Python 用 BigQuery Storage API クライアント ライブラリのリファレンス