BigQuery で Apache Iceberg 用の BigLake テーブルを作成する

概要

タイトルの通り、BigQuery で Apache Iceberg 用の BigLake テーブルを作成する方法を書きます。Google Cloud のドキュメントではいくつかの方法が紹介されていますが、この記事では BigLake Metastore を使用して BigLake テーブルを作成します。

想定読者

Google Cloud のコンソール画面を触ったことがあり、BigQuery 周りの操作についてある程度知識があることを想定しています。データセットの作成や権限付与などの詳細手順が分からない場合は Google Cloud が公開しているドキュメントなどをご参照ください。

作成手順

以下のような手順でテーブルを作成します。それぞれの手順について後述します。

1. BigQuery データセットを作成
2. 外部接続を作成
3. Cloud Storage バケットを用意
4. 外部接続を作成した時に発行されたサービスアカウントに権限を付与
5. Spark ストアドプロシージャを作成
6. ストアドプロシージャを実行し、BigLake Metastore と BigLake テーブルを作成

1. BigQuery データセットを作成

まずデータセットを作成します。BigQuery コンソール画面から作成します。
詳細な手順は以下のドキュメントをご参照ください。
https://cloud.google.com/bigquery/docs/datasets?hl=ja

2. 外部接続を作成

BigQuery から Cloud Storage へのアクセスを承認するための外部接続を作成します。外部接続の詳細については以下のドキュメントをご参照ください。
https://cloud.google.com/bigquery/docs/connections-api-intro?hl=ja
外部接続は BigQuery のコンソール画面から作成することができます。以下の赤枠の作成ボタンをクリックします。

クリックすると以下のようにいくつか選択肢が出てくるので、「外部データソースへの接続」をクリックして外部接続を作ります。
今回作成する外部接続は2つです。接続タイプが「Apache Spark」と「Vertex AI リモートモデル、リモート関数、BigLake（Cloud リソース）」の接続を作成します。作成後設定の詳細画面を開くとサービスアカウントが作成されていることがわかります。以降の手順でこのサービスアカウントに権限を付与していきます。便宜的に接続タイプが「Apache Spark」の外部接続を「spark-connection」、「Vertex AI リモートモデル、リモート関数、BigLake（Cloud リソース）」の外部接続を「biglake-connection」としておきます。

3. Cloud Storage バケットを用意

Iceberg テーブルを構成するためのバケットを2つ用意します。カタログ用のバケット(...-example-catalog)とソースデータを配置するためのバケット(...-example-source-data)です。

...-example-source-data には データが格納された Parquet ファイルを配置しておきます。

4. 接続を作成した時に発行されたサービスアカウントに権限を付与

外部接続を作成した時に発行されたサービスアカウントに権限を付与します。以下のようになります。

• biglake-connection
  • BigLake 管理者
  • ...-example-catalog のストレージオブジェクト管理者
  • 作成した BigQuery データセットのデータオーナー
• spark-connection
  • ...-example-catalog のストレージオブジェクト管理者
  • ...-example-source-data のオブジェクト閲覧者
  • BigLake 管理者
  • 作成した BigQuery データセットのデータオーナー
  • BigQuery ユーザー

5. Spark ストアドプロシージャを作成

Spark を使用して BigLake Metastore と BigLake テーブルの作成をします。
以下のドキュメントにサンプルコードが記載されていますが、データを追加するためコードを少し追加しています。
https://cloud.google.com/bigquery/docs/iceberg-external-tables?hl=ja#create-using-biglake-metastore
コードは以下のようになります。

CREATE OR REPLACE PROCEDURE
    `{project_id}.iceberg_example_dataset.iceberg_setup_3_3`()
WITH CONNECTION `{project_id}.asia-northeast1.spark-connection` OPTIONS (engine='SPARK',
        runtime_version='2.2',
        properties=[("spark.sql.catalog.iceberg_catalog.warehouse",
            "gs://{catalog_bucket}/iceberg-warehouse"),
        ("spark.sql.catalog.iceberg_catalog.blms_catalog",
            "iceberg_catalog"),
        ("spark.sql.catalog.iceberg_catalog.gcp_location",
            "asia-northeast1"),
        ("spark.sql.catalog.iceberg_catalog.gcp_project",
            "{project_id}"),
        ("spark.jars.packages",
            "org.apache.iceberg:iceberg-spark-runtime-3.3_2.12:1.2.0"),
        ("spark.sql.catalog.iceberg_catalog",
            "org.apache.iceberg.spark.SparkCatalog"),
        ("spark.sql.catalog.iceberg_catalog.catalog-impl",
            "org.apache.iceberg.gcp.biglake.BigLakeCatalog")],
        jar_uris=["gs://spark-lib/biglake/biglake-catalog-iceberg1.2.0-0.1.0-with-dependencies.jar"])
    LANGUAGE python AS R"""
from pyspark.sql import SparkSession

spark = SparkSession \
.builder \
.appName("BigLake Iceberg Example") \
.enableHiveSupport() \
.getOrCreate()

spark.sql("CREATE NAMESPACE IF NOT EXISTS iceberg_catalog;")
spark.sql("CREATE DATABASE IF NOT EXISTS iceberg_catalog.sample_db;")
spark.sql("DROP TABLE IF EXISTS iceberg_catalog.sample_db.sample_table;")

# Create a BigLake Metastore table and a BigQuery Iceberg table.
spark.sql("CREATE TABLE IF NOT EXISTS iceberg_catalog.sample_db.sample_table (id bigint, demo_name string) USING iceberg TBLPROPERTIES(bq_table='{dataset}.iceberg_table', bq_connection='{project_id}.asia-northeast1.biglake-connection');")


source_data = "gs://{source_data_bucket}/*.parquet"
df = spark.read.parquet("{}".format(source_data))
df.createOrReplaceTempView("temp_view")

spark.sql("INSERT INTO iceberg_catalog.sample_db.sample_table (id, demo_name) SELECT id, demo_name FROM temp_view;")
""";

また、以下の項目は環境に合わせて埋めてください。

• {project_id}：Google Cloud のプロジェクトID
• {catalog_bucket}：カタログ用のバケット
• {source_data_bucket}：データが格納されているバケット
• {dataset}：BigQuery のデータセット

6. ストアドプロシージャを実行し、BigLake Metastore と BigLake テーブルを作成

保存したプロシージャを実行して、Iceberg テーブルを作成します。

CALL `{project_id}.{dataset}.iceberg_setup_3_3`();

実行すると、テーブルを作成しデータのINSERTをします。実行後SELECT文でデータの中身を確認すると以下のようになりテーブルの作成に成功したことがわかります。

SELECT * FROM `mp-sandbox-230708.iceberg_example_dataset.iceberg_table`

---------------------------
id	demo_name
1	aaa
2	bbb
3	ccc
---------------------------

まとめ

本記事では、BigQuery で Apache Iceberg 用の BigLake テーブルを作成する方法を紹介しました。具体的に、Apache Spark のプロシージャを実行することによって BigLake Metastore とBigLake テーブルを作成しました。この方法によってGCSに構築されたIceberg形式のデータにBigQueryからクエリすることが可能になりました。