設定 Spark 與 Delta 資料表

本文說明如何使用 Lakehouse 執行階段目錄設定 Apache Spark，以使用 Delta Lake 資料表。您將瞭解如何建立 Apache Hive 目錄、設定 Spark 工作階段以連線至 Metastore，以及執行工作負載來建立及註冊 Delta 資料表，以便在 BigQuery 中查詢。

注意：您無法直接透過 Lakehouse 執行階段目錄 Hive 端點，或使用 Hive DDL 建立新的 Delta 資料表。 請改用 Apache Spark 等與 Delta 相容的引擎，在 Cloud Storage 中建立基礎資料表檔案，並在 Metastore 中註冊資料表。

事前準備

1. 請參閱「關於 Hive 目錄端點」。
2. 請參閱「支援的儲存格式和資料類型」。
3. 查看限制和注意事項。
登入 Google Cloud 帳戶。如果您是 Google Cloud新手，歡迎 建立帳戶，親自體驗產品的實際應用成效。新客戶還能獲得價值 $300 美元的免費抵免額，用於執行、測試及部署工作負載。

Verify that billing is enabled for your Google Cloud project.

Enable the Lakehouse, Managed Service for Apache Spark 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

Verify that billing is enabled for your Google Cloud project.

Enable the Lakehouse, Managed Service for Apache Spark 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

必要的角色

如要取得使用 Lakehouse 執行階段目錄和 Delta 資料表所需的權限，請要求管理員授予您專案的下列 IAM 角色：

• 建立 Managed Service for Apache Spark 叢集： Dataproc 編輯器 (roles/dataproc.editor)
• 叢集服務帳戶：
  • Dataproc worker (roles/dataproc.worker)
  • Storage 物件使用者 (roles/storage.objectUser)
  • BigLake 編輯者 (roles/biglake.editor)
  • 服務使用情形消費者 (roles/serviceusage.serviceUsageConsumer)
• 所有 Lakehouse 執行階段目錄資源的寫入權限：BigLake 編輯者 (roles/biglake.editor)
• 所有 Lakehouse 執行階段目錄資源的唯讀存取權：BigLake 檢視者 (roles/biglake.viewer)

如要進一步瞭解如何授予角色，請參閱「管理專案、資料夾和組織的存取權」。

您或許也能透過自訂角色或其他預先定義的角色，取得必要權限。

如需操作說明，請參閱「授予單一角色」。

一般工作流程

如要搭配使用 Lakehouse 執行階段目錄和 Delta 資料表，請按照下列工作流程操作：

1. 建立 Lakehouse Hive 目錄。
2. 使用 Lakehouse metastore 用戶端和 Delta Lake 依附元件設定 Spark 工作階段。
3. 使用 Spark 在 Cloud Storage 中建立及填入 Delta 資料表，並向 Metastore 註冊。
4. 從 Apache Spark (讀取和寫入) 或直接從 BigQuery (唯讀) 查詢 Delta 資料表。

建立 Lakehouse Hive 目錄

如要搭配使用 Lakehouse 執行階段目錄與 Spark 和 Delta 資料表，請建立 Hive 目錄。

Lakehouse Hive 目錄是 Hive 資料庫的集合。執行 Spark 工作前，請先建立目錄，並向 Lakehouse Metastore 註冊。目錄必須有名稱和資料的 Cloud Storage 位置。

控制台

1. 在 Google Cloud 控制台開啟「Lakehouse」Lakehouse頁面。

   前往 Lakehouse

2. 按一下「建立目錄」。

3. 選取「Lakehouse runtime catalog」(Lakehouse 執行階段目錄)。

4. 在「Catalog type」(目錄類型) 部分，選取「Hive Metastore」。

5. 在「Select a Cloud Storage bucket」(選取 Cloud Storage bucket) 欄位中，輸入目錄的 Cloud Storage bucket 名稱。或者，按一下「Browse」(瀏覽) 選擇或建立 bucket。

6. 在「目錄 ID」中，輸入 Lakehouse Hive 目錄的名稱。

7. 在「Primary location」(主要位置)，指定與值區相同的區域。

8. 點選「建立」。

gcloud

如要建立 Hive 目錄，請執行下列指令：

gcloud biglake hive catalogs create LAKEHOUSE_CATALOG_ID \
    --project=PROJECT_ID \
    --location-uri="gs://GCS_WAREHOUSE_PATH" \
    --primary-location=REGION \
    --description="DESCRIPTION"

更改下列內容：

• LAKEHOUSE_CATALOG_ID：Hive 目錄名稱。

• GCS_WAREHOUSE_PATH：倉儲資料的 Cloud Storage 路徑。

• PROJECT_ID：您的 Google Cloud專案 ID。

• REGION：metastore 的主要區域。如果是單一區域值區，這會與值區區域相符。如果是雙區域或多區域 bucket，請為主要副本指定組成區域。另一個區域則會成為次要副本。

• DESCRIPTION：目錄說明。

設定 Spark 使用 Delta 資料表

如要搭配使用 Lakehouse 執行階段目錄和 Delta 資料表，請使用 Lakehouse metastore 用戶端工廠和 Delta Lake 擴充功能設定 Spark 工作階段。

spark-sql

1. 使用 SSH 連線至 Managed Service for Apache Spark 叢集的主要節點。

2. 在指令列中執行 spark-sql，並使用下列屬性啟動互動式 Spark SQL 工作階段：

   spark-sql \
       --conf spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension \
       --conf spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog \
       --conf spark.hive.metastore.client.factory.class=com.google.cloud.bigquery.metastore.client.BigLakeMetastoreClientFactory \
       --conf spark.hive.metastore.blms.project.id=PROJECT_ID \
       --conf spark.hive.metastore.blms.catalog.default=LAKEHOUSE_CATALOG_ID \
       --conf spark.hive.metastore.warehouse.dir=gs://GCS_WAREHOUSE_PATH
   

3. 工作階段開始後，Spark 會連線至 Lakehouse 執行階段目錄。

   執行下列指令，建立及查詢 Delta 資源：

   -- Show all the databases in the current project.
   SHOW DATABASES;
   
   -- Create a database.
   CREATE DATABASE spark_delta_blms_database;
   
   -- Create a Delta data source table.
   CREATE TABLE spark_delta_blms_database.delta_quick_start (id INT, name STRING) USING DELTA;
   
   -- Insert data into the table.
   INSERT INTO TABLE spark_delta_blms_database.delta_quick_start VALUES (1, 'my-first-user');
   
   -- Query the Delta table.
   SELECT * FROM spark_delta_blms_database.delta_quick_start;
   

   更改下列內容：

   • PROJECT_ID：您的 Google Cloud 專案 ID。

   • LAKEHOUSE_CATALOG_ID：Hive 目錄名稱。

   • GCS_WAREHOUSE_PATH：儲存倉儲資料的 Cloud Storage 路徑。

Jupyter 筆記本

1. 在 Managed Service for Apache Spark 叢集上執行 Jupyter 筆記本。
2. 在 Google Cloud 控制台，前往叢集詳細資料頁面的「網路介面」分頁，存取 Jupyter 網頁介面。

3. 在新的筆記本中，建立為 Delta Lake 設定的 Spark 工作階段，並執行查詢：

   from pyspark.sql import SparkSession
   
   # If a Spark session exists, stop it first by running spark.stop()
   spark = SparkSession.builder\
       .master("local")\
       .appName("Lakehouse runtime catalog Delta tutorial")\
       .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")\
       .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")\
       .config("spark.hive.metastore.client.factory.class", "com.google.cloud.bigquery.metastore.client.BigLakeMetastoreClientFactory")\
       .config("spark.hive.metastore.blms.project.id", "PROJECT_ID")\
       .config("spark.hive.metastore.warehouse.dir", "gs://GCS_WAREHOUSE_PATH")\
       .config("spark.hive.metastore.blms.catalog.default", "LAKEHOUSE_CATALOG_ID")\
       .getOrCreate()
   
   # Show all the databases.
   df = spark.sql("SHOW DATABASES;")
   df.show()
   
   # Create a database.
   spark.sql("CREATE DATABASE jupyter_delta_blms_db")
   
   # Create a Delta data source table.
   spark.sql("CREATE TABLE jupyter_delta_blms_db.delta_table(id INT, name STRING) USING DELTA")
   
   # Insert data into the table.
   spark.sql("INSERT INTO TABLE jupyter_delta_blms_db.delta_table VALUES (1, 'my-first-user');")
   
   # Query from table.
   spark.sql("SELECT * FROM jupyter_delta_blms_db.delta_table;").show()
   

   更改下列內容：

   • PROJECT_ID：您的 Google Cloud 專案 ID。

   • GCS_WAREHOUSE_PATH：儲存倉儲資料的 Cloud Storage 路徑。

   • LAKEHOUSE_CATALOG_ID：Hive 目錄名稱。

BigQuery 筆記本

1. 前往 Google Cloud 控制台的「BigQuery」頁面。

   前往「BigQuery」

2. 在「Explorer」窗格中，依序點選「+ ADD」和「Python notebook」。

3. 在程式碼儲存格中，使用 Delta 擴充功能和 Lakehouse 執行階段目錄屬性，設定 Managed Service for Apache Spark 工作階段：

   from google.cloud.dataproc_spark_connect import DataprocSparkSession
   from google.cloud.dataproc_v1 import Session
   import os
   
   os.environ['DATAPROC_SPARK_CONNECT_DEFAULT_DATASOURCE'] = ""
   
   session = Session()
   session.environment_config.execution_config.ttl = {"seconds": 864000}
   session.runtime_config.version = "2.3"
   session.runtime_config.properties = {
    "spark.sql.extensions": "io.delta.sql.DeltaSparkSessionExtension",
    "spark.sql.catalog.spark_catalog": "org.apache.spark.sql.delta.catalog.DeltaCatalog",
    "spark.hive.metastore.blms.project.id": "PROJECT_ID",
    "spark.hive.metastore.blms.catalog.default": "LAKEHOUSE_CATALOG_ID",
    "spark.hive.metastore.warehouse.dir": "gs://GCS_WAREHOUSE_PATH",
    "spark.hive.metastore.client.factory.class": "com.google.cloud.bigquery.metastore.client.BigLakeMetastoreClientFactory",
    "spark.sql.catalogImplementation": "hive"
   }
   
   spark = DataprocSparkSession.builder.dataprocSessionConfig(session).getOrCreate()
   print("Spark session created successfully")
   

4. 在另一個程式碼儲存格中，建立資料庫和 Delta 資料表：

   # Create a database
   spark.sql("CREATE DATABASE bq_spark_delta_database;")
   
   # Create a Delta data source table
   spark.sql("CREATE TABLE bq_spark_delta_database.delta_quick_start (id INT, name STRING) USING DELTA;")
   
   # Insert data into the table
   spark.sql("INSERT INTO TABLE bq_spark_delta_database.delta_quick_start VALUES (1, 'my-first-user');")
   
   # Query from table
   spark.sql("SELECT * FROM bq_spark_delta_database.delta_quick_start;").show()
   

   更改下列內容：

   • PROJECT_ID：您的 Google Cloud 專案 ID。

   • GCS_WAREHOUSE_PATH：儲存倉儲資料的 Cloud Storage 路徑。

   • LAKEHOUSE_CATALOG_ID：Hive 目錄名稱。

提交 Managed Service for Apache Spark 批次工作

您可以將 PySpark 批次工作負載提交至 Managed Service for Apache Spark，透過 Lakehouse 執行階段目錄建立及查詢 Delta 資料表。

1. 建立含有 PySpark 工作的 Python 檔案：

   from pyspark.sql import SparkSession
   
   spark = (
       SparkSession.builder.appName("Lakehouse runtime catalog Delta tutorial")
       .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
       .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
       .config(
           "spark.hive.metastore.client.factory.class",
           "com.google.cloud.bigquery.metastore.client.BigLakeMetastoreClientFactory",
       )
       .config("spark.hive.metastore.blms.project.id", "PROJECT_ID")
       .config("spark.hive.metastore.blms.catalog.default", "LAKEHOUSE_CATALOG_ID")
       .config(
           "spark.hive.metastore.warehouse.dir",
           "gs://GCS_WAREHOUSE_PATH",
       )
       .enableHiveSupport()
       .getOrCreate()
   )
   
   # Show all the databases.
   spark.sql("SHOW DATABASES;").show()
   
   # Create a database.
   spark.sql("CREATE DATABASE dp_serverless_delta_db")
   
   # Create a Delta data source table.
   spark.sql(
       "CREATE TABLE dp_serverless_delta_db.delta_table(id INT, name STRING) USING DELTA"
   )
   
   # Insert data into the table.
   spark.sql("INSERT INTO TABLE dp_serverless_delta_db.delta_table VALUES (1, 'my-first-user');")
   
   # Query from table.
   spark.sql("SELECT * FROM dp_serverless_delta_db.delta_table;").show()
   

   更改下列內容：

   • PROJECT_ID：您的 Google Cloud 專案 ID。

   • LAKEHOUSE_CATALOG_ID：Hive 目錄名稱。

   • GCS_WAREHOUSE_PATH：儲存倉儲資料的 Cloud Storage 路徑。

2. 提交批次工作：

   gcloud dataproc batches submit pyspark PYTHON_SCRIPT_FILE \
       --version=2.2 \
       --project=PROJECT_ID \
       --region=REGION \
       --deps-bucket=gs://CLOUD_STORAGE_BUCKET
   

   更改下列內容：

   • PYTHON_SCRIPT_FILE：PySpark 應用程式檔案的本機或 Cloud Storage 路徑。
   • PROJECT_ID：您的專案 ID。
   • REGION：執行批次工作的區域。
   • CLOUD_STORAGE_BUCKET：用於暫存工作負載依附元件的 Cloud Storage bucket。

透過 BigQuery 查詢 Delta 資料表

在 Lakehouse 執行階段目錄中透過 Spark 建立 Delta 資料表後，即可從 BigQuery Studio 查詢這些資料表。

從 BigQuery 對 Hive 目錄端點中的 Delta 資料表執行的查詢，皆為唯讀。

1. 前往 Google Cloud 控制台的「BigQuery」頁面。

   前往「BigQuery」

2. 在查詢編輯器中輸入下列陳述式：

   SELECT * FROM `PROJECT_ID.LAKEHOUSE_CATALOG_ID.DATABASE_NAME.TABLE_NAME`;
   

   更改下列內容：

   • PROJECT_ID：您的專案 ID。
   • LAKEHOUSE_CATALOG_ID：您的 Hive 目錄名稱。
   • DATABASE_NAME：資料庫名稱。
   • TABLE_NAME：Delta 資料表名稱。

限制

使用 Delta 資料表搭配 Lakehouse 執行階段目錄 Hive 目錄端點時，請留意下列限制：

• BigQuery 中的唯讀查詢：從 BigQuery 對 Delta 資料表執行的查詢為唯讀。使用 Apache Spark 執行資料修改、DML 作業和結構定義變更。
• GetHiveTable 上的預留位置結構定義和實際結構定義檢查：GetHiveTable 對 Delta 資料表的呼叫會傳回預留位置結構定義，而非實際資料表結構定義。如要檢查 Delta 資料表的實際結構定義，請使用 Apache Spark。BigQuery 會計算修改後的結構定義，以利執行 Dremel 查詢，這與實際的 Delta 資料表結構定義並非 1:1 對應。
• Hive 用戶端不相容：Hive 用戶端 (例如 Hive CLI) 不支援使用 Lakehouse 執行階段目錄 Hive 端點的 Delta 資料表。使用 Apache Spark 管理 Delta 資料表，並透過 Spark 或 BigQuery 查詢。
• 不支援目錄管理作業：Delta 表格不支援目錄管理模式。您無法使用獨立 metastore DDL 建立 Delta 資料表。請改為在 Cloud Storage 中建立具有外部儲存位置的 Delta 資料表 (例如使用 Apache Spark)，並在目錄中註冊。
• BigQuery 外部 Delta 資料表限制：Lakehouse 執行階段目錄 Hive 端點使用的基礎查詢引擎實作方式，與 BigQuery 外部 Delta Lake 資料表相同。所有外部 Delta Lake 資料表限制和 BigLake 資料表限制均適用。舉例來說，不支援變更資料擷取 (CDC) 作業和 BigQuery Storage Read API，也不支援 timestamp_ntz。

後續步驟

• 關於 Hive 目錄端點。
• 支援的儲存格式和資料類型。
• Hive 目錄的限制和注意事項。