RevComm Tech Blog

コミュニケーションを再発明し 人が人を想う社会を創る

Python + BigQuery の始め方

Python GCP データベース

はじめに

この記事は 2022 年の RevComm アドベントカレンダー 6 日目の記事です。

こんにちは。株式会社RevCommで社内向けシステム開発・運用を担当している Machida Kensuke です。

みなさんは GCP を利用したサービスの開発をされることはありますか？RevComm では基本的にクラウド基盤として AWS を採用していますが、一部では GCP も採用しています。私はデータ基盤として BigQuery を利用しているサービスの開発に参加しました。

BigQuery は他のデータウェアハウスと比較してもコスト効率が良い課金形態で、社内システムでも通話料計算やデータ分析等の用途で使用しています。

環境構築に必要なものが多岐にわたるため、開発の導入部分でとても手こずりました。

この経験から、初めて GCP や BigQuery を触る方でもすぐに着手できるようになったらいいなと思い、この記事を書くことにしました。

手順

  1. 前提条件
  2. 準備
    1. クライアントライブラリをインストール
    2. Cloud Platform プロジェクトを作成
    3. 認証設定
  3. クライアントを初期化
  4. データセットを作成
  5. テーブルを作成
  6. BigQuery APIを使用したテーブルの CRUD 操作

やってみる

では早速やっていきましょう！

前提条件

Python はインストール済みとします。

準備

BigQuery API の Cloud クライアント ライブラリをインストールする

公式のクライアントライブラリを使用します。

※2022/12/06現在、対応の Python バージョンは3.7～3.10 です。

pip install google-cloud-bigquery

Cloud Platform プロジェクトを作成

「プロジェクトを作成」からプロジェクトを作成します。 プロジェクトでは、API の管理、課金の有効化、共同編集者の追加と削除、Google Cloud リソースに対する権限の管理などを行います。

認証設定

Python から作成したプロジェクトへ API アクセスできるように認証を設定します。

  1. API へのアクセスの有効化からAPI アクセスを有効化します。
  2. Google Cloud のホームページの手順にしたがって、サービスアカウントを作成します。
  3. Google Cloud のホームページの手順にしたがって、サービスアカウントキーを作成して service-account-key-file.json をダウンロードします。この json ファイルを GCP の認証に使用します。

service-account-key-file.json

クライアントを初期化

設定が完了したので、実際に Python から BigQuery を動かしてみましょう！

まずは、下記のコマンドを実行してファイルを作成します。

touch create_dataset.py

作成した create_dataset.py にライブラリのインポートと認証情報を使用して BigQuery クライアントを初期化します。

from google.cloud import bigquery
from google.oauth2 import service_account

# 前ステップでダウンロードした service-account-key-file.jsonのフルパスを指定
key_path = "/path/to/service-account-key-file.json"
credentials = service_account.Credentials.from_service_account_file(key_path, scopes=["https://www.googleapis.com/auth/cloud-platform"])
client = bigquery.Client(credentials=credentials, project=credentials.project_id)

今回は、google.oauth2.service_account.Credentials.from_service_account_file を使用して、サービスアカウントキーファイルによる認証を行います。

データセットを作成

次に、データセットを作成します。 BigQuery におけるデータセットとは「テーブルを作成するために必要な箱」のようなものです。

参考: データセットの概要 | Google Cloud

dataset_id = "{}.sample_dataset".format(client.project)
dataset = bigquery.Dataset(dataset_id)
dataset.location = "asia-northeast1"
client.create_dataset(dataset)

前述したクライアントの初期化を含めたソースコード全体は下記のとおりです。

from google.cloud import bigquery
from google.oauth2 import service_account

def init_client() -> bigquery.Client:
    key_path = "/path/to/service-account-key-file.json"
    credentials = service_account.Credentials.from_service_account_file(
        key_path, scopes=["https://www.googleapis.com/auth/cloud-platform"]
    )
    return bigquery.Client(credentials=credentials, project=credentials.project_id)

def create_dataset(client: bigquery.Client):
    dataset_id = "{}.sample_dataset".format(client.project)
    dataset = bigquery.Dataset(dataset_id)
    dataset.location = "asia-northeast1"
    client.create_dataset(dataset)

if __name__ == "__main__":
    client = init_client()
    create_dataset(client)

Python コマンドから上記のコードを実行します。

$ python3 create_dataset.py

Cloud コンソールから作成したデータセットを確認して、sample_dataset というデータセットが作成されていれば OK です。

データセット情報

テーブルを作成

次に、作成したデータセットの中にテーブルを作成します。 テーブルは、先ほど作成したデータセット ID とスキーマを定義することで作成できます。

from google.cloud import bigquery
from google.oauth2 import service_account

def create_table(client: bigquery.Client):
    dataset_id = "{}.sample_dataset".format(client.project)
    dataset = client.get_dataset(dataset_id)

    table_id = "{}.{}.sample_table_name".format(client.project, dataset.dataset_id)
    schema = [
        bigquery.SchemaField("name", "STRING", mode="REQUIRED"),
        bigquery.SchemaField("age", "INTEGER", mode="NULLABLE"),
    ]
    table = bigquery.Table(table_id, schema=schema)
    client.create_table(table)

if __name__ == "__main__":
    # init_client() は前述のものを使用
    client = init_client()
    create_table(client)

上記のコードを実行して、Cloud コンソールから sample_table_name という名前の空テーブルが作成されていれば OK です。定義したスキーマも設定されています。

テーブル情報

BigQuery APIを使用したテーブルの CRUD 操作

空のテーブルが作成できたところで、テーブル上のレコードを CRUD 操作してみます。

データの登録（Create）

まずはデータの登録から。 方法は色々ありますが、今回は Python でよく使用される Pandas DataFrame のデータをテーブルに登録します。

参考: テーブルデータの管理 | Google Cloud

データの登録は、先ほど作成したテーブルのスキーマを job_config に定義することで行えます。BigQuery のジョブとは、データの読み込み、データのエクスポート、データのクエリ、データのコピーなど、BigQuery がユーザーに代わって実行するアクションのことです。

参考: BigQuery ジョブの概要 | Google Cloud

from google.cloud import bigquery
from google.oauth2 import service_account

def create_data(client: bigquery.Client):
    dataset_id = "{}.sample_dataset".format(client.project)
    dataset = client.get_dataset(dataset_id)
    table_id = "{}.{}.sample_table_name".format(client.project, dataset.dataset_id)

    schema = [
        bigquery.SchemaField("name", bigquery.enums.SqlTypeNames.STRING, mode="REQUIRED"),
        bigquery.SchemaField("age", bigquery.enums.SqlTypeNames.INTEGER, mode="NULLABLE"),
    ]
    job_config = bigquery.LoadJobConfig(schema=schema)
    dataframe = pd.DataFrame(
        [
            {
                "name": "Machida",
                "age": 2,
            },
            {
                "name": "“Kensuke”",
                "age": 5,
            },
        ]
    )
    job = client.load_table_from_dataframe(dataframe, table_id, job_config=job_config)
    job.result()

if __name__ == "__main__":
    client = init_client()
    create_data(client)

上記のコードを実行して、Cloud コンソールから作成したテーブル sample_table_name に 2 行 2 列のレコードが作成されていれば OK です。

テーブルのプレビュー

データの取得（Read）

続いて、データの取得です。 BigQuery では標準 SQL によるクエリ実行が可能です。 データの取得は、SELECT 句を使用して BigQuery クライアント ライブラリのクエリジョブで取得します。

先ほど作成した {"name": "Machida", "age": 2} と {"name": "Kensuke", "age": 5} の 2 レコードを取得します。
pd.to_dataframe() メソッドを使用すると Pandas DataFrame のデータとして取得できます。

from google.cloud import bigquery
from google.oauth2 import service_account

def read_data(client: bigquery.Client):
    dataset_id = "{}.sample_dataset".format(client.project)
    dataset = client.get_dataset(dataset_id)
    table_id = "{}.{}.sample_table_name".format(client.project, dataset.dataset_id)

    query = f"""
    SELECT
        *
    FROM 
        `{table_id}`
    LIMIT 2;
    """
    dataframe = (
        client.query(query)
        .result()
        .to_dataframe(
            create_bqstorage_client=True,
        )
    )
    print(dataframe)

if __name__ == "__main__":
    client = init_client()
    read_data(client)

上記のコードを実行して、{"name": "Machida", "age": 2} と {"name": "Kensuke", "age": 5} の Pandas DataFrame 型データを取得できていれば OK です。

データの取得

データの更新（Update）

続いて、データの更新です。データの更新は UPDATE 句を使用したクエリジョブを使用します。

先ほど作成した "name": "Machida", "age": 2 レコードの name を Yamada に更新します。

from google.cloud import bigquery
from google.oauth2 import service_account

def update_data(client: bigquery.Client):
    dataset_id = "{}.sample_dataset".format(client.project)
    dataset = client.get_dataset(dataset_id)
    table_id = "{}.{}.sample_table_name".format(client.project, dataset.dataset_id)

    query = f"""
    UPDATE
        `{table_id}`
    SET
        name = "Yamada"
    WHERE
        name = "Machida";
    """
    client.query(query).result()

if __name__ == "__main__":
    client = init_client()
    update_data(client)

上記のコードを実行して、Cloud コンソールから sample_table_name テーブルの {"name": "Machida", "age": 2} レコードが {"name": "Yamada", "age": 2} レコードに更新されていれば OK です。

データの削除（Delete）

最後は、データの削除です。データの削除は DELETE 句を使用したクエリジョブを使用します。

先ほど変更した "name": "Yamada", "age": 2 レコードを削除します。

from google.cloud import bigquery
from google.oauth2 import service_account

def delete_data(client: bigquery.Client):
    dataset_id = "{}.sample_dataset".format(client.project)
    dataset = client.get_dataset(dataset_id)
    table_id = "{}.{}.sample_table_name".format(client.project, dataset.dataset_id)

    query = f"""
    DELETE
    FROM
        `{table_id}`
    WHERE
        name = "Yamada";
    """
    client.query(query).result()

if __name__ == "__main__":
    delete_data(client)

上記のコードを実行して、Cloud コンソールから sample_table_name テーブルの "name": "Yamada", "age": 2 が削除されていれば OK です。

最後に

いかがでしたか？ 本記事では、 Python と BigQuery を使用した開発の始め方と基本的なテーブル操作を記述しました。本記事が Python と BigQuery を使用した開発を始める方の一助となれば幸いです。