Dataproc で BigQuery metastore を使用する

新しいクラスタ

新しい Dataproc クラスタを作成するには、次の gcloud dataproc clusters create コマンドを実行します。この構成には、BigQuery metastore の使用に必要な設定が含まれています。

gcloud dataproc clusters create CLUSTER_NAME \
  --project=PROJECT_ID \
  --region=LOCATION \
  --optional-components=ZEPPELIN \
  --enable-component-gateway \
  --single-node

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

• CLUSTER_NAME: Dataproc クラスタの名前。
• PROJECT_ID: クラスタを作成する Google Cloud プロジェクトの ID。
• LOCATION: クラスタを作成する Google Cloud リージョン。

既存のクラスタ

既存のクラスタを構成するには、次の Iceberg Spark ランタイムをクラスタに追加します。

org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.5.2

ランタイムは、次のいずれかの方法で追加できます。

• 初期化スクリプト。作成時に実行されるカスタム初期化スクリプトにランタイム依存関係を追加します。

  スクリプトにランタイム依存関係を追加したら、手順に沿ってクラスタの作成、再作成、更新を行います。

• 手動インストール。Iceberg カタログ プラグイン JAR ファイルを手動で追加し、クラスタにランタイムが含まれるように Spark プロパティを構成します。

gcloud CLI

gcloud dataproc jobs submit spark-sql \
--project=PROJECT_ID \
--cluster=CLUSTER_NAME \
--region==REGION \
--jars=https://storage-download.googleapis.com/maven-central/maven2/org/apache/iceberg/iceberg-spark-runtime-3.5_2.12/1.5.2/iceberg-spark-runtime-3.5_2.12-1.5.2.jar,gs://spark-lib/bigquery/iceberg-bigquery-catalog-1.5.2-1.0.1-beta.jar \
--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=WAREHOUSE_DIRECTORY \
--execute="SPARK_SQL_COMMAND"

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

• PROJECT_ID: Dataproc クラスタを含む Google Cloud プロジェクトの ID。
• CLUSTER_NAME: Spark SQL ジョブの実行に使用している Dataproc クラスタの名前。
• REGION: クラスタが配置されている Compute Engine リージョン。
• LOCATION: BigQuery リソースのロケーション。
• CATALOG_NAME: SQL ジョブで使用している Spark カタログの名前。
• WAREHOUSE_DIRECTORY: データ ウェアハウスを含む Cloud Storage フォルダ。この値は gs:// で始まります。
• SPARK_SQL_COMMAND: 実行する Spark SQL クエリ。このクエリには、リソースを作成するコマンドを含めます。たとえば、名前空間とテーブルを作成します。

インタラクティブ Spark

Spark に接続してカタログ プラグインをインストールする

BigQuery metastore のカタログ プラグインをインストールするには、SSH を使用して Dataproc クラスタに接続します。

1. Google Cloud コンソールで、[VM インスタンス] ページに移動します。

2. Dataproc VM インスタンスに接続するには、仮想マシン インスタンスのリストで [SSH] をクリックします。出力は次のようになります。

   Connected, host fingerprint: ssh-rsa ...
   Linux cluster-1-m 3.16.0-0.bpo.4-amd64 ...
   ...
   example-cluster@cluster-1-m:~$
   

3. ターミナルで、次の BigQuery metastore の初期化コマンドを実行します。

   spark-sql \
   --jars https://storage-download.googleapis.com/maven-central/maven2/org/apache/iceberg/iceberg-spark-runtime-3.5_2.12/1.5.2/iceberg-spark-runtime-3.5_2.12-1.5.2.jar,gs://spark-lib/bigquery/iceberg-bigquery-catalog-1.5.2-1.0.1-beta.jar \
   --conf spark.sql.catalog.CATALOG_NAME=org.apache.iceberg.spark.SparkCatalog \
   --conf spark.sql.catalog.CATALOG_NAME.catalog-impl=org.apache.iceberg.gcp.bigquery.BigQueryMetastoreCatalog \
   --conf spark.sql.catalog.CATALOG_NAME.gcp_project=PROJECT_ID \
   --conf spark.sql.catalog.CATALOG_NAME.gcp_location=LOCATION \
   --conf spark.sql.catalog.CATALOG_NAME.warehouse=WAREHOUSE_DIRECTORY

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

   • CATALOG_NAME: SQL ジョブで使用している Spark カタログの名前。
   • PROJECT_ID: Spark カタログがリンクされる BigQuery metastore カタログの Google Cloud プロジェクト ID。
   • LOCATION: BigQuery metastore の Google Cloud ロケーション。
   • WAREHOUSE_DIRECTORY: データ ウェアハウスを含む Cloud Storage フォルダ。この値は gs:// で始まります。

   クラスタに正常に接続すると、Spark ターミナルに「spark-sql」というプロンプトが表示されます。

   spark-sql (default)>
   

BigQuery metastore リソースを管理する

これで BigQuery metastore に接続されました。BigQuery metastore に保存されているメタデータに基づいて、既存のリソースを表示したり、新しいリソースを作成したりできます。

たとえば、インタラクティブ Spark SQL セッションで次のコマンドを実行して、Iceberg 名前空間とテーブルを作成します。

• カスタムの Iceberg カタログを使用する:

  USE `CATALOG_NAME`;

• 名前空間を作成する:

  CREATE NAMESPACE IF NOT EXISTS NAMESPACE_NAME;

• 作成した名前空間を使用する:

  USE NAMESPACE_NAME;

• Iceberg テーブルを作成する:

  CREATE TABLE TABLE_NAME (id int, data string) USING ICEBERG;

• テーブルに行を挿入する:

  INSERT INTO TABLE_NAME VALUES (1, "first row");

• テーブルの列を追加する:

  ALTER TABLE TABLE_NAME ADD COLUMNS (newDoubleCol double);

• テーブルのメタデータを表示する:

  DESCRIBE EXTENDED TABLE_NAME;

• 名前空間内のテーブルの一覧を取得する:

  SHOW TABLES;

Zeppelin ノートブック

1. Google Cloud コンソールで、Dataproc の [クラスタ] ページに移動します。

   Dataproc クラスタに移動

2. 使用するクラスタの名前をクリックします。

   [クラスタの詳細] ページが開きます。

3. ナビゲーション メニューで [ウェブ インターフェース] をクリックします。

4. [コンポーネント ゲートウェイ] で [Zeppelin] をクリックします。Zeppelin ノートブック ページが開きます。

5. ナビゲーション メニューで [Notebook]、[+ Create new note] の順にクリックします。

6. ダイアログでノートブック名を入力します。デフォルトのインタープリタとして [Spark] が選択されたままにします。

7. [Create] をクリックします。新しいノートブックが作成されます。

8. ノートブックで設定メニューをクリックし、[Interpreter] をクリックします。

9. [Search interpreters] フィールドで「Spark」を検索します。

10. [Edit] をクリックします。

11. [Spark.jars] フィールドに、Spark JAR の URI を入力します。

    https://storage-download.googleapis.com/maven-central/maven2/org/apache/iceberg/iceberg-spark-runtime-3.5_2.12/1.5.2/iceberg-spark-runtime-3.5_2.12-1.5.2.jar,gs://spark-lib/bigquery/iceberg-bigquery-catalog-1.5.2-1.0.1-beta.jar
    

12. [Save] をクリックします。

13. [OK] をクリックします。

14. 次の PySpark コードを Zeppelin ノートブックにコピーします。

    %pyspark
    from pyspark.sql import SparkSession
    spark = SparkSession.builder \
    .appName("BigQuery Metastore Iceberg") \
    .config("spark.sql.catalog.CATALOG_NAME", "org.apache.iceberg.spark.SparkCatalog") \
    .config("spark.sql.catalog.CATALOG_NAME.catalog-impl", "org.apache.iceberg.gcp.bigquery.BigQueryMetastoreCatalog") \
    .config("spark.sql.catalog.CATALOG_NAME.gcp_project", "PROJECT_ID") \
    .config("spark.sql.catalog.CATALOG_NAME.gcp_location", "LOCATION") \
    .config("spark.sql.catalog.CATALOG_NAME.warehouse", "WAREHOUSE_DIRECTORY") \
    .getOrCreate()
    spark.sql("select version()").show()
    spark.sql("USE `CATALOG_NAME`;")
    spark.sql("CREATE NAMESPACE IF NOT EXISTS NAMESPACE_NAME;")
    spark.sql("USE NAMESPACE_NAME;")
    spark.sql("CREATE TABLE TABLE_NAME (id int, data string) USING ICEBERG;")
    spark.sql("DESCRIBE TABLE_NAME;").show()

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

    • CATALOG_NAME: SQL ジョブに使用する Spark カタログの名前。
    • PROJECT_ID: Dataproc クラスタを含む Google Cloud プロジェクトの ID。
    • WAREHOUSE_DIRECTORY: データ ウェアハウスを含む Cloud Storage フォルダ。この値は gs:// で始まります。
    • NAMESPACE_NAME: Spark テーブルを参照する名前空間名。
    • WAREHOUSE_DIRECTORY: データ ウェアハウスが保存されている Cloud Storage フォルダの URI。
    • TABLE_NAME: Spark テーブルのテーブル名。

15. 実行アイコンをクリックするか、Shift-Enter キーを押してコードを実行します。ジョブが完了すると、「Spark Job Finished」というステータス メッセージが表示され、出力にテーブルの内容が表示されます。