将 BigLake metastore 与 BigQuery 中的表搭配使用

本文档介绍如何将 BigLake metastore 与 BigQuery 表和 Spark 搭配使用。

借助 BigLake metastore，您可以创建和使用标准（内置）表、BigQuery 中适用于 Apache Iceberg 的 BigLake 表以及来自 BigQuery 的 Apache Iceberg 外部表。

准备工作

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

  2. 启用 BigQuery API 和 Dataproc API。

    启用 API

  3. 可选：了解 BigQuery metastore 的工作原理以及为什么您应该使用它。

所需的角色

如需获得将 BigLake metastore 作为元数据存储区并与 Spark 和 Dataproc 搭配使用所需的权限，请让您的管理员为您授予以下 IAM 角色：

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

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

连接到表

  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 BigLake 表。

    例如，如需创建表，请运行以下 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：BigQuery 中的 Apache Iceberg BigLake 表的名称。例如 iceberg_managed_table。
    • CONNECTION_NAME：您的连接的名称。您在上一步中创建了此连接。例如 myproject.us.myconnection。
    • STORAGE_URI：完全限定的 Cloud Storage URI。例如 gs://mybucket/table。

  6. 将数据插入 BigQuery 中的 Apache Iceberg BigLake 表中。

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

  7. 创建 Apache Iceberg 外部表。

    例如，如需创建 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. 从 PySpark 查询标准表、BigQuery 中适用于 Apache Iceberg 的 BigLake 表以及 Apache Iceberg 外部表。

    from pyspark.sql import SparkSession

# Create a spark session
spark = SparkSession.builder \
.appName("BigLake Metastore 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 blms_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 中的 BigLake Iceberg 表和 Iceberg 外部表相关联的 Cloud Storage 文件夹的 URI。
    • CATALOG_NAME：您要使用的目录的名称。
    • MATERIALIZATION_NAMESPACE：用于存储临时结果的命名空间。

  9. 使用 Serverless Spark 运行 PySpark 脚本。

    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。

后续步骤