将 Lakehouse 运行时目录与 BigQuery 中的表搭配使用

借助 BigQuery，您可以创建和管理标准（内置）表、BigQuery 托管式 Apache Iceberg 表和外部 Apache Iceberg 表。然后，您可以使用 Lakehouse 运行时目录配置 Apache Spark 会话，以便直接从 Managed Service for Apache Spark 工作负载访问和查询这些表。

准备工作

  1. 为您的 Google Cloud 项目启用结算功能。了解如何检查项目是否已启用结算功能。

  2. 启用 BigQuery API 和 Dataproc API。

    启用 API

所需的角色

如需获得将 Managed Service for Apache Spark 与 Lakehouse 运行时目录搭配使用作为元数据存储区所需的权限，请让您的管理员为您授予以下 IAM 角色：

  • 在 Apache Spark 中创建 Lakehouse 运行时目录表：
    • 针对项目中 Managed Service for Apache Spark 服务账号的 Dataproc Worker (roles/dataproc.worker)
    • 针对项目中 Managed Service for Apache Spark 服务账号的 BigQuery Data Editor (roles/bigquery.dataEditor) 角色
    • 针对项目中 Managed Service for Apache Spark 服务账号的 Storage Object User (roles/storage.objectUser) 角色
  • 在 BigQuery 中查询 Lakehouse 运行时目录表：

如需详细了解如何授予角色，请参阅管理对项目、文件夹和组织的访问权限。

您也可以通过自定义角色或其他预定义角色来获取所需的权限。

连接到表

  1. 在 Google Cloud 控制台中创建数据集。

    CREATE SCHEMA `PROJECT_ID`.DATASET_NAME;

    替换以下内容：

    • PROJECT_ID：要创建数据集的 Google Cloud 项目的 ID。
    • DATASET_NAME：数据集的名称。

  2. 创建 Cloud 资源连接。

  3. 创建标准 BigQuery 表。

    CREATE TABLE `PROJECT_ID`.DATASET_NAME.TABLE_NAME (name STRING,id INT64);

    替换以下内容：

    • TABLE_NAME：表的名称。

  4. 将数据插入到标准 BigQuery 表中。

    INSERT INTO `PROJECT_ID`.DATASET_NAME.TABLE_NAME VALUES ('test_name1', 123),('test_name2', 456),('test_name3', 789);

  5. 创建 BigQuery 托管式 Apache Iceberg 表。

    例如，如需创建表，请运行以下 CREATE 语句。

    CREATE TABLE `PROJECT_ID`.DATASET_NAME.ICEBERG_TABLE_NAME(
name STRING,id INT64
)
WITH CONNECTION `CONNECTION_NAME`
OPTIONS (
file_format = 'PARQUET',
table_format = 'ICEBERG',
storage_uri = 'STORAGE_URI');

    替换以下内容：

    • ICEBERG_TABLE_NAME：代管式 Apache Iceberg 表的名称。例如 iceberg_managed_table。
    • CONNECTION_NAME：您的连接的名称。您在上一步中创建了此连接。例如 myproject.us.myconnection。
    • STORAGE_URI：完全限定的 Cloud Storage URI。例如 gs://mybucket/table。

  6. 将数据插入 BigQuery 托管式 Apache Iceberg 表中。

    INSERT INTO `PROJECT_ID`.DATASET_NAME.ICEBERG_TABLE_NAME VALUES ('test_name1', 123),('test_name2', 456),('test_name3', 789);

  7. 创建外部 Apache Iceberg 表。

    例如，如需创建外部 Apache Iceberg 表，请运行以下 CREATE 语句。

    CREATE OR REPLACE EXTERNAL TABLE  `PROJECT_ID`.DATASET_NAME.READONLY_ICEBERG_TABLE_NAME
WITH CONNECTION `CONNECTION_NAME`
OPTIONS (
  format = 'ICEBERG',
  uris =
    ['BUCKET_PATH'],
  require_partition_filter = FALSE);

    替换以下内容：

    • READONLY_ICEBERG_TABLE_NAME：只读表的名称。
    • BUCKET_PATH：包含外部表数据的 Cloud Storage 存储桶的路径，格式为 ['gs://bucket_name/[folder_name/]file_name']。

  8. 从 Apache Spark 查询标准表、BigQuery 托管式 Apache Iceberg 表和外部 Apache Iceberg 表。

    from pyspark.sql import SparkSession

# Create a spark session
spark = SparkSession.builder \
.appName("Lakehouse runtime catalog 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.conf.set("viewsEnabled","true")

# Use the Lakehouse runtime catalog
spark.sql("USE `CATALOG_NAME`;")
spark.sql("USE NAMESPACE DATASET_NAME;")

# Configure spark for temp results
spark.sql("CREATE namespace if not exists MATERIALIZATION_NAMESPACE");
spark.conf.set("materializationDataset","MATERIALIZATION_NAMESPACE")

# List the tables in the dataset
df = spark.sql("SHOW TABLES;")
df.show();

# Query the tables
sql = """SELECT * FROM DATASET_NAME.TABLE_NAME"""
df = spark.read.format("bigquery").load(sql)
df.show()

sql = """SELECT * FROM DATASET_NAME.ICEBERG_TABLE_NAME"""
df = spark.read.format("bigquery").load(sql)
df.show()

sql = """SELECT * FROM DATASET_NAME.READONLY_ICEBERG_TABLE_NAME"""
df = spark.read.format("bigquery").load(sql)
df.show()

    替换以下内容：

    • WAREHOUSE_DIRECTORY：与 BigQuery 托管式 Apache Iceberg 表和外部 Apache Iceberg 表相关联的 Cloud Storage 文件夹的 URI。
    • CATALOG_NAME：您要使用的目录的名称。
    • MATERIALIZATION_NAMESPACE：用于存储临时结果的命名空间。

  9. 使用 Managed Service for Apache Spark 运行 Apache Spark 脚本。

    gcloud dataproc batches submit pyspark SCRIPT_PATH \
  --version=2.2 \
  --project=PROJECT_ID \
  --region=REGION \
  --deps-bucket=YOUR_BUCKET \

    替换以下内容：

    • SCRIPT_PATH：批量作业使用的脚本的路径。
    • PROJECT_ID：要运行批量作业的 Google Cloud 项目的 ID。
    • REGION：工作负载运行所在的区域。
    • YOUR_BUCKET：用于上传工作负载依赖项的 Cloud Storage 存储桶的位置。存储桶的 gs:// URI 前缀不是必需的。您可以指定存储桶路径或存储桶名称，例如 mybucketname1。

后续步骤