"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 翻訳には誤りが含まれる場合があります。

Lakehouse ランタイムカタログでメタデータを使用して Apache Iceberg テーブルを作成する

このドキュメントでは、Managed Service for Apache Spark SQL と PySpark のバッチワークロードを実行して、Lakehouse ランタイムカタログにメタデータが保存された Apache Iceberg テーブルを作成する方法について説明します。Spark コードを実行するその他の方法については、BigQuery ノートブックで PySpark コードを実行すると Apache Spark ワークロードを実行するをご覧ください。

始める前に

Google Cloud プロジェクトと Cloud Storage バケットをまだ作成していない場合は、作成します。

プロジェクトを設定する
プロジェクトの Cloud Storage バケットを作成します。
1. Google Cloud コンソールで、Cloud Storage の [バケット] ページに移動します。
  [バケット] に移動
2. [ 作成] をクリックします。
3. [バケットの作成] ページでユーザーのバケット情報を入力します。次のステップに進むには、[続行] をクリックします。
  1. [始める] セクションで、次の操作を行います。
    - バケット名の要件を満たす、グローバルに一意の名前を入力します。
    - バケットラベルを追加するには、[ラベル] セクション（）を開き、[ ラベルを追加] をクリックして、ラベルの key と value を指定します。
  2. [データの保存場所の選択] セクションで、次の操作を行います。
    1. ロケーションタイプを選択してください。
    2. [ロケーションタイプ] プルダウンメニューから、バケットのデータが永続的に保存されるロケーションを選択します。
      - ロケーションタイプとして [デュアルリージョン] を選択した場合は、関連するチェックボックスを使用してターボレプリケーションを有効にすることもできます。
    3. クロスバケットレプリケーションを設定するには、[Storage Transfer Service 経由でクロスバケットレプリケーションを追加する] を選択し、次の手順を実施します。
      クロスバケットレプリケーションを設定する
      
      [バケット] メニューで、バケットを選択します。
      
      [レプリケーション設定] セクションで、[構成] をクリックして、レプリケーションジョブの設定を構成します。
      
      [クロスバケットレプリケーションを構成する] ペインが表示されます。
      
      オブジェクト名の接頭辞で複製するオブジェクトをフィルタするには、オブジェクトを追加または除外する接頭辞を入力し、 [接頭辞を追加] をクリックします。
      
      複製されたオブジェクトのストレージクラスを設定するには、[ストレージクラス] メニューからストレージクラスを選択します。この手順をスキップすると、複製されたオブジェクトはデフォルトで宛先バケットのストレージクラスを使用します。
      
      [完了] をクリックします。
  3. [データの保存場所を選択する] セクションで、次の操作を行います。
    1. バケットのデフォルトのストレージクラスを選択するか、バケットデータのストレージクラスを自動的に管理する Autoclass を選択します。
    2. 階層名前空間を有効にするには、[データ量が多いワークロード向けにストレージを最適化] セクションで、[このバケットで階層的な名前空間を有効にする] を選択します。
      注: 既存のバケットで階層型名前空間を有効にすることはできません。
  4. [オブジェクトへのアクセスを制御する方法を選択する] セクションで、バケットに公開アクセスの防止を適用するかどうかを選択し、バケットのオブジェクトに使用するアクセス制御方法を選択します。
    注: この設定が組織のポリシーで適用されている場合、[公開アクセスの防止] 設定を変更することはできません。
  5. [オブジェクトデータを保護する方法を選択する] セクションで、次の操作を行います。
    - [データ保護] で、バケットに設定するオプションを選択します。
      - 削除（復元可能）を有効にするには、[削除（復元可能）ポリシー（データ復元用）] チェックボックスをオンにして、削除後にオブジェクトを保持する日数を指定します。
      - オブジェクトのバージョニングを設定するには、[オブジェクトのバージョニング（バージョン管理用）] チェックボックスをオンにして、オブジェクトあたりの最大バージョン数と、非現行バージョンの有効期限が切れるまでの日数を指定します。
      - オブジェクトとバケットで保持ポリシーを有効にするには、[保持（コンプライアンス用）] チェックボックスをオンにして、次の操作を行います。
        
        オブジェクト保持ロックを有効にするには、[オブジェクト保持を有効にする] チェックボックスをオンにします。
        
        バケットロックを有効にするには、[バケット保持ポリシーを設定] チェックボックスをオンにして、保持期間の単位と期間を選択します。
    - オブジェクトデータの暗号化方法を選択するには、[データ暗号化] セクション（）を開き、データ暗号化方法を選択します。
4. [作成] をクリックします。
Compute Engine のデフォルトサービスアカウント（PROJECT_NUMBER-compute@developer.gserviceaccount.com）に BigQuery データ編集者（roles/bigquery.dataEditor）ロールを付与します。手順については、単一ロールの付与をご覧ください。

Google Cloud CLI の例:
```
gcloud projects add-iam-policy-binding PROJECT_ID \
    --member PROJECT_NUMBER-compute@developer.gserviceaccount.com \
--role roles/bigquery.dataEditor
```
注:
- PROJECT_ID と PROJECT_NUMBER は、 Google Cloud コンソールのダッシュボードの [プロジェクト情報] セクションに表示されます。

OSS リソースと BigQuery リソースのマッピング

オープンソースリソースと BigQuery リソースの用語のマッピングは次のとおりです。

OSS リソース	BigQuery リソース
名前空間、データベース	データセット
パーティション分割ありまたはパーティション分割なしのテーブル	テーブル
ビュー	ビュー

Iceberg テーブルを作成する

このセクションでは、Managed Service for Apache Spark の Spark SQL と PySpark バッチワークロードを使用して、メタデータが Lakehouse ランタイムカタログに保存された Iceberg テーブルを作成する方法について説明します。

Spark SQL

Spark SQL ワークロードを実行して Iceberg テーブルを作成する

次の手順では、Managed Service for Apache Spark SQL バッチワークロードを実行して、テーブルメタデータが Lakehouse ランタイムカタログに保存された Iceberg テーブルを作成する方法を示します。

次の Spark SQL コマンドをローカルまたは Cloud Shell の iceberg-table.sql ファイルにコピーします。

USE CATALOG_NAME;
CREATE NAMESPACE IF NOT EXISTS example_namespace;
USE example_namespace;
DROP TABLE IF EXISTS example_table;
CREATE TABLE example_table (id int, data string) USING ICEBERG LOCATION 'gs://BUCKET/WAREHOUSE_FOLDER';
INSERT INTO example_table VALUES (1, 'first row');
ALTER TABLE example_table ADD COLUMNS (newDoubleCol double);
DESCRIBE TABLE example_table;

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

CATALOG_NAME: Iceberg カタログ名。
BUCKET と WAREHOUSE_FOLDER: Iceberg ウェアハウスのディレクトリとして使用する Cloud Storage バケットとフォルダ。

iceberg-table.sql を含むディレクトリからローカルまたは Cloud Shell で次のコマンドを実行して、Spark SQL ワークロードを送信します。
```
gcloud dataproc batches submit spark-sql iceberg-table.sql \
    --project=PROJECT_ID \
    --region=REGION \
    --deps-bucket=BUCKET_NAME \
    --version=2.2 \
    --subnet=SUBNET_NAME \
    --properties="spark.sql.catalog.CATALOG_NAME=org.apache.iceberg.spark.SparkCatalog,spark.sql.catalog.CATALOG_NAME.catalog-impl=org.apache.iceberg.gcp.bigquery.BigQueryMetastoreCatalog,spark.sql.catalog.CATALOG_NAME.gcp_project=PROJECT_ID,spark.sql.catalog.CATALOG_NAME.gcp_location=LOCATION,spark.sql.catalog.CATALOG_NAME.warehouse=gs://BUCKET/WAREHOUSE_FOLDER"
  
```
注:
- PROJECT_ID: 実際の Google Cloud プロジェクト ID。プロジェクト ID は、 Google Cloud コンソールのダッシュボードの [プロジェクト情報] セクションに表示されます。
- REGION: ワークロードを実行できる利用可能な Compute Engine リージョン。
- BUCKET_NAME: Cloud Storage バケットの名前。Spark は、バッチワークロードを実行する前に、ワークロードの依存関係をこのバケットの /dependencies フォルダにアップロードします。WAREHOUSE_FOLDER はこのバケットにあります。
- --version: Managed Service for Apache Spark ランタイムバージョン 2.2 以降。
- SUBNET_NAME: REGION 内の VPC サブネットの名前。このフラグを省略すると、Managed Service for Apache Spark はセッションリージョンの default サブネットを選択します。Managed Service for Apache Spark は、サブネットでプライベート Google アクセス（PGA）を有効にします。ネットワーク接続の要件については、Managed Service for Apache Spark のネットワーク構成をご覧ください。
- LOCATION: サポートされている BigQuery のロケーション。デフォルトのロケーションは「US」です。
- --properties カタログプロパティ。
BigQuery でテーブルのメタデータを確認します。
1. Google Cloud コンソールで、[BigQuery] ページに移動します。
  
  [BigQuery Studio] に移動
2. Iceberg テーブルのメタデータを確認します。

PySpark

次の手順では、Managed Service for Apache Spark PySpark バッチワークロードを実行して、テーブルメタデータが Lakehouse ランタイムカタログに保存された Iceberg テーブルを作成する方法を示します。

次の PySpark コードをローカルまたは Cloud Shell の iceberg-table.py ファイルにコピーします。

from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("iceberg-table-example").getOrCreate()

catalog = "CATALOG_NAME"
namespace = "NAMESPACE"

spark.sql(f"USE `{catalog}`;")
spark.sql(f"CREATE NAMESPACE IF NOT EXISTS `{namespace}`;")
spark.sql(f"USE `{namespace}`;")

# Create table and display schema
spark.sql("DROP TABLE IF EXISTS example_iceberg_table")
spark.sql("CREATE TABLE example_iceberg_table (id int, data string) USING ICEBERG") 
spark.sql("DESCRIBE example_iceberg_table;")

# Insert table data.
spark.sql("INSERT INTO example_iceberg_table VALUES (1, 'first row');")

# Alter table, then display schema. 
spark.sql("ALTER TABLE example_iceberg_table ADD COLUMNS (newDoubleCol double);")
spark.sql("DESCRIBE example_iceberg_table;")