将 BigQuery Metastore 与 Dataproc 搭配使用

本文档介绍了如何将 BigQuery Metastore 与 Compute Engine 上的 Dataproc 搭配使用。借助此连接，您可以获得一个能在 Apache Spark 等开源软件引擎上运行的共享 Metastore。

准备工作

为 Google Cloud 项目启用结算功能。了解如何检查项目是否已启用结算功能。
启用 BigQuery API 和 Dataproc API。

启用 API
可选：了解 BigQuery Metastore 的工作原理以及为什么您应该使用它。

所需的角色

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

创建 Dataproc 集群：项目中 Compute Engine 默认服务账号的 Dataproc Worker (roles/dataproc.worker) 角色
在 Spark 中创建 BigQuery Metastore 表：
- 项目中 Dataproc 虚拟机服务账号的 Dataproc Worker (roles/dataproc.worker) 角色
- 项目中 Dataproc 虚拟机服务账号的 BigQuery Data Editor (roles/bigquery.dataEditor) 角色
- 项目中 Dataproc 虚拟机服务账号的 Storage Object Admin (roles/storage.objectAdmin) 角色
在 BigQuery 中查询 BigQuery Metastore 表：
- 项目的 BigQuery Data Viewer (roles/bigquery.dataViewer) 角色
- 项目的 BigQuery User (roles/bigquery.user) 角色
- 项目的 Storage Object Viewer (roles/storage.objectViewer) 角色

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

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

常规工作流程

如需将 Compute Engine 上的 Dataproc 与 BigQuery Metastore 搭配使用，请按照以下常规步骤操作：

创建 Dataproc 集群或配置现有集群。
连接到您偏好的开源软件引擎，例如 Spark。
使用 JAR 文件在集群上安装 Apache Iceberg 目录插件。
根据您使用的开源软件引擎，按需创建和管理 BigQuery Metastore 资源。
在 BigQuery 中，访问和使用您的 BigQuery Metastore 资源。

将 BigQuery Metastore 连接到 Spark

以下说明介绍了如何使用交互式 Spark SQL 将 Dataproc 连接到 BigQuery Metastore。

下载 Iceberg 目录插件

如需将 BigQuery Metastore 与 Dataproc 和 Spark 相关联，您必须使用 BigQuery Metastore Iceberg 目录插件 JAR 文件。

此文件默认包含在 Dataproc 映像版本 2.2 中。如果您的 Dataproc 集群无法直接访问互联网，您必须下载该插件并将其上传到您的 Dataproc 集群可以访问的 Cloud Storage 存储桶。

下载 BigQuery Metastore Apache Iceberg 目录插件。

配置 Dataproc 集群

在连接到 BigQuery Metastore 之前，您必须设置 Dataproc 集群。

为此，您可以创建新的集群或使用现有集群。之后，您可以使用此集群运行交互式 Spark SQL 并管理 BigQuery Metastore 资源。

在创建集群的区域内，子网必须启用专用 Google 访问通道 (PGA)。默认情况下，使用版本 2.2（默认值）或更高版本的映像创建的 Dataproc 集群虚拟机仅使用内部 IP 地址。如需允许集群虚拟机与 Google API 通信，请在创建集群所在区域的 default（或用户指定的网络名称，如适用）网络子网上启用专用 Google 访问通道。
如果您想运行本指南中的 Zeppelin 网页界面示例，则必须使用或创建已启用 Zeppelin 可选组件的 Dataproc 集群。

新集群

如需创建新的 Dataproc 集群，请运行以下 gcloud dataproc clusters create 命令。此配置包含使用 BigQuery Metastore 所需的设置。

gcloud dataproc clusters create CLUSTER_NAME \
  --project=PROJECT_ID \
  --region=LOCATION \
  --optional-components=ZEPPELIN \
  --enable-component-gateway \
  --single-node

替换以下内容：

CLUSTER_NAME：Dataproc 集群的名称。
PROJECT_ID：您要创建集群的 Google Cloud 项目的 ID。
LOCATION：您在其中创建集群的 Google Cloud 区域。

现有集群

如需配置现有集群，请将以下 Iceberg Spark 运行时添加到您的集群。

org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.5.2

您可以使用以下其中一个选项来添加运行时：

初始化脚本。将运行时依赖项添加到在创建时运行的自定义初始化脚本。

将运行时依赖项添加到脚本后，请按照说明创建集群、重新创建和更新集群。
手动安装。手动添加 Iceberg 目录插件 JAR 文件，并将 Spark 属性配置为在集群中包含运行时。

提交 Spark 作业

如需提交 Spark 作业，请使用以下方法之一：

gcloud CLI

gcloud dataproc jobs submit spark-sql \
--project=PROJECT_ID \
--cluster=CLUSTER_NAME \
--region==REGION \
--jars=https://storage-download.googleapis.com/maven-central/maven2/org/apache/iceberg/iceberg-spark-runtime-3.5_2.12/1.5.2/iceberg-spark-runtime-3.5_2.12-1.5.2.jar,gs://spark-lib/bigquery/iceberg-bigquery-catalog-1.5.2-1.0.1-beta.jar \
--properties=spark.sql.catalog.CATALOG_NAME=org.apache.iceberg.spark.SparkCatalog, \
spark.sql.catalog.CATALOG_NAME.catalog-impl=org.apache.iceberg.gcp.bigquery.BigQueryMetastoreCatalog, \
spark.sql.catalog.CATALOG_NAME.gcp_project=PROJECT_ID, \
spark.sql.catalog.CATALOG_NAME.gcp_location=LOCATION, \
spark.sql.catalog.CATALOG_NAME.warehouse=WAREHOUSE_DIRECTORY \
--execute="SPARK_SQL_COMMAND"

替换以下内容：

PROJECT_ID：包含 Dataproc 集群的 Google Cloud 项目的 ID。
CLUSTER_NAME：您用于运行 Spark SQL 作业的 Dataproc 集群的名称。
REGION：您的集群所在的 Compute Engine 区域。
LOCATION：BigQuery 资源的位置。
CATALOG_NAME：您在 SQL 作业中使用的 Spark 目录的名称。
WAREHOUSE_DIRECTORY：包含数据仓库的 Cloud Storage 文件夹。此值以 gs:// 开头。
SPARK_SQL_COMMAND：您要运行的 Spark SQL 查询。此查询包含用于创建资源的命令。例如，创建命名空间和表。

互动式 Spark

连接到 Spark 并安装目录插件

如需为 BigQuery Metastore 安装目录插件，请使用 SSH 连接到您的 Dataproc 集群。

在 Google Cloud 控制台中，转到虚拟机实例页面。

如需连接到 Dataproc 虚拟机实例，请点击虚拟机实例列表中的 SSH。输出类似于以下内容：

Connected, host fingerprint: ssh-rsa ...
Linux cluster-1-m 3.16.0-0.bpo.4-amd64 ...
...
example-cluster@cluster-1-m:~$

在终端中，运行以下 BigQuery Metastore 初始化命令：

spark-sql \
--jars https://storage-download.googleapis.com/maven-central/maven2/org/apache/iceberg/iceberg-spark-runtime-3.5_2.12/1.5.2/iceberg-spark-runtime-3.5_2.12-1.5.2.jar,gs://spark-lib/bigquery/iceberg-bigquery-catalog-1.5.2-1.0.1-beta.jar \
--conf spark.sql.catalog.CATALOG_NAME=org.apache.iceberg.spark.SparkCatalog \
--conf spark.sql.catalog.CATALOG_NAME.catalog-impl=org.apache.iceberg.gcp.bigquery.BigQueryMetastoreCatalog \
--conf spark.sql.catalog.CATALOG_NAME.gcp_project=PROJECT_ID \
--conf spark.sql.catalog.CATALOG_NAME.gcp_location=LOCATION \
--conf spark.sql.catalog.CATALOG_NAME.warehouse=WAREHOUSE_DIRECTORY

替换以下内容：

CATALOG_NAME：您在 SQL 作业中使用的 Spark 目录的名称。
PROJECT_ID：Spark 目录关联的 BigQuery Metastore 目录的 Google Cloud 项目 ID。
LOCATION：BigQuery Metastore 的 Google Cloud 位置。
WAREHOUSE_DIRECTORY：包含数据仓库的 Cloud Storage 文件夹。此值以 gs:// 开头。

成功连接到集群后，Spark 终端会显示 spark-sql 提示。

spark-sql (default)>

管理 BigQuery Metastore 资源

您现在已连接到 BigQuery Metastore。您可以查看现有资源，也可以根据存储在 BigQuery Metastore 中的元数据创建新资源。

例如，尝试在交互式 Spark SQL 会话中运行以下命令，以创建 Iceberg 命名空间和表。

使用自定义 Iceberg 目录：
```
USE `CATALOG_NAME`;
```

创建命名空间：

CREATE NAMESPACE IF NOT EXISTS NAMESPACE_NAME;

使用创建的命名空间：
```
USE NAMESPACE_NAME;
```

创建 Iceberg 表：

CREATE TABLE TABLE_NAME (id int, data string) USING ICEBERG;

插入表格行：

INSERT INTO TABLE_NAME VALUES (1, "first row");

添加表格列：

ALTER TABLE TABLE_NAME ADD COLUMNS (newDoubleCol double);

查看表元数据：
```
DESCRIBE EXTENDED TABLE_NAME;
```
列出命名空间中的表：
```
SHOW TABLES;
```

Zeppelin 笔记本

在 Google Cloud 控制台中，前往 Dataproc 集群页面。

转到 Dataproc 集群
点击您要使用的集群的名称。

系统会打开集群详情页面。
在导航菜单中，点击网页界面。
在组件网关下，点击 Zeppelin。系统会打开 Zeppelin 笔记本页面。
在导航菜单中，点击笔记本，然后点击+创建新笔记本。
在对话框中，输入笔记本名称。将 Spark 保留为默认解释器。
点击创建。系统会创建一个新笔记本。
在笔记本中，点击设置菜单，然后点击解释器。
在搜索解释器字段中，搜索 Spark。
点击修改。

在 Spark.jars 字段中，输入 Spark JAR 的 URI。

https://storage-download.googleapis.com/maven-central/maven2/org/apache/iceberg/iceberg-spark-runtime-3.5_2.12/1.5.2/iceberg-spark-runtime-3.5_2.12-1.5.2.jar,gs://spark-lib/bigquery/iceberg-bigquery-catalog-1.5.2-1.0.1-beta.jar

点击保存。
点击确定。

将以下 PySpark 代码复制到 Zeppelin 笔记本中。

%pyspark
from pyspark.sql import SparkSession
spark = SparkSession.builder \
.appName("BigQuery 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.sql("select version()").show()
spark.sql("USE `CATALOG_NAME`;")
spark.sql("CREATE NAMESPACE IF NOT EXISTS NAMESPACE_NAME;")
spark.sql("USE NAMESPACE_NAME;")
spark.sql("CREATE TABLE TABLE_NAME (id int, data string) USING ICEBERG;")
spark.sql("DESCRIBE TABLE_NAME;").show()

替换以下内容：

CATALOG_NAME：要为 SQL 作业使用的 Spark 目录的名称。
PROJECT_ID：包含 Dataproc 集群的 Google Cloud 项目的 ID。
WAREHOUSE_DIRECTORY：包含数据仓库的 Cloud Storage 文件夹。此值以 gs:// 开头。
NAMESPACE_NAME：引用 Spark 表的命名空间名称。
WAREHOUSE_DIRECTORY：存储数据仓库的 Cloud Storage 文件夹的 URI。
TABLE_NAME：Spark 表的表名。

点击运行图标或按 Shift-Enter 以运行代码。作业完成后，状态消息会显示“Spark Job Finished”，输出内容会显示表格内容：

将 BigQuery Metastore 与 Dataproc 搭配使用

准备工作

所需的角色

常规工作流程

将 BigQuery Metastore 连接到 Spark

下载 Iceberg 目录插件

配置 Dataproc 集群

新集群

现有集群

提交 Spark 作业

gcloud CLI

互动式 Spark

连接到 Spark 并安装目录插件

管理 BigQuery Metastore 资源

Zeppelin 笔记本

后续步骤