Lakehouse ランタイム カタログ、Spark、BigQuery を使用して Iceberg テーブルにクエリを実行する

IAM ロールを付与する

Managed Service for Apache Spark PySpark ジョブと Lakehouse ランタイム カタログが Cloud Storage と BigQuery を操作できるようにするには、対応するプリンシパルに必要なロールを付与します。

1. Google Cloud コンソールで、[Cloud Shell をアクティブにする] をクリックします。

   Cloud Shell をアクティブにする

2. [承認] をクリックします。

3. Dataproc ワーカー ロールをプロジェクトの Compute Engine のデフォルトのサービス アカウントに付与します。Managed Service for Apache Spark は、デフォルトでこのアカウントを使用します。詳細については、Managed Service for Apache Spark サービス アカウントをご覧ください。

   gcloud projects add-iam-policy-binding PROJECT_ID \
       --member="serviceAccount:$(gcloud projects describe PROJECT_ID --format='value(projectNumber)')-compute@developer.gserviceaccount.com" \
       --role="roles/dataproc.worker"

4. プロジェクトの Compute Engine デフォルト サービス アカウントに Service Usage コンシューマー ロールを付与します。

   gcloud projects add-iam-policy-binding PROJECT_ID \
       --member="serviceAccount:$(gcloud projects describe PROJECT_ID --format='value(projectNumber)')-compute@developer.gserviceaccount.com" \
       --role="roles/serviceusage.serviceUsageConsumer"

5. プロジェクトの Compute Engine のデフォルトのサービス アカウントに BigLake 編集者ロールを付与します。

   gcloud projects add-iam-policy-binding PROJECT_ID \
       --member="serviceAccount:$(gcloud projects describe PROJECT_ID --format='value(projectNumber)')-compute@developer.gserviceaccount.com" \
       --role="roles/biglake.editor"

6. プロジェクトの Compute Engine のデフォルトのサービス アカウントに BigQuery データ編集者ロールを付与します。

   gcloud projects add-iam-policy-binding PROJECT_ID \
       --member="serviceAccount:$(gcloud projects describe PROJECT_ID --format='value(projectNumber)')-compute@developer.gserviceaccount.com" \
       --role="roles/bigquery.dataEditor"

   次のように置き換えます。

   • PROJECT_ID: 実際の Google Cloud プロジェクト ID

Lakehouse ランタイム カタログを作成する

Iceberg テーブルのメタデータを管理する Lakehouse ランタイム カタログを作成します。

1. Cloud Shell で次のコマンドを実行して、認証情報ベンダーを使用してマルチバケット（bl://）を作成します。

   gcloud biglake iceberg catalogs create LAKEHOUSE_CATALOG_ID \
       --project=PROJECT_ID \
       --catalog-type=biglake \
       --default-location=gs://BUCKET_NAME \
       --credential-mode=vended-credentials

   次のように置き換えます。

   • LAKEHOUSE_CATALOG_ID: カタログの一意の名前。
   • PROJECT_ID: 実際の Google Cloud プロジェクト ID。
   • BUCKET_NAME: PySpark アプリケーション ファイルを含む Cloud Storage バケットの名前。

2. バケットに対する権限をカタログのサービス アカウントに付与します。

   1. Google Cloud コンソールで、[Lakehouse] に移動します。

      [レイクハウス] に移動

   2. 作成したカタログの名前（LAKEHOUSE_CATALOG_ID）をクリックします。

   3. [認証方法] で、[バケットの権限を設定] をクリックします。

   4. ダイアログで [確認] をクリックします。これにより、カタログのサービス アカウントにバケットに対する Storage オブジェクト ユーザーのロールがあることが確認されます。

PySpark ジョブを作成して実行する

Iceberg テーブルを作成してクエリするには、まず必要な Spark SQL ステートメントを使用して PySpark ジョブを作成します。次に、Managed Service for Apache Spark でジョブを実行します。

Namespace とテーブルを含む PySpark スクリプトを作成する

テキスト エディタで、次の内容の quickstart.py というファイルを作成します。

この PySpark スクリプトは、Iceberg カタログでいくつかのオペレーションを実行するために Spark セッションを初期化します。スクリプトは、まず Namespace が存在しない場合は Namespace を作成します。次に、基本的なスキーマを使用して quickstart_table という名前の Iceberg テーブルを作成します。テーブルが作成されると、スクリプトは 3 行のデータを挿入します。最後に、テーブルに対してクエリを実行して、挿入されたすべてのレコードを取得します。

これらの値は、次のステップで gcloud dataproc batches submit pyspark ジョブを実行するときに使用されます。

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("quickstart").getOrCreate()

# Create a namespace (dataset) if it doesn't exist
spark.sql("CREATE NAMESPACE IF NOT EXISTS `quickstart_catalog`.quickstart_namespace")

# Create the table
spark.sql("""
    CREATE OR REPLACE TABLE `quickstart_catalog`.quickstart_namespace.quickstart_table (
        id INT,
        name STRING
    )
    USING iceberg
""")

# Insert data into the table
spark.sql("""
    INSERT INTO `quickstart_catalog`.quickstart_namespace.quickstart_table
    VALUES (1, 'one'), (2, 'two'), (3, 'three')
""")

スクリプトを Cloud Storage バケットにアップロードする

quickstart.py スクリプトを作成したら、Cloud Storage バケットにアップロードします。

1. Google Cloud コンソールで、Cloud Storage バケットに移動します。

   [バケット] に移動

2. バケットの名前をクリックします。

3. [オブジェクト] タブで、[アップロード] > [ファイルをアップロード] をクリックします。

4. ファイル ブラウザで quickstart.py ファイルを選択し、[開く] をクリックします。

PySpark ジョブを実行する

quickstart.py スクリプトをアップロードしたら、Managed Service for Apache Spark バッチジョブとして実行します。

1. Cloud Shell で、quickstart.py スクリプトを使用して次の Managed Service for Apache Spark バッチジョブを実行します。

   gcloud dataproc batches submit pyspark gs://BUCKET_NAME/quickstart.py \
       --project=PROJECT_ID \
       --region=REGION \
       --version=2.2 \
       --properties="\
   spark.sql.defaultCatalog=quickstart_catalog,\
   spark.sql.catalog.quickstart_catalog=org.apache.iceberg.spark.SparkCatalog,\
   spark.sql.catalog.quickstart_catalog.type=rest,\
   spark.sql.catalog.quickstart_catalog.uri=https://biglake.googleapis.com/iceberg/v1/restcatalog,\
   spark.sql.catalog.quickstart_catalog.warehouse=bl://projects/PROJECT_ID/catalogs/LAKEHOUSE_CATALOG_ID,\
   spark.sql.catalog.quickstart_catalog.io-impl=org.apache.iceberg.gcp.gcs.GCSFileIO,\
   spark.sql.catalog.quickstart_catalog.header.x-goog-user-project=PROJECT_ID,\
   spark.sql.catalog.quickstart_catalog.rest.auth.type=org.apache.iceberg.gcp.auth.GoogleAuthManager,\
   spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions,\
   spark.sql.catalog.quickstart_catalog.header.X-Iceberg-Access-Delegation=vended-credentials,\
   spark.sql.catalog.quickstart_catalog.gcs.oauth2.refresh-credentials-endpoint=https://oauth2.googleapis.com/token"

   次のように置き換えます。

   • BUCKET_NAME: PySpark アプリケーション ファイルを含む Cloud Storage バケットの名前。

   • LAKEHOUSE_CATALOG_ID: BigLake カタログの名前。この名前は、後で BigQuery で P.C.N.T 構文を使用してカタログをクエリするときに使用されます。例: my-project.biglake-catalog.quickstart_namespace.quickstart_table

   • PROJECT_ID: 実際の Google Cloud プロジェクト ID。

   • REGION: Managed Service for Apache Spark バッチ ワークロードを実行するリージョン。

   ジョブが完了すると、次のような出力が表示されます。

   Batch [cb9d84e9489d408baca4f9e7ab4c64ff] finished.
   metadata:
   '@type': type.googleapis.com/google.cloud.dataproc.v1.BatchOperationMetadata
   batch: projects/your-project/locations/us-central1/batches/cb9d84e9489d408baca4f9e7ab4c64ff
   batchUuid: 54b0b9d2-f0a1-4fdf-ae44-eead3f8e60e9
   createTime: '2026-01-24T00:10:50.224097Z'
   description: Batch
   labels:
       goog-dataproc-batch-id: cb9d84e9489d408baca4f9e7ab4c64ff
       goog-dataproc-batch-uuid: 54b0b9d2-f0a1-4fdf-ae44-eead3f8e60e9
       goog-dataproc-drz-resource-uuid: batch-54b0b9d2-f0a1-4fdf-ae44-eead3f8e60e9
       goog-dataproc-location: us-central1
   operationType: BATCH
   name: projects/your-project/regions/us-central1/operations/32287926-5f61-3572-b54a-fbad8940d6ef
   

BigQuery からテーブルに対してクエリを実行する

1. Google Cloud コンソールで、[BigQuery] に移動します。

   BigQuery に移動

2. クエリエディタで次のステートメントを入力します。このクエリでは project.catalog.namespace.table 構文を使用します。

   SELECT * FROM `PROJECT_ID.LAKEHOUSE_CATALOG_ID.quickstart_namespace.quickstart_table`;
   

   次のように置き換えます。

   • PROJECT_ID: 実際の Google Cloud プロジェクト ID。

   • LAKEHOUSE_CATALOG_ID: BigQuery クエリで使用するカタログ ID。

3. [実行] をクリックします。

   クエリ結果には、PySpark ジョブで挿入したデータが表示されます。