将 Iceberg REST 目录与 Cloud Storage 搭配使用

授予 IAM 角色

为了让 Managed Service for Apache Spark PySpark 作业和 Lakehouse 运行时目录能够使用 Cloud Storage 和 BigQuery，请向其相应的主账号授予必要的角色：

1. 在 Google Cloud 控制台中，点击激活 Cloud Shell。

   激活 Cloud Shell

2. 点击授权。

3. 向项目的 Compute Engine 默认服务账号授予 Dataproc Worker 角色，Managed Service for Apache Spark 默认使用该服务账号，详情请参阅 Managed Service for Apache Spark 服务账号。

   gcloud projects add-iam-policy-binding PROJECT_ID \
       --member="serviceAccount:$(gcloud projects describe PROJECT_ID --format='value(projectNumber)')-compute@developer.gserviceaccount.com" \
       --role="roles/dataproc.worker"

4. 向项目的 Compute Engine 默认服务账号授予 Service Usage Consumer 角色。

   gcloud projects add-iam-policy-binding PROJECT_ID \
       --member="serviceAccount:$(gcloud projects describe PROJECT_ID --format='value(projectNumber)')-compute@developer.gserviceaccount.com" \
       --role="roles/serviceusage.serviceUsageConsumer"

5. 向项目的 Compute Engine 默认服务账号授予 BigLake 编辑者角色。

   gcloud projects add-iam-policy-binding PROJECT_ID \
       --member="serviceAccount:$(gcloud projects describe PROJECT_ID --format='value(projectNumber)')-compute@developer.gserviceaccount.com" \
       --role="roles/biglake.editor"

6. 向项目的 Compute Engine 默认服务账号授予 BigQuery Data Editor 角色。

   gcloud projects add-iam-policy-binding PROJECT_ID \
       --member="serviceAccount:$(gcloud projects describe PROJECT_ID --format='value(projectNumber)')-compute@developer.gserviceaccount.com" \
       --role="roles/bigquery.dataEditor"

   替换以下内容：

   • PROJECT_ID：您的 Google Cloud 项目 ID

创建 Lakehouse 运行时目录

创建 Lakehouse 运行时目录，以管理 Iceberg 表的元数据。 您可以在 PySpark 作业中连接到此目录。

1. 在 Google Cloud 控制台中，前往 Lakehouse。

   前往 Lakehouse

2. 点击 add_box 创建目录。

   系统会打开创建目录页面。

3. 选择 Iceberg REST 目录，然后选择 Cloud Storage 存储桶。

4. 在选择 Cloud Storage 存储桶部分，点击浏览，然后点击创建新存储桶。

5. 为您的存储桶输入一个唯一的名称。

   重要提示

   请记住您的存储桶名称。它还会自动用作 Lakehouse 目录名称。此设置无法更改。如果您想将该名称存储在变量中，以便在本教程的后续步骤中使用，现在可以在此处添加该名称。

   LAKEHOUSE_CATALOG_ID

   如果您的存储桶位于多区域（例如 us、eu），请使用同一地理位置中的区域，例如 us-east1 或 europe-west4。您可以将区域存储在此变量中以供日后使用：

   REGION

   请记住您创建存储桶的区域。在本教程的后面部分，当您使用 dataproc batches submit pyspark 命令运行 PySpark 作业时，必须使用相同的区域。如果您在多区域（例如 us、eu）中创建存储桶，则应使用同一地理位置中的区域，例如 us-east1 或 europe-west4。如果您想将该名称存储在变量中，以便在本教程的后续部分中使用，现在就可以在此处添加该名称。

   REGION

6. 从存储桶列表中选择您的存储桶，然后点击选择。

7. 对于身份验证方法，请选择凭证分发模式。

8. 点击创建。

   系统会创建您的目录，并打开目录详情页面。

9. 在身份验证方法下，点击设置存储桶权限。

10. 在对话框中，点击确认。

    这会验证目录的服务账号是否具有存储桶的 Storage Object User 角色。

创建并运行 PySpark 作业

如需创建和查询 Iceberg 表，请先使用必要的 Spark SQL 语句创建 PySpark 作业。然后使用 Managed Service for Apache Spark 运行作业。

创建包含命名空间和表的 PySpark 脚本

在文本编辑器中，创建一个名为 quickstart.py 且包含以下内容的文件。

此 PySpark 脚本会初始化一个 Spark 会话，以对 Iceberg 目录执行多项操作。如果命名空间尚不存在，脚本会先创建一个。然后，它会创建一个名为 quickstart_table 且具有基本架构的 Iceberg 表。创建表后，脚本会插入三行数据。 最后，它会查询该表以检索所有插入的记录。

然后，在下一步中运行 gcloud dataproc batches submit pyspark 作业时，系统会使用这些值。

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("quickstart").getOrCreate()

# Create a namespace (dataset) if it doesn't exist
spark.sql("CREATE NAMESPACE IF NOT EXISTS `quickstart_catalog`.quickstart_namespace")

# Create the table
spark.sql("""
    CREATE OR REPLACE TABLE `quickstart_catalog`.quickstart_namespace.quickstart_table (
        id INT,
        name STRING
    )
    USING iceberg
""")

# Insert data into the table
spark.sql("""
    INSERT INTO `quickstart_catalog`.quickstart_namespace.quickstart_table
    VALUES (1, 'one'), (2, 'two'), (3, 'three')
""")

将脚本上传到您的 Cloud Storage 存储桶

创建 quickstart.py 脚本后，将其上传到 Cloud Storage 存储桶。

1. 在 Google Cloud 控制台中，前往 Cloud Storage 存储分区。

   进入“存储桶”

2. 点击相应存储桶的名称。

3. 在对象标签页中，依次点击上传 > 上传文件。

4. 在文件浏览器中，选择 quickstart.py 文件，然后点击打开。

运行 PySpark 作业

上传 quickstart.py 脚本后，以 Managed Service for Apache Spark 批量作业的形式运行该脚本。

1. 在 Cloud Shell 中，使用 quickstart.py 脚本运行以下 Managed Service for Apache Spark 批量作业。

   gcloud dataproc batches submit pyspark gs://LAKEHOUSE_CATALOG_ID/quickstart.py \
       --project=PROJECT_ID \
       --region=REGION \
       --version=2.2 \
       --properties="\
   spark.sql.defaultCatalog=quickstart_catalog,\
   spark.sql.catalog.quickstart_catalog=org.apache.iceberg.spark.SparkCatalog,\
   spark.sql.catalog.quickstart_catalog.type=rest,\
   spark.sql.catalog.quickstart_catalog.uri=https://biglake.googleapis.com/iceberg/v1/restcatalog,\
   spark.sql.catalog.quickstart_catalog.warehouse=gs://LAKEHOUSE_CATALOG_ID,\
   spark.sql.catalog.quickstart_catalog.io-impl=org.apache.iceberg.gcp.gcs.GCSFileIO,\
   spark.sql.catalog.quickstart_catalog.header.x-goog-user-project=PROJECT_ID,\
   spark.sql.catalog.quickstart_catalog.rest.auth.type=org.apache.iceberg.gcp.auth.GoogleAuthManager,\
   spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions,\
   spark.sql.catalog.quickstart_catalog.header.X-Iceberg-Access-Delegation=vended-credentials,\
   spark.sql.catalog.quickstart_catalog.gcs.oauth2.refresh-credentials-endpoint=https://oauth2.googleapis.com/token"

   替换以下内容：

   • LAKEHOUSE_CATALOG_ID：包含 PySpark 应用文件的 Cloud Storage 存储桶的名称。

     重要提示：

     此标识符也是您目录的名称。例如，如果您创建了一个用于存储目录的存储桶，并将其命名为 iceberg-bucket，则目录名称和存储桶名称均为 iceberg-bucket。稍后，当您使用 P.C.N.T 语法在 BigQuery 中查询目录时，系统会使用此名称。例如 my-project.biglake-catalog-id-name.quickstart_namespace.quickstart_table。

   • PROJECT_ID：您的 Google Cloud 项目 ID。

   • REGION：运行 Managed Service for Apache Spark 批量工作负载的区域。

   作业完成后，系统会显示类似于以下内容的输出：

   Batch [cb9d84e9489d408baca4f9e7ab4c64ff] finished.
   metadata:
   '@type': type.googleapis.com/google.cloud.dataproc.v1.BatchOperationMetadata
   batch: projects/your-project/locations/us-central1/batches/cb9d84e9489d408baca4f9e7ab4c64ff
   batchUuid: 54b0b9d2-f0a1-4fdf-ae44-eead3f8e60e9
   createTime: '2026-01-24T00:10:50.224097Z'
   description: Batch
   labels:
       goog-dataproc-batch-id: cb9d84e9489d408baca4f9e7ab4c64ff
       goog-dataproc-batch-uuid: 54b0b9d2-f0a1-4fdf-ae44-eead3f8e60e9
       goog-dataproc-drz-resource-uuid: batch-54b0b9d2-f0a1-4fdf-ae44-eead3f8e60e9
       goog-dataproc-location: us-central1
   operationType: BATCH
   name: projects/your-project/regions/us-central1/operations/32287926-5f61-3572-b54a-fbad8940d6ef
   

查询 BigQuery 中的表

1. 在 Google Cloud 控制台中，前往 BigQuery。

   转到 BigQuery

2. 在查询编辑器中，输入以下语句。该查询使用 project.catalog.namespace.table 语法。

   SELECT * FROM `PROJECT_ID.LAKEHOUSE_CATALOG_ID.quickstart_namespace.quickstart_table`;
   

   将

   • PROJECT_ID：您的 Google Cloud 项目 ID。

   • LAKEHOUSE_CATALOG_ID：要在 BigQuery 查询中使用的目录标识符。

     重要提示

     此标识符也是您的 Cloud Storage 存储桶的名称。

     例如，如果您创建了一个用于存储目录的存储桶，并将其命名为 iceberg-bucket，则目录名称和存储桶名称均为 iceberg-bucket。稍后，当您使用 P.C.N.T 语法在 BigQuery 中查询目录时，系统会使用此值。例如 my-project.biglake-catalog-id-name.quickstart_namespace.quickstart_table。

3. 点击运行。

   查询结果会显示您通过 PySpark 作业插入的数据。