使用 Spark 和 Lakehouse 运行时目录创建 Lakehouse

湖仓一体架构将数据湖的灵活性与数据仓库的数据管理功能相结合。本文档介绍了如何在 Google Cloud上设置湖仓一体。您使用 Apache Iceberg 作为表格式，使用 Managed Service for Apache Spark 进行处理，并使用 Lakehouse 运行时目录 Iceberg REST Catalog 进行统一元数据管理。

此架构使用 Iceberg 等开放式表格式，为 Cloud Storage 中的数据添加数据仓储功能，例如事务和架构演变。此方法可为您的数据创建单一事实来源，供各种引擎访问。

本文档介绍了您在 Lakehouse 运行时目录中创建的目录。如需通过 Lakehouse 运行时目录联合目录查询由 AWS Glue Data Catalog、Databricks Unity Catalog 或 Snowflake Horizon Catalog 管理的 Iceberg 表，请参阅使用 Spark 查询联合目录。

湖仓一体架构图。

关于 Lakehouse 运行时目录

Lakehouse 运行时目录是一种无服务器、全代管式 metastore，基于 Lakehouse 运行时目录构建。它会存储 Iceberg 等开放表格式的表元数据，并公开以下端点：

• Iceberg REST 目录端点：允许 Spark 和其他引擎通过标准 Iceberg REST 协议读取和写入 Iceberg 表。
• Hive 目录端点：让 Spark 和 Hive 工作负载无需自行托管 Hive Metastore 即可管理标准 Hive 和 Spark 表，例如 Parquet、ORC、Avro 和 Delta Lake 表。

从 Managed Service for Apache Spark 映像和运行时版本 4.0 开始，Lakehouse 运行时目录是默认 Spark 目录，因此您无需任何目录配置即可创建和查询 Iceberg 表。

与 Hive Metastore 相比的优势

默认情况下，开源 Apache Spark 会将表元数据存储在与集群一起运行的 Hive Metastore 中，例如嵌入式 Apache Derby 数据库或集群主服务器上的 Hive Metastore 服务。Lakehouse 运行时目录具有以下优势：

能力	Spark Hive Metastore	Lakehouse 运行时目录
基础设施	在集群上运行。您需要自行调整其大小、修补、备份和扩缩。	无服务器且全代管式。无需操作 Metastore 基础设施。
元数据生命周期	与集群绑定。删除临时集群时，元数据会丢失，并且无服务器工作负载没有永久性元存储。	独立于集群、批处理工作负载和会话而持久存在。
跨工作负载共享	每个集群都有自己的元存储，除非您设置并连接外部元存储。	集群、批处理工作负载、会话和其他项目共享同一目录。
多引擎访问	Spark 和 Hive。	Spark、Hive、BigQuery 以及任何支持 Iceberg REST 协议的引擎。
表格式	Hive 表。Iceberg 需要额外的目录配置。	Iceberg 表通过 Iceberg REST 目录端点访问，Hive 表通过 Hive 目录端点访问。
访问权限控制	Metastore 级访问权限，通常与数据访问权限分开管理。	基于 IAM 的目录和表权限。
设置	每个集群或工作负载中的手动目录属性。	对于映像和运行时版本 4.0 及更高版本，默认处于开启状态。

如需查看 Hive 目录端点支持的 Hive Metastore 操作的详细列表，请参阅与 Hive Metastore 的功能比较。

准备工作

登录您的 Google Cloud 账号。如果您是 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.

Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

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.

Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

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

1. 创建 Cloud Storage 存储桶以存储 Iceberg 数据。
2. 在 Lakehouse 运行时目录中创建一个目录，并将其指向您的存储桶。建议使用多存储桶目录。如果您使用的是自动创建的默认入门目录，则对于映像和运行时版本 4.0 及更高版本，此步骤是可选的。

所需的角色

您需要拥有特定的 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 Editor (roles/dataproc.editor)
  • Compute Engine 默认服务账号或自定义服务账号的 Service Account User (roles/iam.serviceAccountUser)
• 使用无服务器运行时 3.0 或更高版本：项目的 Dataproc Serverless Editor (roles/dataproc.serverlessEditor)
• 在无服务器运行时 3.0 或更高版本上使用服务账号凭证：针对自定义服务账号的 Service Account User (roles/iam.serviceAccountUser)

如果您将 Managed Service for Apache Spark 无服务器运行时 3.0 或更高版本与最终用户凭证（默认）搭配使用，则还需要以下数据访问角色：

如需获得使用最终用户凭据读取和写入目录中的表所需的权限，请让您的管理员为您授予以下 IAM 角色：

• 在凭据自动售卖模式下读取和写入表数据：针对包含目录的项目的 BigLake Editor (roles/biglake.editor) 角色
• 在非凭证自动售卖模式下管理目录资源和写入表数据：
  • 包含目录的项目上的 BigLake Editor (roles/biglake.editor) 角色
  • 针对与目录关联的所有 Cloud Storage 存储桶的 Storage Object User (roles/storage.objectUser)

服务账号角色

向运行工作负载的服务账号授予以下角色。 仅当工作负载使用服务账号凭证运行时，Managed Service for Apache Spark 无服务器运行时 3.0 及更高版本才需要这些角色。

为确保 Compute Engine 默认服务账号或自定义服务账号具有运行 Managed Service for Apache Spark 作业以及在目录中读取和写入表所需的权限，请让您的管理员为 Compute Engine 默认服务账号或自定义服务账号授予以下 IAM 角色：

重要提示：您必须向 Compute Engine 默认服务账号或自定义服务账号（而不是向您的用户账号）授予这些角色。如果未能向正确的主账号授予这些角色，可能会导致权限错误。

• 在集群或无服务器运行时 2.3 或更早版本上运行 Managed Service for Apache Spark 作业：项目的 Dataproc Worker (roles/dataproc.worker)
• 在无服务器运行时 3.0 或更高版本上运行 Managed Service for Apache Spark 作业：
  • 针对项目的 Dataproc Serverless Node (roles/dataproc.serverlessNode)
  • 针对项目的 Dataproc Resource Manager Node Service Agent (roles/dataprocrm.nodeServiceAgent)
• 在凭据自动售卖模式下读取和写入表数据：针对包含目录的项目的 BigLake Editor (roles/biglake.editor) 角色
• 在非凭证自动售卖模式下管理目录资源和写入表数据：
  • 包含目录的项目上的 BigLake Editor (roles/biglake.editor) 角色
  • 针对与目录关联的所有 Cloud Storage 存储桶的 Storage Object User (roles/storage.objectUser)

使用 Lakehouse 运行时目录

从映像和运行时版本 4.0 开始，Managed Service for Apache Spark 会将 Lakehouse 运行时目录设置为集群、批量工作负载、交互式会话和运行时配置文件的默认 Spark 目录。您无需创建目录或设置任何目录属性。

在映像和运行时版本 3.0 及更早版本中，您可以通过设置以下属性来选择启用默认 Lakehouse 运行时目录：

• 批处理工作负载、交互式会话和运行时配置文件： dataproc.lakehouse.defaultCatalog=standard-lh-catalog-REGION
• 集群： dataproc:dataproc.lakehouse.defaultCatalog=standard-lh-catalog-REGION

将 REGION 替换为工作负载的区域，例如 us-central1。

Managed Service for Apache Spark 会按以下方式解析默认目录：

1. 如果您设置了 dataproc.lakehouse.defaultCatalog（对于集群，则为 dataproc:dataproc.lakehouse.defaultCatalog），Managed Service for Apache Spark 会使用您的值，而不会更改该值。

2. 否则，对于映像和运行时版本 4.0 及更高版本，Managed Service for Apache Spark 会将 dataproc.lakehouse.defaultCatalog 设置为区域级入门目录 standard-lh-catalog-REGION，其中 REGION 是工作负载的区域。如果起始目录不存在，Managed Service for Apache Spark 会在运行工作负载的项目中自动创建该目录。您无需任何其他 IAM 角色即可创建入门目录。

3. 对于映像和运行时版本 3.0 及更早版本，Managed Service for Apache Spark 不会设置默认目录。如需选择启用，请明确设置 dataproc.lakehouse.defaultCatalog=standard-lh-catalog-REGION，如本部分前面所述。

设置默认目录后，Spark 会使用目录名称 default_catalog 和以下解析属性注册该目录：

Spark 属性	值
spark.sql.defaultCatalog	default_catalog
spark.sql.catalog.default_catalog	org.apache.iceberg.spark.SparkCatalog
spark.sql.catalog.default_catalog.type	rest
spark.sql.catalog.default_catalog.warehouse	bl://projects/PROJECT_ID/catalogs/CATALOG_ID

因此，不限定的表名称（例如 my_namespace.my_table）在 Lakehouse 运行时目录中会解析为 default_catalog.my_namespace.my_table。

设置默认目录

如需在较低映像或运行时版本上将 Lakehouse 运行时目录用作默认目录，或使用入门目录以外的目录，请设置 dataproc.lakehouse.defaultCatalog=CATALOG 属性（对于集群，则为 dataproc:dataproc.lakehouse.defaultCatalog=CATALOG），其中 CATALOG 是以下项之一：

• standard-lh-catalog-REGION：工作负载项目中的区域级入门目录。
• projects/PROJECT_ID/catalogs/CATALOG_ID：现有目录，可以位于其他项目中。

例如，以下命令会提交一个使用现有目录作为默认目录的批处理工作负载：

gcloud dataproc batches submit pyspark PYSPARK_FILE \
    --region=REGION \
    --version=RUNTIME_VERSION \
    --properties=dataproc.lakehouse.defaultCatalog=projects/PROJECT_ID/catalogs/CATALOG_ID

如需在 Google Cloud 控制台中配置默认目录，请参阅运行时配置文件和集群配置说明。

停用入门目录

在映像和运行时版本 4.0 及更高版本中，您可以通过将 dataproc.lakehouse.defaultCatalog=CATALOG（对于集群，为 dataproc:dataproc.lakehouse.defaultCatalog=CATALOG）设置为另一个目录（例如 projects/PROJECT_ID/catalogs/CATALOG_ID）的完整资源名称，来停用区域级入门目录。Managed Service for Apache Spark 会使用您的目录，而不是初始目录，并且在您明确设置该属性时不会覆盖该属性。

注意： 如果您设置了 spark.sql.defaultCatalog（如配置 Managed Service for Apache Spark 并创建 Spark 会话中的示例所示），Managed Service for Apache Spark 不会将入门目录设置为默认目录。

验证默认目录

1. 检查已解析的属性：

   • 对于交互式会话，请运行以下命令：

     gcloud beta dataproc sessions describe SESSION_ID \
         --location=REGION \
         --format="yaml(runtimeConfig.properties)"

   • 对于集群，请检查主节点上的 /etc/spark/conf/spark-defaults.conf。

2. 在 Spark 应用中，确认默认目录处于活跃状态：

   print(spark.conf.get("spark.sql.defaultCatalog"))  # default_catalog
   

配置 Managed Service for Apache Spark 并创建 Spark 会话

将集群、批量工作负载或交互式会话配置为使用 Lakehouse 运行时目录 Iceberg REST 目录。这三种方法都支持 dataproc.lakehouse.catalog.CATALOG_NAME 简化配置属性（推荐），该属性可让 Managed Service for Apache Spark 读取目录定义并生成 Iceberg REST Catalog Spark 属性。只有集群形式采用 dataproc: 前缀。

在映像和运行时版本 4.0 及更高版本中，您可以跳过此配置，并使用默认 Lakehouse 运行时目录。按照以下说明注册您自己的目录，或在早期版本中配置 Lakehouse 运行时目录。

支持的映像和运行时版本：

简化配置和手动配置都使用 Iceberg 的 Google 授权 (GoogleAuthManager)，这需要以下版本之一：

• Managed Service for Apache Spark 集群：2.2 映像版本 2.2.65 及更高版本、2.3 映像版本 2.3.11 及更高版本，或 4.0 及更高版本的映像版本。
• Managed Service for Apache Spark 无服务器批处理工作负载和交互式会话：2.2 运行时 2.2.60 及更高版本、2.3 运行时 2.3.10 及更高版本，或 3.0 及更高版本的运行时。

在映像和运行时版本 4.0 及更高版本中，Lakehouse 运行时目录是默认目录。

对于无服务器批处理工作负载和交互式会话，所需的角色取决于运行时版本。运行时 3.0 及更高版本默认使用最终用户凭据。如需了解详情，请参阅必需的角色。

集群

使用简化的配置属性（推荐）或在笔记本中手动指定目录属性，创建包含 Iceberg 和 Jupyter 可选组件的 Managed Service for Apache Spark 集群。

注意 ：在映像版本 4.0 及更高版本中，集群使用默认 Lakehouse 运行时目录，没有任何目录属性。

使用属性简化配置（推荐）

使用目录属性创建集群：

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 属性即可正常运行。在数据沿袭示例中，添加了此列以进行沿袭跟踪。

注意 ：您可以向 --properties 标志传递多个以逗号分隔的目录。每个集群目录属性都需要 dataproc: 前缀。

使用 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 笔记本，然后手动配置 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 的 X-Iceberg-Access-Delegation 标头添加到 Iceberg REST 目录请求中：

.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 中的标头部分。

控制台配置

如需在Google Cloud 控制台中创建集群时配置 Lakehouse 运行时目录，请完成以下步骤：

1. 在 Google Cloud 控制台中，前往 Managed Service for Apache Spark 集群页面。

   转到集群

2. 点击创建集群，然后点击在 Compute Engine 上创建集群行中的创建。

3. 在 Metastore 配置步骤中，在选择 Metastore 和项目下，选择 Lakehouse 运行时目录。

4. 按照运行时配置文件标签页中的说明，配置项目 ID、Iceberg REST 端点目录和可选的 Hive 端点目录。

5. 完成其余集群设置，然后点击创建。

无服务器

使用简化的配置属性（推荐）或手动指定目录属性，将 PySpark 批量工作负载提交到 Managed Service for Apache Spark 无服务器。

注意 ：在运行时版本 4.0 及更高版本中，批处理工作负载使用默认 Lakehouse 运行时目录，无需任何目录属性。

使用属性简化配置（推荐）

在提交批处理工作负载时传递 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。

注意： 您可以向 --properties 标志传递多个以逗号分隔的目录。

在 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"

当目录使用凭证分发模式时，简化配置会添加此标头。如需了解详情，请参阅 Iceberg 文档的 RESTCatalog 中的标头部分。

无服务器

使用 DataprocSparkSession API，通过简化的配置属性（推荐）或手动指定目录属性，从笔记本创建 Managed Service for Apache Spark 无服务器交互式会话。

注意： 在运行时版本 4.0 及更高版本中，交互式会话使用默认 Lakehouse 运行时目录，不包含任何目录属性。

使用属性简化配置（推荐）

在安装了 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。

您还可以在运行时配置文件中配置 Lakehouse 目录。如需了解详情，请参阅运行时配置文件标签页。

手动配置

如果您不使用简化的配置属性，请在创建会话时指定目录属性：

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 中的标头部分。

无服务器

运行时配置文件（在 gcloud CLI 和 Dataproc API 中称为会话模板）存储可重复使用的无服务器设置，包括 Metastore。您通过运行时配置文件创建的交互式会话会使用其目录配置。

控制台配置

1. 在 Google Cloud 控制台中，前往运行时配置文件页面。

   前往“运行时配置文件”

2. 点击创建，或点击现有运行时配置文件的名称，然后点击修改。

3. 展开其他配置，然后点击元存储区配置旁边的修改。

4. 选择 Lakehouse 运行时目录。这是默认选项。

5. 配置目录：

   • 项目 ID：包含目录的项目。默认值为当前项目。
   • Iceberg REST 端点：默认处于选中状态。选择以下任一选项：
     • 定义一个新目录，以使用区域级入门目录 standard-lh-catalog-REGION。
     • 选择一个现有的目录 ID。
   • 可选：Hive 端点：选中此复选框，然后指定 Hive 目录的项目 ID、目录 ID 和目录名称。

6. 点击保存，然后点击创建或保存。

gcloud 配置

1. 创建一个 YAML 文件，用于在 runtimeConfig.properties 中设置默认目录：

   runtimeConfig:
   version: "RUNTIME_VERSION"
   properties:
     dataproc.lakehouse.defaultCatalog: projects/PROJECT_ID/catalogs/CATALOG_ID
   sparkConnectSession: {}

   在运行时版本 4.0 及更高版本中，您可以省略 dataproc.lakehouse.defaultCatalog 以使用区域级入门目录。

2. 导入文件以创建或更新运行时配置文件：

   gcloud beta dataproc session-templates import RUNTIME_PROFILE_ID \
     --location=REGION \
     --source=FILE_NAME.yaml

替换以下内容：

• RUNTIME_VERSION：Managed Service for Apache Spark 无服务器运行时版本，例如 4.0。
• PROJECT_ID：包含目录的项目。
• CATALOG_ID：目录的 ID。
• RUNTIME_PROFILE_ID：运行时配置文件的 ID。
• REGION：运行时配置文件的区域。
• FILE_NAME：YAML 文件的名称。

使用 Spark SQL 管理数据

配置 Spark 会话后，使用 Spark SQL 执行数据管理操作。

设置目录：dataproc.lakehouse.catalog. 属性会将目录注册到 Spark，但不会将其设为默认目录。上一部分中的示例设置了 spark.sql.defaultCatalog（集群上的 spark:spark.sql.defaultCatalog），以便您可以使用非限定表名称。如果省略该属性，请使用目录名称限定表名称，例如 CATALOG_NAME.NAMESPACE_NAME.base_table。在映像和运行时版本 4.0 及更高版本中，如果您未设置 dataproc.lakehouse.defaultCatalog，则不限定的表名称会解析为默认 Lakehouse 运行时目录 (default_catalog)。

1. 创建命名空间。在 Lakehouse 运行时目录 Iceberg REST 目录中，命名空间对应于 BigQuery 数据集。

   spark.sql("CREATE NAMESPACE IF NOT EXISTS NAMESPACE_NAME")
   spark.sql("USE NAMESPACE_NAME")
   

   将 NAMESPACE_NAME 替换为命名空间的名称，例如 spark_lakehouse。

2. 创建 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|
      +---+
   

3. 为新数据创建另一个表。

   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|
      +---+
   

4. 将新数据合并到基表中。

   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|
      +---+
   

5. 更新基表中的记录。

   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|
      +---+
   

6. 从基表中删除记录。

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

   输出类似于以下内容：

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

查询历史快照

通过查询快照 ID 检索表的先前版本。此操作也称为“时间旅行”。

1. 检索 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 替换为您创建的命名空间。

2. 使用检索到的快照 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 目录表之间的数据移动。

数据沿袭示例

1. 创建源 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
     """)
   

2. 在 Google Cloud 控制台中，前往 Knowledge Catalog 搜索页面。

   转到搜索

3. 搜索其中一个表，然后点击 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。