使用 Spark 和 Lakehouse 运行时目录查询联合目录

预览

此功能受服务专用条款的“通用服务条款”部分中的“非正式版产品条款”的约束。 非正式版功能“按原样”提供，可能只能获得有限的支持。 如需了解详情，请参阅发布阶段说明。

在 Lakehouse 运行时目录中设置跨云数据访问后，您就可以从 Managed Service for Apache Spark 查询远程 Iceberg 表，而无需迁移文件或管理单独的云端存储空间凭据。

Lakehouse 运行时目录通过凭据分发处理元数据同步和安全数据访问。当 Spark 查询联合表时，Lakehouse 运行时目录会转换目录元数据，并发出短期、有范围的凭据，以便 Spark 从 Amazon S3 或 Cloud Storage 读取数据，并将这些表视为Google Cloud 环境中的本地表。

支持的联合目录类型

Managed Service for Apache Spark 支持通过 Lakehouse 运行时目录 Iceberg REST Catalog 端点查询以下联合目录类型：

联合目录类型	支持的存储后端	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 配置参数。

准备工作

1. 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

2. Verify that you have the permissions required to complete this guide.

3. Verify that billing is enabled for your Google Cloud project.

4. 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

5. 确保远程目录中有 Iceberg 表。
6. 在 Lakehouse 运行时目录中为远程目录提供商设置跨云连接和联合目录：
   • AWS Glue Data Catalog
   • Databricks Unity Catalog
   • Snowflake Horizon Catalog

所需的 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 Editor (roles/dataproc.editor)
  • Compute Engine 默认服务账号或自定义服务账号的 Service Account User (roles/iam.serviceAccountUser)
• 使用无服务器运行时 3.0 或更高版本：项目的 Dataproc Serverless Editor (roles/dataproc.serverlessEditor)
• 在无服务器运行时 3.0 或更高版本上使用最终用户凭据读取联合表：针对包含联合目录的项目的 BigLake Viewer (roles/biglake.viewer) 角色
• 在无服务器运行时 3.0 或更高版本上使用服务账号凭证：针对自定义服务账号的 Service Account User (roles/iam.serviceAccountUser)

服务账号角色

向运行工作负载的服务账号授予以下角色。 仅当工作负载使用服务账号凭证运行时，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 Viewer (roles/biglake.viewer) 角色

为确保 Managed Service for Apache Spark Service Agent (service-PROJECT_NUMBER@dataproc-accounts.) 拥有在您创建集群、提交批处理工作负载或创建交互式会话时解析 dataproc.lakehouse.catalog. 简化配置属性所需的权限，请让您的管理员为 Managed Service for Apache Spark Service Agent (service-PROJECT_NUMBER@dataproc-accounts.) 授予包含联合目录的项目上的 BigLake Viewer (roles/biglake.viewer) IAM 角色。

重要提示：您必须向 Managed Service for Apache Spark Service Agent (service-PROJECT_NUMBER@dataproc-accounts.)（而非您的用户账号）授予此角色。如果未能向正确的主账号授予此角色，可能会导致权限错误。

为 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 运行时目录中读取联合目录定义，并生成所需的 Iceberg REST 目录 Spark 属性，包括存储 io-impl 和凭据分发标头（该属性的集群形式会添加 dataproc: 前缀）。

简化配置和手动配置都使用 Google 授权来访问 Iceberg (GoogleAuthManager)，这需要以下集群映像或无服务器运行时版本之一：

• 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。

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

查询 Snowflake Horizon Catalog

如果您要查询 Snowflake Horizon Catalog，请注意以下事项：

• Snowflake 存储：简化配置支持存储在 Amazon S3 中的 Snowflake 表。对于存储在 Cloud Storage 中的 Snowflake 表，请使用手动配置，并将 io-impl 设置为 org.apache.iceberg.gcp.gcs.GCSFileIO。

• Snowflake STRING 列：如果您在查询 Snowflake STRING 列时遇到错误，而该表在 Snowflake 中的存储序列化政策不是 COMPATIBLE，则可以在 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。

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

查询 Snowflake Horizon Catalog

如果您要查询 Snowflake Horizon Catalog，请注意以下事项：

• Snowflake 存储：简化配置支持存储在 Amazon S3 中的 Snowflake 表。对于存储在 Cloud Storage 中的 Snowflake 表，请使用手动配置，并将 io-impl 设置为 org.apache.iceberg.gcp.gcs.GCSFileIO。

• Snowflake STRING 列：如果您在查询 Snowflake STRING 列时遇到错误，而该表在 Snowflake 中的存储序列化政策不是 COMPATIBLE，则可以在 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 表，请省略此属性。

无服务器

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

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

在安装了 google-cloud-spark-connect 软件包的笔记本（例如 Google Cloud Data Agent Kit 笔记本或 Vertex AI Workbench 笔记本）中，在创建会话时传递不带 dataproc: 前缀的联合目录属性：

from google.cloud.managed_spark_connect import ManagedSparkSession
from google.cloud.dataproc_v1 import Session

catalog_name = "CATALOG_NAME"

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

spark = ManagedSparkSession.builder.sessionId("SESSION_ID") \
.sessionConfig(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 无服务器运行时版本。google-cloud-spark-connect 需要运行时 3.0 或更高版本。所需的 IAM 角色取决于运行时版本。
• SESSION_ID：Spark 会话的自定义 ID，例如 my-spark-session。
• PROJECT_ID：您的Google Cloud 项目 ID。
• FEDERATED_CATALOG_NAME：您在准备工作中创建的联合目录的 ID。

您还可以在运行时配置文件（会话模板）中配置联合目录，方法是在Google Cloud 控制台中的创建运行时配置文件页面上，依次选择其他配置 > Metastore 配置下的 Lakehouse 运行时目录，或者使用 gcloud CLI 或 Dataproc API 在会话模板中设置这些属性。

查询 Snowflake Horizon Catalog

如果您要查询 Snowflake Horizon Catalog，请注意以下事项：

• Snowflake 存储：简化配置支持存储在 Amazon S3 中的 Snowflake 表。对于存储在 Cloud Storage 中的 Snowflake 表，请使用手动配置，并将 io-impl 设置为 org.apache.iceberg.gcp.gcs.GCSFileIO。

• Snowflake STRING 列：如果您在查询 Snowflake STRING 列时遇到错误，而 Snowflake 中表的存储序列化政策不是 COMPATIBLE，则可以在 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.managed_spark_connect import ManagedSparkSession
from google.cloud.dataproc_v1 import Session

catalog_name = "CATALOG_NAME"

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

spark = ManagedSparkSession.builder.sessionId("SESSION_ID") \
.sessionConfig(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 查询联合目录中的表。

注意： dataproc.lakehouse.catalog. 属性会将目录注册到 Spark，但不会将其设置为默认目录。上述示例设置了无服务器 spark.sql.defaultCatalog 或 spark:spark.sql.defaultCatalog 集群属性，以便您使用非限定表名称。如果您省略这些属性，则必须使用目录名称限定表名称，如以下示例所示。

1. 使用 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：联合目录中的命名空间（数据库或架构）。
   • TABLE_NAME：联合目录中的表名称。

2. 使用 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.managed_spark_connect import ManagedSparkSession
from google.cloud.dataproc_v1 import Session

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

spark = ManagedSparkSession.builder.sessionId("SESSION_ID") \
  .sessionConfig(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 Editor (roles/biglake.editor) 角色
• 在非凭证自动售卖模式下写入表数据：
  • 针对包含本地目录的项目的 BigLake Editor (roles/biglake.editor) 角色
  • 针对与本地目录关联的所有 Cloud Storage 存储桶的 Storage Object User (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。
• 详细了解如何在 Lakehouse 运行时目录中查询远程数据。
• 阅读跨云数据访问简介。
• 了解凭证分发。