"Managed Service for Apache Spark" is the new name for the product formerly known as "Dataproc on Compute Engine" (cluster deployment) and "Google Cloud Serverless for Apache Spark" (serverless deployment).

Google は AI 技術を使用して、コンテンツをご希望の言語に翻訳しています。AI 翻訳には誤りが含まれる場合があります。

Spark と Lakehouse ランタイムカタログを使用して Lakehouse を作成する

レイクハウスアーキテクチャは、データレイクの柔軟性とデータウェアハウスのデータマネジメント機能を組み合わせたものです。このドキュメントでは、 Google Cloudにレイクハウスを設定する方法について説明します。テーブル形式として Apache Iceberg、処理に Managed Service for Apache Spark、統合メタデータ管理に Lakehouse ランタイムカタログの Iceberg REST カタログを使用します。

このアーキテクチャでは、Iceberg などのオープンテーブル形式を使用して、トランザクションやスキーマの進化などのデータウェアハウス機能を Cloud Storage のデータに追加します。このアプローチでは、さまざまなエンジンがアクセスできるデータの信頼できる唯一の情報源が作成されます。

このドキュメントでは、Lakehouse ランタイムカタログで作成するカタログについて説明します。Lakehouse ランタイムカタログの連携カタログを介して、AWS Glue Data Catalog、Databricks Unity Catalog、Snowflake Horizon Catalog で管理されている Iceberg テーブルをクエリするには、Spark で連携カタログをクエリするをご覧ください。

Managed Service for Apache Spark、Cloud Storage、Lakehouse REST カタログなど、レイクハウスアーキテクチャのコンポーネントを示す図。 — レイクハウスアーキテクチャ図。

始める前に

Google Cloud アカウントにログインします。 Google Cloudを初めて使用する場合は、アカウントを作成して、実際のシナリオで Google プロダクトのパフォーマンスを評価してください。新規のお客様には、ワークロードの実行、テスト、デプロイができる無料クレジット $300 分も差し上げます。

In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

Roles required to select or create a project

Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

Go to project selector

Verify that you have the permissions required to complete this guide.

Verify that billing is enabled for your Google Cloud project.

Enable the Dataproc, BigLake, BigQuery, and Cloud Storage APIs, if any are not already enabled.

Roles required to enable APIs

To enable APIs, you need the serviceusage.services.enable permission. If you created the project, then you likely already have this permission through the Owner role (roles/owner). Otherwise, you can get this permission through the Service Usage Admin role (roles/serviceusage.serviceUsageAdmin). Learn how to grant roles.

Enable the APIs

In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

Roles required to select or create a project

Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

Go to project selector

Verify that you have the permissions required to complete this guide.

Verify that billing is enabled for your Google Cloud project.

Enable the Dataproc, BigLake, BigQuery, and Cloud Storage APIs, if any are not already enabled.

Roles required to enable APIs

Enable the APIs

Iceberg データを保存する Cloud Storage バケットを作成します。
バケットを参照する Lakehouse ランタイムカタログにカタログを作成します。マルチバケットカタログをおすすめします。

必要なロール

このページの例を実行するには、特定の Identity and Access Management（IAM）ロールが必要です。組織のポリシーによっては、これらのロールがすでに付与されている場合があります。ロールの付与を確認するには、ロールを付与する必要があるかどうかをご覧ください。

ロールの付与については、プロジェクト、フォルダ、組織に対するアクセス権の管理をご覧ください。

必要なロールは、ワークロードの実行方法によって異なります。Managed Service for Apache Spark クラスタと Managed Service for Apache Spark サーバーレスランタイム 2.3 以前では、ワークロードがサービスアカウントとして実行されます。Managed Service for Apache Spark サーバーレスランタイム 3.0 以降では、デフォルトでエンドユーザー認証情報（EUC）を使用してワークロードが実行されるため、サービスアカウントではなくユーザーアカウントにデータアクセスロールを付与します。3.0 以降のランタイムにカスタムサービスアカウントを指定すると、ワークロードはサービスアカウントの認証情報で実行されます。詳細については、サーバーレスデプロイのロールをご覧ください。

ユーザーロール

Managed Service for Apache Spark クラスタの作成、バッチワークロードの送信、インタラクティブセッションの作成に必要な権限を取得するには、次の IAM ロールを付与するよう管理者に依頼してください。

クラスタまたはサーバーレスランタイム 2.3 以前を使用します。
- プロジェクトに対する Dataproc 編集者（roles/dataproc.editor）
- Compute Engine のデフォルトのサービスアカウントまたはカスタムサービスアカウントに対するサービスアカウントユーザー（roles/iam.serviceAccountUser）
サーバーレスランタイム 3.0 以降を使用する: プロジェクトに対する Dataproc Serverless 編集者（roles/dataproc.serverlessEditor）
サーバーレスランタイム 3.0 以降でサービスアカウント認証情報を使用する: カスタムサービスアカウントに対するサービスアカウントユーザー（roles/iam.serviceAccountUser）

エンドユーザー認証情報（デフォルト）で Managed Service for Apache Spark サーバーレスランタイム 3.0 以降を使用する場合は、次のデータアクセスロールも必要です。

エンドユーザーの認証情報を使用してカタログ内のテーブルの読み取りと書き込みに必要な権限を取得するには、次の IAM ロールを付与するよう管理者に依頼してください。

認証情報提供モードでテーブルデータの読み取りと書き込みを行う: カタログを含むプロジェクトに対する BigLake 編集者（roles/biglake.editor）
カタログリソースを管理し、認証情報ベンディング以外のモードでテーブルデータを書き込む:
- カタログを含むプロジェクトに対する BigLake 編集者（roles/biglake.editor）
- カタログに関連付けられているすべての Cloud Storage バケットに対する Storage オブジェクトユーザー（roles/storage.objectUser）

サービスアカウントのロール

ワークロードを実行するサービスアカウントに次のロールを付与します。Managed Service for Apache Spark サーバーレスランタイム 3.0 以降では、ワークロードがサービスアカウント認証情報で実行される場合にのみ、これらのロールが必要です。

Compute Engine のデフォルトサービスアカウントまたはカスタムサービスアカウントに、Managed Service for Apache Spark ジョブを実行し、カタログ内のテーブルを読み書きするために必要な権限が付与されるようにするには、Compute Engine のデフォルトサービスアカウントまたはカスタムサービスアカウントに次の IAM ロールを付与するよう管理者に依頼してください。

クラスタまたはサーバーレスランタイム 2.3 以前で Managed Service for Apache Spark ジョブを実行する: プロジェクトに対する Dataproc ワーカー（roles/dataproc.worker）
サーバーレスランタイム 3.0 以降で Managed Service for Apache Spark ジョブを実行します。
- プロジェクトに対する Dataproc Serverless ノード（roles/dataproc.serverlessNode）
- プロジェクトに対する Dataproc Resource Manager ノードサービスエージェント（roles/dataprocrm.nodeServiceAgent）
認証情報ベンディングモードでテーブルデータを読み取り、書き込む: カタログを含むプロジェクトに対する BigLake 編集者（roles/biglake.editor）
カタログリソースを管理し、非認証情報提供モードでテーブルデータを書き込む:
- カタログを含むプロジェクトに対する BigLake 編集者（roles/biglake.editor）
- カタログに関連付けられているすべての Cloud Storage バケットに対するStorage オブジェクトユーザー（roles/storage.objectUser）

Managed Service for Apache Spark を構成して Spark セッションを作成する

Managed Service for Apache Spark クラスタ、Managed Service for Apache Spark サーバーレスバッチワークロード、または Managed Service for Apache Spark サーバーレスインタラクティブセッションを構成して、Lakehouse ランタイムカタログ Iceberg REST カタログを使用します。3 つすべてが dataproc.lakehouse.catalog.CATALOG_NAME 簡素化された構成プロパティ（推奨）をサポートしています。これにより、Managed Service for Apache Spark はカタログ定義を読み取り、Iceberg REST カタログの Spark プロパティを生成できます。クラスタ形式のみが dataproc: 接頭辞を取ります。

サポートされているイメージとランタイムのバージョン:

簡素化された構成と手動構成の両方で、Iceberg の Google 認証（GoogleAuthManager）が使用されます。これには、次のいずれかのバージョンが必要です。

Managed Service for Apache Spark クラスタ: 2.2 イメージバージョン 2.2.65 以降、または 2.3 イメージバージョン 2.3.11 以降。
Managed Service for Apache Spark のサーバーレスバッチワークロードとインタラクティブセッション: 2.2 ランタイム 2.2.60 以降、2.3 ランタイム 2.3.10 以降、または 3.0 ランタイム。

サーバーレスバッチワークロードとインタラクティブセッションの場合、必要なロールはランタイムバージョンによって異なります。ランタイム 3.0 以降では、デフォルトでエンドユーザー認証情報が使用されます。詳細については、必要なロールをご覧ください。

クラスタ

簡略化された構成プロパティ（推奨）を使用するか、ノートブックでカタログプロパティを手動で指定して、Iceberg と Jupyter のオプションコンポーネントを含む Managed Service for Apache Spark クラスタを作成します。

プロパティを使用した簡略化された構成（推奨）

カタログプロパティを使用してクラスタを作成します。

gcloud dataproc clusters create CLUSTER_NAME \
  --project=PROJECT_ID \
  --region=REGION \
  --image-version=IMAGE_VERSION \
  --optional-components=ICEBERG,JUPYTER \
  --enable-component-gateway \
  --properties='dataproc:dataproc.lineage.enabled=true,dataproc:dataproc.lakehouse.catalog.CATALOG_NAME=projects/PROJECT_ID/catalogs/CATALOG_ID,spark:spark.sql.defaultCatalog=CATALOG_NAME'

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

CLUSTER_NAME: クラスタの名前。
PROJECT_ID: 実際のGoogle Cloud プロジェクト ID。
REGION: クラスタの Google Cloudリージョン（us-central1 など）。
IMAGE_VERSION: Managed Service for Apache Spark のイメージバージョン（例: 2.3-debian12）。サポートされているイメージとランタイムのバージョンをご覧ください。
CATALOG_NAME: ローカル Spark カタログの名前（例: iceberg_catalog）。CATALOG_ID と同じ値を指定できます。
CATALOG_ID: 始める前にで作成したカタログの ID。

Lakehouse ランタイムカタログの Iceberg REST カタログが正しく機能するために、dataproc:dataproc.lineage.enabled=true プロパティは必要ありません。これは、データリネージの例でリネージトラッキング用に追加されています。

Jupyter ノートブック（Vertex AI Workbench ノートブックまたはクラスタで起動されたノートブック）を使用してクラスタに接続し、カタログ構成を指定せずに SparkSession を作成します。

from pyspark.sql import SparkSession

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

APP_NAME は、Spark アプリケーションの名前に置き換えます。

手動設定

簡略化された構成プロパティを使用しない場合は、前述のように --properties フラグに dataproc:dataproc.lakehouse.catalog. を含めずにクラスタを作成し、Jupyter Notebook に接続して、次のように SparkSession を手動で構成します。

import pyspark
from pyspark.context import SparkContext
from pyspark.sql import SparkSession

catalog_name = "CATALOG_NAME"

spark = SparkSession.builder.appName("APP_NAME") \
.config('spark.sql.defaultCatalog', catalog_name) \
.config(f'spark.sql.catalog.{catalog_name}', 'org.apache.iceberg.spark.SparkCatalog') \
.config(f'spark.sql.catalog.{catalog_name}.type', 'rest') \
.config(f'spark.sql.catalog.{catalog_name}.uri', 'https://biglake.googleapis.com/iceberg/v1/restcatalog') \
.config(f'spark.sql.catalog.{catalog_name}.warehouse', 'WAREHOUSE_PATH') \
.config(f'spark.sql.catalog.{catalog_name}.header.x-goog-user-project', 'PROJECT_ID') \
.config(f'spark.sql.catalog.{catalog_name}.rest.auth.type', 'org.apache.iceberg.gcp.auth.GoogleAuthManager') \
.config(f'spark.sql.catalog.{catalog_name}.io-impl', 'org.apache.iceberg.gcp.gcs.GCSFileIO') \
.config('spark.sql.extensions', 'org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions') \
.getOrCreate()

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

CATALOG_NAME: ローカル Spark カタログの名前（例: iceberg_catalog）。
APP_NAME: Spark アプリケーションの名前。
WAREHOUSE_PATH: ウェアハウスのパス。複数のバケットカタログ（推奨）の場合は、bl://projects/PROJECT_ID/catalogs/CATALOG_ID を使用します。単一バケットカタログの場合は、gs://CLOUD_STORAGE_BUCKET_NAME を使用します。
PROJECT_ID: 実際の Google Cloud プロジェクト ID。

認証情報ベンディングを使用して構成する

認証情報の自動販売を使用するには、認証情報の自動販売モードのカタログを使用し、次の行を SparkSession ビルダーに追加して、vended-credentials の値を持つ Iceberg REST カタログリクエストに X-Iceberg-Access-Delegation ヘッダーを追加する必要があります。

.config(f'spark.sql.catalog.{catalog_name}.header.X-Iceberg-Access-Delegation', 'vended-credentials')

次の例では、認証情報のベンディングを使用して SparkSession を構成します。

from pyspark.sql import SparkSession

catalog_name = "CATALOG_NAME"

spark = SparkSession.builder.appName("APP_NAME") \
.config('spark.sql.defaultCatalog', catalog_name) \
.config(f'spark.sql.catalog.{catalog_name}', 'org.apache.iceberg.spark.SparkCatalog') \
.config(f'spark.sql.catalog.{catalog_name}.type', 'rest') \
.config(f'spark.sql.catalog.{catalog_name}.uri', 'https://biglake.googleapis.com/iceberg/v1/restcatalog') \
.config(f'spark.sql.catalog.{catalog_name}.warehouse', 'WAREHOUSE_PATH') \
.config(f'spark.sql.catalog.{catalog_name}.header.x-goog-user-project', 'PROJECT_ID') \
.config(f'spark.sql.catalog.{catalog_name}.rest.auth.type', 'org.apache.iceberg.gcp.auth.GoogleAuthManager') \
.config(f'spark.sql.catalog.{catalog_name}.io-impl', 'org.apache.iceberg.gcp.gcs.GCSFileIO') \
.config(f'spark.sql.catalog.{catalog_name}.header.X-Iceberg-Access-Delegation', 'vended-credentials') \
.config('spark.sql.extensions', 'org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions') \
.getOrCreate()

カタログが認証情報自動販売モードを使用している場合、簡素化された構成でこのヘッダーが追加されます。詳細については、Iceberg ドキュメントの RESTCatalog のヘッダーセクションをご覧ください。

サーバーレス

簡素化された構成プロパティ（推奨）を使用するか、カタログプロパティを手動で指定して、PySpark バッチワークロードを Managed Service for Apache Spark サーバーレスに送信します。

プロパティを使用した簡略化された構成（推奨）

バッチワークロードを送信するときに、dataproc.lakehouse.catalog.CATALOG_NAME プロパティ（dataproc: 接頭辞なし）を渡します。

gcloud dataproc batches submit pyspark PYSPARK_FILE \
  --project=PROJECT_ID \
  --region=REGION \
  --version=RUNTIME_VERSION \
  --properties="dataproc.lakehouse.catalog.CATALOG_NAME=projects/PROJECT_ID/catalogs/CATALOG_ID,spark.sql.defaultCatalog=CATALOG_NAME"

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

PYSPARK_FILE: PySpark アプリケーションファイルへの gs:// Cloud Storage パス。
PROJECT_ID: 実際のGoogle Cloud プロジェクト ID。
REGION: バッチワークロードのリージョン。
RUNTIME_VERSION: Managed Service for Apache Spark サーバーレスランタイムバージョン（例: 3.0）。必要なロールは、ランタイムバージョンによって異なります（サポートされているイメージとランタイムのバージョンをご覧ください）。
CATALOG_NAME: ローカル Spark カタログの名前（例: iceberg_catalog）。CATALOG_ID と同じ値を指定できます。
CATALOG_ID: 始める前にで作成したカタログの ID。

PySpark アプリケーションファイルで、カタログ構成を指定せずに SparkSession を作成します。

from pyspark.sql import SparkSession

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

手動設定

簡素化された構成プロパティを使用しない場合は、バッチワークロードを送信するときに --properties フラグでカタログプロパティを指定します。

gcloud dataproc batches submit pyspark PYSPARK_FILE \
  --project=PROJECT_ID \
  --region=REGION \
  --version=RUNTIME_VERSION \
  --properties="\
spark.sql.defaultCatalog=CATALOG_NAME,\
spark.sql.catalog.CATALOG_NAME=org.apache.iceberg.spark.SparkCatalog,\
spark.sql.catalog.CATALOG_NAME.type=rest,\
spark.sql.catalog.CATALOG_NAME.uri=https://biglake.googleapis.com/iceberg/v1/restcatalog,\
spark.sql.catalog.CATALOG_NAME.warehouse=WAREHOUSE_PATH,\
spark.sql.catalog.CATALOG_NAME.header.x-goog-user-project=PROJECT_ID,\
spark.sql.catalog.CATALOG_NAME.rest.auth.type=org.apache.iceberg.gcp.auth.GoogleAuthManager,\
spark.sql.catalog.CATALOG_NAME.io-impl=org.apache.iceberg.gcp.gcs.GCSFileIO,\
spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions"

WAREHOUSE_PATH は、ウェアハウスのパスに置き換えます。複数バケットカタログ（推奨）の場合は、bl://projects/PROJECT_ID/catalogs/CATALOG_ID を使用します。単一バケットカタログの場合は、gs://CLOUD_STORAGE_BUCKET_NAME を使用します。

認証情報ベンディングを使用して構成する

認証情報の自動販売を使用するには、認証情報の自動販売モードのカタログを使用し、--properties フラグに次のプロパティを追加して、vended-credentials の値を持つ X-Iceberg-Access-Delegation ヘッダーを Iceberg REST カタログリクエストに追加する必要があります。

spark.sql.catalog.CATALOG_NAME.header.X-Iceberg-Access-Delegation=vended-credentials

次の例では、認証情報の提供を使用してバッチワークロードを送信します。

gcloud dataproc batches submit pyspark PYSPARK_FILE \
  --project=PROJECT_ID \
  --region=REGION \
  --version=RUNTIME_VERSION \
  --properties="\
spark.sql.defaultCatalog=CATALOG_NAME,\
spark.sql.catalog.CATALOG_NAME=org.apache.iceberg.spark.SparkCatalog,\
spark.sql.catalog.CATALOG_NAME.type=rest,\
spark.sql.catalog.CATALOG_NAME.uri=https://biglake.googleapis.com/iceberg/v1/restcatalog,\
spark.sql.catalog.CATALOG_NAME.warehouse=WAREHOUSE_PATH,\
spark.sql.catalog.CATALOG_NAME.header.x-goog-user-project=PROJECT_ID,\
spark.sql.catalog.CATALOG_NAME.rest.auth.type=org.apache.iceberg.gcp.auth.GoogleAuthManager,\
spark.sql.catalog.CATALOG_NAME.io-impl=org.apache.iceberg.gcp.gcs.GCSFileIO,\
spark.sql.catalog.CATALOG_NAME.header.X-Iceberg-Access-Delegation=vended-credentials,\
spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions"

サーバーレス

簡略化された構成プロパティ（推奨）を使用するか、カタログプロパティを手動で指定して、DataprocSparkSession API を使用してノートブックから Managed Service for Apache Spark サーバーレスインタラクティブセッションを作成します。

プロパティを使用した簡素化された構成（推奨）

dataproc-spark-connect パッケージがインストールされているノートブック（BigQuery Studio ノートブックや Vertex AI Workbench ノートブックなど）で、セッションの作成時に dataproc.lakehouse.catalog.CATALOG_NAME プロパティ（dataproc: 接頭辞なし）を渡します。

from google.cloud.dataproc_spark_connect import DataprocSparkSession
from google.cloud.dataproc_v1 import Session

catalog_name = "CATALOG_NAME"

session_config = Session()
session_config.runtime_config.version = 'RUNTIME_VERSION'

spark = DataprocSparkSession.builder.appName("APP_NAME") \
.dataprocSessionConfig(session_config) \
.config(f'dataproc.lakehouse.catalog.{catalog_name}', 'projects/PROJECT_ID/catalogs/CATALOG_ID') \
.config('spark.sql.defaultCatalog', catalog_name) \
.getOrCreate()

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

CATALOG_NAME: ローカル Spark カタログの名前（例: iceberg_catalog）。CATALOG_ID と同じ値を指定できます。
RUNTIME_VERSION: Managed Service for Apache Spark サーバーレスランタイムバージョン（例: 3.0）。必要なロールは、ランタイムバージョンによって異なります（サポートされているイメージとランタイムのバージョンをご覧ください）。
APP_NAME: Spark アプリケーションの名前。
PROJECT_ID: 実際のGoogle Cloud プロジェクト ID。
CATALOG_ID: 始める前にで作成したカタログの ID。

これらのプロパティは、セッションテンプレートで設定することもできます。

手動設定

簡略化された構成プロパティを使用しない場合は、セッションの作成時にカタログプロパティを指定します。

from google.cloud.dataproc_spark_connect import DataprocSparkSession
from google.cloud.dataproc_v1 import Session

catalog_name = "CATALOG_NAME"

session_config = Session()
session_config.runtime_config.version = 'RUNTIME_VERSION'

spark = DataprocSparkSession.builder.appName("APP_NAME") \
.dataprocSessionConfig(session_config) \
.config('spark.sql.defaultCatalog', catalog_name) \
.config(f'spark.sql.catalog.{catalog_name}', 'org.apache.iceberg.spark.SparkCatalog') \
.config(f'spark.sql.catalog.{catalog_name}.type', 'rest') \
.config(f'spark.sql.catalog.{catalog_name}.uri', 'https://biglake.googleapis.com/iceberg/v1/restcatalog') \
.config(f'spark.sql.catalog.{catalog_name}.warehouse', 'WAREHOUSE_PATH') \
.config(f'spark.sql.catalog.{catalog_name}.header.x-goog-user-project', 'PROJECT_ID') \
.config(f'spark.sql.catalog.{catalog_name}.rest.auth.type', 'org.apache.iceberg.gcp.auth.GoogleAuthManager') \
.config(f'spark.sql.catalog.{catalog_name}.io-impl', 'org.apache.iceberg.gcp.gcs.GCSFileIO') \
.config('spark.sql.extensions', 'org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions') \
.getOrCreate()

このタブの前の簡略化された構成例で説明したように、RUNTIME_VERSION を置き換えます。

WAREHOUSE_PATH は、ウェアハウスのパスに置き換えます。マルチバケットカタログ（推奨）の場合は、bl://projects/PROJECT_ID/catalogs/CATALOG_ID を使用します。単一バケットカタログの場合は、gs://CLOUD_STORAGE_BUCKET_NAME を使用します。

認証情報ベンディングで構成する

認証情報の提供を使用するには、認証情報の提供モードのカタログを使用し、セッションビルダーに次の行を追加して、vended-credentials の値を持つ X-Iceberg-Access-Delegation ヘッダーを Iceberg REST カタログリクエストに追加する必要があります。

.config(f'spark.sql.catalog.{catalog_name}.header.X-Iceberg-Access-Delegation', 'vended-credentials')

次の例では、認証情報の自動販売を使用してセッションを作成します。

from google.cloud.dataproc_spark_connect import DataprocSparkSession
from google.cloud.dataproc_v1 import Session

catalog_name = "CATALOG_NAME"

session_config = Session()
session_config.runtime_config.version = 'RUNTIME_VERSION'

spark = DataprocSparkSession.builder.appName("APP_NAME") \
.dataprocSessionConfig(session_config) \
.config('spark.sql.defaultCatalog', catalog_name) \
.config(f'spark.sql.catalog.{catalog_name}', 'org.apache.iceberg.spark.SparkCatalog') \
.config(f'spark.sql.catalog.{catalog_name}.type', 'rest') \
.config(f'spark.sql.catalog.{catalog_name}.uri', 'https://biglake.googleapis.com/iceberg/v1/restcatalog') \
.config(f'spark.sql.catalog.{catalog_name}.warehouse', 'WAREHOUSE_PATH') \
.config(f'spark.sql.catalog.{catalog_name}.header.x-goog-user-project', 'PROJECT_ID') \
.config(f'spark.sql.catalog.{catalog_name}.rest.auth.type', 'org.apache.iceberg.gcp.auth.GoogleAuthManager') \
.config(f'spark.sql.catalog.{catalog_name}.io-impl', 'org.apache.iceberg.gcp.gcs.GCSFileIO') \
.config(f'spark.sql.catalog.{catalog_name}.header.X-Iceberg-Access-Delegation', 'vended-credentials') \
.config('spark.sql.extensions', 'org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions') \
.getOrCreate()

簡素化された構成では、カタログが認証情報ベンディングモードを使用している場合にこのヘッダーが追加されます。詳細については、Iceberg ドキュメントの RESTCatalog のヘッダーセクションをご覧ください。

Spark SQL を使用してデータを管理する

Spark セッションを構成したら、Spark SQL を使用してデータマネジメントオペレーションを実行します。

カタログの設定: dataproc.lakehouse.catalog. プロパティはカタログを Spark に登録しますが、デフォルトのカタログにはなりません。前のセクションの例では、修飾なしのテーブル名を使用できるように spark.sql.defaultCatalog（クラスタでは spark:spark.sql.defaultCatalog）を設定しています。このプロパティを省略した場合は、テーブル名をカタログ名で修飾します（例: CATALOG_NAME.NAMESPACE_NAME.base_table）。

Namespace を作成します。Lakehouse ランタイムカタログの Iceberg REST カタログでは、Namespace は BigQuery データセットに対応します。
```
spark.sql("CREATE NAMESPACE IF NOT EXISTS NAMESPACE_NAME")
spark.sql("USE NAMESPACE_NAME")
```
NAMESPACE_NAME は、名前空間の名前に置き換えます（例: spark_lakehouse）。

Iceberg 形式でベーステーブルを作成し、データを挿入します。

spark.sql("DROP TABLE IF EXISTS base_table PURGE")
spark.sql("CREATE TABLE base_table (id LONG) USING iceberg")
spark.sql("INSERT INTO base_table VALUES 0, 1, 2, 3, 4")
spark.sql("SELECT * FROM base_table").show()

出力は次のようになります。

   +---+
   | id|
   +---+
   |  0|
   |  1|
   |  2|
   |  3|
   |  4|
   +---+

新しいデータ用の別のテーブルを作成します。

spark.sql("DROP TABLE IF EXISTS newdata PURGE")
spark.sql("CREATE TABLE newdata(id LONG) USING iceberg")
spark.sql("INSERT INTO newdata VALUES 3, 4, 5, 6")
spark.sql("SELECT * FROM newdata").show()

出力は次のようになります。

   +---+
   | id|
   +---+
   |  3|
   |  4|
   |  5|
   |  6|
   +---+

新しいデータをベーステーブルに統合します。

spark.sql("""MERGE INTO base_table USING newdata
           ON base_table.id = newdata.id
           WHEN MATCHED THEN
             UPDATE SET base_table.id = newdata.id
           WHEN NOT MATCHED THEN INSERT * """)
spark.sql("SELECT * FROM base_table").show()

出力は次のようになります。

   +---+
   | id|
   +---+
   |  0|
   |  1|
   |  2|
   |  3|
   |  4|
   |  5|
   |  6|
   +---+

ベーステーブルのレコードを更新します。

spark.sql(
     "UPDATE base_table SET id = (id + 100) WHERE (id % 2 == 0)"
 )
spark.sql("SELECT * FROM base_table").show()

出力は次のようになります。

   +---+
   | id|
   +---+
   |  3|
   |104|
   |  5|
   |106|
   |100|
   |102|
   |  1|
   +---+

ベーステーブルからレコードを削除します。

spark.sql("DELETE FROM base_table WHERE (id % 2 == 0)")
spark.sql("SELECT * FROM base_table").show()

出力は次のようになります。

   +---+
   | id|
   +---+
   |  3|
   |  5|
   |  1|
   +---+

過去のスナップショットをクエリする

スナップショット ID をクエリして、テーブルの以前のバージョンを取得します。このオペレーションは、タイムトラベルとも呼ばれます。

MERGE、UPDATE、DELETE オペレーションの前に、テーブルバージョンのスナップショット ID を取得します。
```
snapshot_ids = spark.sql(
     "SELECT snapshot_id FROM `NAMESPACE_NAME`.`base_table`.snapshots"
).collect()
oldest_snapshot_id = snapshot_ids[1]["snapshot_id"]
```
NAMESPACE_NAME は、作成した Namespace に置き換えます。
取得したスナップショット ID を使用してテーブルをクエリします。
```
df = (
     spark.read.format("iceberg")
     .option("versionAsOf", oldest_snapshot_id)
     .load("base_table")
 )
df.show()
```
出力には、MERGE オペレーションの後、UPDATE オペレーションまたは DELETE オペレーションの前のテーブルの状態が表示されます。
```
   +---+
   | id|
   +---+
   |  0|
   |  1|
   |  2|
   |  3|
   |  4|
   |  5|
   |  6|
   +---+
```

データリネージを検出する

Managed Service for Apache Spark 2.2 以降のイメージバージョンを使用すると、データリネージを使用して、Lakehouse ランタイムカタログの Iceberg REST カタログテーブル間のデータ移動を追跡できます。

データリネージの例

ソース Iceberg テーブルとターゲット Iceberg テーブルを作成し、データをコピーします。

spark.sql("DROP TABLE IF EXISTS source_table PURGE")
spark.sql("DROP TABLE IF EXISTS target_table PURGE")
spark.sql("CREATE TABLE source_table (id LONG) USING iceberg")
spark.sql("""CREATE TABLE target_table
  USING ICEBERG
  AS SELECT max(id) as top_id FROM source_table
  """)

Google Cloud コンソールで、Knowledge Catalog の [検索] ページに移動します。

検索に移動
テーブルのいずれかを検索し、[Lineage] タブをクリックします。

Google Cloud コンソールの Knowledge Catalog ページのデータリネージグラフの例。

データリネージは、Lakehouse ランタイムカタログの Iceberg REST カタログテーブルの論理表現（Lakehouse ランタイムカタログテーブル）と物理表現（Cloud Storage）の両方を認識します。

データリネージに関する既知の問題

一部の Managed Service for Apache Spark クラスタでは、OpenLineage ライブラリの問題により、完全なデータリネージが生成されない場合があります。これは既知の制限事項です。ワークロードに影響する場合は、Google Cloud サポートにお問い合わせください。

次のステップ

Lakehouse ランタイムカタログの Iceberg REST カタログの詳細を確認する。
Spark で連携カタログをクエリする。
Apache Iceberg の機能について説明します。
Lakehouse ランタイムカタログから Iceberg データにクエリを実行する方法を確認する。
データリネージと Managed Service for Apache Spark の詳細を確認する。

Spark と Lakehouse ランタイム カタログを使用して Lakehouse を作成する

始める前に

必要なロール

ユーザーロール

サービス アカウントのロール

Managed Service for Apache Spark を構成して Spark セッションを作成する

クラスタ

プロパティを使用した簡略化された構成（推奨）

手動設定

認証情報ベンディングを使用して構成する

サーバーレス

プロパティを使用した簡略化された構成（推奨）

手動設定

認証情報ベンディングを使用して構成する

サーバーレス

プロパティを使用した簡素化された構成（推奨）

手動設定

認証情報ベンディングで構成する

Spark SQL を使用してデータを管理する

過去のスナップショットをクエリする

データリネージを検出する

データリネージの例

データリネージに関する既知の問題

次のステップ

Spark と Lakehouse ランタイムカタログを使用して Lakehouse を作成する

サービスアカウントのロール