"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 ランタイムカタログでクロスクラウドデータアクセスを設定すると、ファイルを移行したり、個別の Cloud Storage 認証情報を管理したりすることなく、Managed Service for Apache Spark からリモート Iceberg テーブルにクエリを実行できます。

Lakehouse ランタイムカタログは、メタデータの同期と、認証情報ベンディングによる安全なデータアクセスを処理します。認証情報ベンディングSpark が連携テーブルにクエリを実行すると、Lakehouse ランタイムカタログはカタログメタデータを変換し、Spark が Amazon S3 または Cloud Storage からデータを読み取るための有効期間の短いスコープ付き認証情報を発行します。これにより、テーブルはローカルGoogle Cloud 環境にあるかのように扱われます。

サポートされている連携カタログのタイプ

Managed Service for Apache Spark は、Lakehouse ランタイムカタログの Iceberg REST カタログエンドポイントを介して、次の連携カタログタイプへのクエリをサポートしています。

連携カタログのタイプ	サポートされているストレージバックエンド	Lakehouse ランタイムカタログの設定ガイド
AWS Glue Data Catalog	Amazon S3（`s3://` - デフォルトの AWS アカウントカタログまたは Amazon S3 テーブルバケット）	AWS Glue に接続する
Databricks Unity Catalog	Amazon S3（`s3://`）または Cloud Storage（`gs://`）の外部ロケーション	Databricks Unity Catalog に接続する
Snowflake Horizon Catalog	Amazon S3（`s3://`）または Cloud Storage（`gs://`）	Snowflake Horizon Catalog に接続する

カタログの準備要件と、各ストレージプロバイダの基盤となる Spark パラメータ定義については、 Lakehouse ランタイムカタログでリモートデータにクエリを実行すると Spark 構成パラメータをご覧ください。

始める前に

アカウントに Google Cloud ログインします。を初めて使用する場合は、アカウントを作成して、実際のシナリオで Google プロダクトのパフォーマンスを評価してください。 Google Cloud新規のお客様には、ワークロードの実行、テスト、デプロイができる無料クレジット $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, and BigQuery 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, and BigQuery APIs, if any are not already enabled.

Roles required to enable APIs

Enable the APIs

リモートカタログに Iceberg テーブルがあることを確認します。
リモートカタログプロバイダの Lakehouse ランタイムカタログで、クロスクラウド接続と連携カタログを設定します。

必要な Identity and Access Management（IAM）ロール

このページの例を実行するには、特定の 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 以降でエンドユーザー認証情報を使用して連携テーブルを読み取る: BigLake 閲覧者（roles/biglake.viewer）連携カタログを含むプロジェクトに対する
サーバーレスランタイム 3.0 以降でサービスアカウントの認証情報を使用する: サービスアカウントユーザー（roles/iam.serviceAccountUser）カスタムサービスアカウントに対する

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

ワークロードを実行するサービスアカウントに次のロールを付与します。 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.viewer）連携カタログを含むプロジェクトに対する

クラスタの作成、バッチワークロードの送信、インタラクティブセッションの作成時に、Managed Service for Apache Spark サービスエージェント（service-PROJECT_NUMBER@dataproc-accounts.iam.gserviceaccount.com）に dataproc.lakehouse.catalog. 簡略化された構成プロパティを解決するために必要な権限が付与されるようにするには、連携カタログを含むプロジェクトの Managed Service for Apache Spark サービスエージェント（service-PROJECT_NUMBER@dataproc-accounts.iam.gserviceaccount.com）に BigLake 閲覧者（roles/biglake.viewer）IAM ロールを付与するよう管理者に依頼してください。

重要: このロールは、ユーザーアカウントではなく、 Managed Service for Apache Spark サービスエージェント（service-PROJECT_NUMBER@dataproc-accounts.iam.gserviceaccount.com）に付与する必要があります。正しいプリンシパルにロールを付与しないと、権限エラーが発生する場合があります。

Amazon S3 のアウトバウンドネットワーク接続を構成する

Amazon S3 を基盤とする連携カタログ、 Managed Service for Apache Spark サーバーレスバッチワークロードとインタラクティブセッション、内部 IP Managed Service for Apache Spark クラスタにクエリを実行する場合、デフォルトのネットワーク構成では Amazon S3 エンドポイントに接続できません。ワークロード VPC ネットワークに Cloud Router と Cloud NAT をプロビジョニングして、Amazon S3 へのアウトバウンドトラフィックを許可します。

gcloud compute routers create lakehouse-router \
    --network=NETWORK_NAME \
    --region=REGION

gcloud compute routers nats create lakehouse-nat \
    --router=lakehouse-router \
    --auto-allocate-nat-external-ips \
    --nat-all-subnet-ip-ranges \
    --region=REGION

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

NETWORK_NAME: Managed Service for Apache Spark クラスタ、バッチワークロード、またはインタラクティブセッションの VPC ネットワーク（例: default）。
REGION: Managed Service for Apache Spark クラスタ、バッチワークロード、または Google Cloudインタラクティブセッションのリージョン。

Managed Service for Apache Spark を構成して連携テーブルにクエリを実行する

Managed Service for Apache Spark クラスタ、サーバーレスバッチワークロード、またはインタラクティブセッションを構成して、Lakehouse ランタイムカタログの連携カタログに接続できます。これらの Managed Service for Apache Spark リソースは、 dataproc.lakehouse.catalog.CATALOG_NAME 簡略化された構成プロパティ（推奨）をサポートしています。このプロパティは、Lakehouse ランタイムカタログから連携カタログ定義を読み取り、ストレージ io-impl や認証情報ベンディングヘッダーなど、必要な Iceberg REST カタログの Spark プロパティを生成します（プロパティのクラスタ形式では dataproc: 接頭辞が追加されます）。

簡略化された構成と手動構成の両方で、Iceberg（GoogleAuthManager）に Google 認証を使用します。これには、次のいずれかのクラスタイメージまたはサーバーレスランタイムバージョンが必要です。

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 以降では、デフォルトでエンドユーザー認証情報が使用されます。詳細については、必要な IAM ロールとサーバーレスデプロイのロールをご覧ください。

クラスタ

簡略化された構成プロパティ（推奨）を使用して、またはカタログプロパティを手動で指定して、Iceberg オプションコンポーネントを含む 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.lakehouse.catalog.CATALOG_NAME=projects/PROJECT_ID/catalogs/FEDERATED_CATALOG_NAME,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 カタログの名前（federated_catalog など）。FEDERATED_CATALOG_NAME と同じ値を指定できます。
FEDERATED_CATALOG_NAME：始める前にで作成した連携カタログの ID。

Snowflake Horizon Catalog にクエリを実行する

Snowflake Horizon Catalog にクエリを実行する場合は、次の点に注意してください。

Snowflake ストレージ: 簡略化された構成では、 Amazon S3 に保存されている Snowflake テーブルがサポートされます。Cloud Storage に保存されている Snowflake テーブルの場合は、手動構成を io-impl に設定して org.apache.iceberg.gcp.gcs.GCSFileIO ます。
Snowflake STRING 列: Snowflake のテーブルのストレージシリアル化ポリシーが COMPATIBLE でない Snowflake STRING 列にクエリを実行するとエラーが発生する場合は、Snowflake でストレージシリアル化ポリシーを COMPATIBLE に変更するか、次のプロパティを --properties フラグに追加して Spark で Parquet ベクトル化を無効にできます。
```
spark:spark.sql.iceberg.vectorization.enabled=false,spark:spark.sql.catalog.CATALOG_NAME.table-override.read.parquet.vectorization.enabled=false
    
```

Jupyter ノートブックまたは PySpark アプリケーションで、カタログ構成を指定せずに SparkSession を作成します。

from pyspark.sql import SparkSession

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

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

手動設定

簡略化された構成プロパティを使用しない場合は、このセクションで説明したように、dataproc:dataproc.lakehouse.catalog. プロパティを --properties フラグに渡さずにクラスタを作成し、SparkSession を手動で構成します。ストレージプロバイダのパラメータ値の一覧については、 Spark 構成パラメータをご覧ください。

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', 'bl://projects/PROJECT_ID/catalogs/FEDERATED_CATALOG_NAME') \
.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}.header.X-Iceberg-Access-Delegation', 'vended-credentials') \
.config(f'spark.sql.catalog.{catalog_name}.io-impl', 'IO_IMPL') \
.config(f'spark.sql.catalog.{catalog_name}.s3.client-factory-impl', 'org.apache.iceberg.aws.s3.DefaultS3FileIOAwsClientFactory') \
.config('spark.sql.extensions', 'org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions') \
.getOrCreate()

IO_IMPL は、テーブルを含むストレージの FileIO 実装に置き換えます。

Amazon S3 のテーブル: org.apache.iceberg.aws.s3.S3FileIO
Cloud Storage のテーブル: org.apache.iceberg.gcp.gcs.GCSFileIO

s3.client-factory-impl プロパティは、Amazon S3 のテーブルにのみ適用されます。Cloud Storage テーブルの場合は、この行を省略します。

サーバーレス

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

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

バッチワークロードを送信するときに、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/FEDERATED_CATALOG_NAME,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 など）。サポートされているバージョンについては、このセクションの要件をご覧ください。必要な IAM ロールはランタイムバージョンによって異なります。
CATALOG_NAME: ローカル Spark カタログの名前（federated_catalog など）。FEDERATED_CATALOG_NAME と同じ値を指定できます。
FEDERATED_CATALOG_NAME：始める前にで作成した連携カタログの ID。

Snowflake Horizon Catalog にクエリを実行する

Snowflake Horizon Catalog にクエリを実行する場合は、次の点に注意してください。

Snowflake ストレージ: 簡略化された構成では、 Amazon S3 に保存されている Snowflake テーブルがサポートされます。Cloud Storage に保存されている Snowflake テーブルの場合は、手動構成を io-impl に設定して org.apache.iceberg.gcp.gcs.GCSFileIO ます。
Snowflake STRING 列: Snowflake のテーブルのストレージシリアル化ポリシーが COMPATIBLE でない Snowflake STRING 列にクエリを実行するとエラーが発生する場合は、Snowflake でストレージシリアル化ポリシーを COMPATIBLE に変更するか、次のプロパティを --properties フラグに追加して Spark で Parquet ベクトル化を無効にできます。
```
spark.sql.iceberg.vectorization.enabled=false,spark.sql.catalog.CATALOG_NAME.table-override.read.parquet.vectorization.enabled=false
    
```

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

from pyspark.sql import SparkSession

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

手動設定

簡略化された構成プロパティを使用しない場合は、gcloud dataproc batches submit コマンドの --properties フラグを使用してカタログプロパティを指定します。ストレージプロバイダのパラメータ値の一覧については、 Spark 構成パラメータをご覧ください。

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=bl://projects/PROJECT_ID/catalogs/FEDERATED_CATALOG_NAME,\
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.header.X-Iceberg-Access-Delegation=vended-credentials,\
spark.sql.catalog.CATALOG_NAME.io-impl=IO_IMPL,\
spark.sql.catalog.CATALOG_NAME.s3.client-factory-impl=org.apache.iceberg.aws.s3.DefaultS3FileIOAwsClientFactory,\
spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions"

IO_IMPL は、テーブルを含むストレージの FileIO 実装に置き換えます。

Amazon S3 のテーブル: org.apache.iceberg.aws.s3.S3FileIO
Cloud Storage のテーブル: org.apache.iceberg.gcp.gcs.GCSFileIO

s3.client-factory-impl プロパティは、Amazon S3 のテーブルにのみ適用されます。Cloud Storage テーブルの場合は、このプロパティを省略します。

サーバーレス

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

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

dataproc-spark-connect パッケージがインストールされているノートブック（BigQuery Studio ノートブックや Vertex AI Workbench ノートブックなど）で、セッションを作成するときに 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/FEDERATED_CATALOG_NAME') \
.config('spark.sql.defaultCatalog', catalog_name) \
.getOrCreate()

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

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

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

Snowflake Horizon Catalog にクエリを実行する

Snowflake Horizon Catalog にクエリを実行する場合は、次の点に注意してください。

Snowflake ストレージ: 簡略化された構成では、 Amazon S3 に保存されている Snowflake テーブルがサポートされます。Cloud Storage に保存されている Snowflake テーブルの場合は、手動構成を io-impl に設定して org.apache.iceberg.gcp.gcs.GCSFileIO ます。
Snowflake STRING 列: Snowflake のテーブルのストレージシリアル化ポリシーが COMPATIBLE でない Snowflake STRING 列にクエリを実行するとエラーが発生する場合は、Snowflake でストレージシリアル化ポリシーを COMPATIBLE に変更するか、次の .config() 呼び出しをセッションビルダーに追加して Spark で Parquet ベクトル化を無効にできます。
```
.config('spark.sql.iceberg.vectorization.enabled', 'false') \
.config(f'spark.sql.catalog.{catalog_name}.table-override.read.parquet.vectorization.enabled', 'false')
    
```

手動設定

簡略化された構成プロパティを使用しない場合は、セッションの作成時にカタログプロパティを指定します。ストレージプロバイダのパラメータ値の一覧については、 Spark 構成パラメータをご覧ください。

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', 'bl://projects/PROJECT_ID/catalogs/FEDERATED_CATALOG_NAME') \
.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}.header.X-Iceberg-Access-Delegation', 'vended-credentials') \
.config(f'spark.sql.catalog.{catalog_name}.io-impl', 'IO_IMPL') \
.config(f'spark.sql.catalog.{catalog_name}.s3.client-factory-impl', 'org.apache.iceberg.aws.s3.DefaultS3FileIOAwsClientFactory') \
.config('spark.sql.extensions', 'org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions') \
.getOrCreate()

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

IO_IMPL は、テーブルを含むストレージの FileIO 実装に置き換えます。

Amazon S3 のテーブル: org.apache.iceberg.aws.s3.S3FileIO
Cloud Storage のテーブル: org.apache.iceberg.gcp.gcs.GCSFileIO

s3.client-factory-impl プロパティは、Amazon S3 のテーブルにのみ適用されます。Cloud Storage テーブルの場合は、この行を省略します。

連携テーブルにクエリを実行する

Spark セッションを構成したら、PySpark または Spark SQL を使用して連携カタログのテーブルにクエリを実行できます。

PySpark DataFrame API を使用して連携テーブルを読み取ります。
```
df = spark.table("CATALOG_NAME.NAMESPACE_NAME.TABLE_NAME")
df.show(10, truncate=False)
```
次のように置き換えます。
- CATALOG_NAME: ローカル Spark カタログ名。spark.sql.defaultCatalog を CATALOG_NAME に設定した場合は、この接頭辞を省略できます。
- NAMESPACE_NAME: 連携カタログの Namespace（データベースまたはスキーマ）。
- TABLE_NAME: 連携カタログのテーブル名。

Spark SQL を使用して連携テーブルにクエリを実行します。

spark.sql("""
    SELECT *
    FROM CATALOG_NAME.NAMESPACE_NAME.TABLE_NAME
    LIMIT 10
""").show()

連携カタログとローカルカタログ間でデータを結合またはマテリアライズする

連携カタログは読み取り専用です。リモートクロスクラウドデータをローカルテーブルと結合するか、連携クエリの結果を Cloud Storage にマテリアライズするには、カンマ区切りの dataproc.lakehouse.catalog. プロパティを使用して、連携カタログと読み取り / 書き込みカタログを Lakehouse ランタイムカタログに接続します。

クラスタ

Managed Service for Apache Spark クラスタの例:

--properties="dataproc:dataproc.lakehouse.catalog.FEDERATED_SPARK_CATALOG=projects/PROJECT_ID/catalogs/FEDERATED_CATALOG_NAME,dataproc:dataproc.lakehouse.catalog.LOCAL_SPARK_CATALOG=projects/PROJECT_ID/catalogs/LOCAL_CATALOG_ID,spark:spark.sql.defaultCatalog=LOCAL_SPARK_CATALOG"

サーバーレス

Managed Service for Apache Spark サーバーレスバッチワークロードの例:

--properties="dataproc.lakehouse.catalog.FEDERATED_SPARK_CATALOG=projects/PROJECT_ID/catalogs/FEDERATED_CATALOG_NAME,dataproc.lakehouse.catalog.LOCAL_SPARK_CATALOG=projects/PROJECT_ID/catalogs/LOCAL_CATALOG_ID,spark.sql.defaultCatalog=LOCAL_SPARK_CATALOG"

サーバーレス

Managed Service for Apache Spark サーバーレスインタラクティブセッションの例:

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

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

spark = DataprocSparkSession.builder.appName("APP_NAME") \
  .dataprocSessionConfig(session_config) \
  .config('dataproc.lakehouse.catalog.FEDERATED_SPARK_CATALOG', 'projects/PROJECT_ID/catalogs/FEDERATED_CATALOG_NAME') \
  .config('dataproc.lakehouse.catalog.LOCAL_SPARK_CATALOG', 'projects/PROJECT_ID/catalogs/LOCAL_CATALOG_ID') \
  .config('spark.sql.defaultCatalog', 'LOCAL_SPARK_CATALOG') \
  .getOrCreate()

このページのインタラクティブセッションの例で説明されているように、RUNTIME_VERSION を置き換えます。

これらのプロパティを設定すると、連携カタログにクエリを実行し、結果をローカル Lakehouse ランタイムカタログのテーブルに書き込むことができます。

ワークロードを実行するプリンシパルは、クラスタ、サーバーレスランタイム 2.3 以前、サービスアカウントの認証情報を使用するサーバーレスランタイム 3.0 以降のサービスアカウントです。エンドユーザー認証情報（デフォルト）を使用するサーバーレスランタイム 3.0 以降の場合、プリンシパルはユーザーアカウントです。

ワークロードを実行するプリンシパルに、テーブルデータをローカルカタログに書き込むために必要な権限が付与されるようにするには、ワークロードを実行するプリンシパルに次の IAM ロールを付与するよう管理者に依頼してください。

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

次の例では、連携テーブルからローカルカタログにテーブルを作成します。

spark.sql("""
    CREATE TABLE IF NOT EXISTS LOCAL_SPARK_CATALOG.LOCAL_NAMESPACE.materialized_table
    USING iceberg
    AS SELECT *
    FROM FEDERATED_SPARK_CATALOG.REMOTE_NAMESPACE.REMOTE_TABLE
""")

次のステップ

Spark と Lakehouse ランタイムカタログを使用してレイクハウスを作成する。
Lakehouse ランタイムカタログでリモートデータにクエリを実行するの詳細を確認する。
クロスクラウドデータアクセスについてを読む。
認証情報ベンディングについて学習する。

Spark と Lakehouse ランタイム カタログを使用して連携カタログにクエリを実行する

サポートされている連携カタログのタイプ

始める前に

必要な Identity and Access Management（IAM）ロール

ユーザーロール

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

Amazon S3 のアウトバウンド ネットワーク接続を構成する

Managed Service for Apache Spark を構成して連携テーブルにクエリを実行する

クラスタ

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

Snowflake Horizon Catalog にクエリを実行する

手動設定

サーバーレス

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

Snowflake Horizon Catalog にクエリを実行する

手動設定

サーバーレス

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

Snowflake Horizon Catalog にクエリを実行する

手動設定

連携テーブルにクエリを実行する

連携カタログとローカル カタログ間でデータを結合またはマテリアライズする

クラスタ

サーバーレス

サーバーレス

次のステップ

Spark と Lakehouse ランタイムカタログを使用して連携カタログにクエリを実行する

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

Amazon S3 のアウトバウンドネットワーク接続を構成する

連携カタログとローカルカタログ間でデータを結合またはマテリアライズする