使用清单查询打开的表格式

本文档介绍了如何使用清单文件查询以开放表格式（例如 Apache Hudi 和 Delta Lake）存储的数据。

某些开放表格式（例如 Hudi 和 Delta Lake）会将其当前状态导出为一个或多个清单文件。清单文件包含用于创建表的数据文件列表。借助 BigQuery 中的清单支持，您可以查询和加载以开放表格式存储的数据。

准备工作

• Enable the BigQuery Connection, BigQuery Reservation, and BigLake APIs.

  Enable the APIs

• 如需创建 BigLake 表，您可以使用以下任一方法运行 Spark 命令：

  • 创建 Dataproc 集群。如需查询 Hudi 表，请将 --optional-components 字段设置为 HUDI。如需查询 Delta 表，请将 --optional-components 设置为 Presto。

  • 在 BigQuery 中使用 Spark 的存储过程。请按以下步骤操作：

    1. 创建 Spark 连接。
    2. 为该连接设置访问权限控制。

• 如需将清单文件存储在 Cloud Storage 中，请创建 Cloud Storage 存储桶。您需要连接到 Cloud Storage 存储桶才能访问清单文件。请按以下步骤操作：

  1. 创建 Cloud 资源连接。
  2. 设置该连接的访问权限。

所需的角色

如需根据 Hudi 和 Delta Lake 数据查询 BigLake 表，请确保您具有以下角色：

• BigQuery Connection User (roles/bigquery.connectionUser)
• BigQuery Data Viewer (roles/bigquery.dataViewer)
• BigQuery User (roles/bigquery.user)

您还可以查询 Hudi 外部表。不过，我们建议您将外部表升级到 BigLake。如需查询 Hudi 外部表，请确保您具有以下角色：

• BigQuery Data Viewer (roles/bigquery.dataViewer)
• BigQuery User (roles/bigquery.user)
• Storage Object Viewer (roles/storage.objectViewer)

根据您的权限，您可以自行授予这些角色给自己，或者让管理员授予给您。如需详细了解如何授予角色，请参阅查看可针对资源授予的角色。

如需查看查询 BigLake 表所需的确切权限，请展开所需权限部分：

所需权限

• bigquery.connections.use
• bigquery.jobs.create
• bigquery.readsessions.create（仅当您使用 BigQuery Storage Read API 读取数据时才需要）
• bigquery.tables.get
• bigquery.tables.getData

您也可以使用自定义角色或其他预定义角色来获取这些权限。

查询 Hudi 工作负载

重要提示：如需执行以下部分中所述的操作，您必须使用 Hudi-BigQuery 连接器 版本 0.14.0 或更高版本，或是 Dataproc 2.1 中的 Hudi 组件（该组件具有向后移植的相应连接器版本）。 在先前版本中，连接器在清单文件中创建了视图，而这些视图并不适合查询性能。如果您使用的是先前版本的连接器，则必须删除 BigQuery 中表示 HBase 表的现有视图，以避免架构不匹配错误。

如需查询 Hudi 数据，请按以下步骤操作：

1. 根据 Hudi 数据创建外部表。
2. 将外部表升级到 BigLake。

创建 Hudi 外部表

使用 Hudi 和 BigQuery 的同步工具同步表时，请启用 use-bq-manifest-file 标志以转换为清单文件方法。此标志还会以 BigQuery 支持的格式导出清单文件，并通过它创建使用 --table 参数中指定的名称的外部表。

如需创建 Hudi 外部表，请按照以下步骤操作：

1. 如需创建 Hudi 外部表，请向现有 Dataproc 集群提交作业。构建 Hudi-BigQuery 连接器时，请启用 use-bq-manifest-file 标志以转换为清单文件方法。此标志会以 BigQuery 支持的格式导出清单文件，并通过它创建使用 --table 参数中指定的名称的外部表。

   spark-submit \
      --master yarn \
      --packages com.google.cloud:google-cloud-bigquery:2.10.4 \
      --class org.apache.hudi.gcp.bigquery.BigQuerySyncTool  \
      JAR \
      --project-id PROJECT_ID \
      --dataset-name DATASET \
      --dataset-location LOCATION \
      --table TABLE \
      --source-uri URI  \
      --source-uri-prefix URI_PREFIX \
      --base-path BASE_PATH  \
      --partitioned-by PARTITION_BY \
      --use-bq-manifest-file
   

   替换以下内容：

   • JAR：如果您使用的是 Hudi-BigQuery 连接器，请指定 hudi-gcp-bundle-0.14.0.jar。如果您使用的是 Dataproc 2.1 中的 Hudi 组件，请指定 /usr/lib/hudi/tools/bq-sync-tool/hudi-gcp-bundle-0.12.3.1.jar

   • PROJECT_ID：您要在其中创建 Hudi BigLake 表的项目 ID

   • DATASET：您要在其中创建 Hudi BigLake 表的数据集

   • LOCATION：您要在其中创建 Hudi BigLake 表的位置

   • TABLE：您要创建的表的名称

     如果您要从在清单文件上创建视图的早期版本的 Hud-BigQuery 连接器（0.13.0 及更低版本）进行转换，请确保您使用相同的表名称，因为它可让您保留现有的下游流水线代码。

   • URI：您创建的用于存储 Hudi 清单文件的 Cloud Storage URI

     此 URI 指向第一级分区：请务必添加分区键。例如 gs://mybucket/hudi/mydataset/EventDate=*。

   • URI_PREFIX：Cloud Storage URI 路径的前缀，通常是 Hudi 表的路径

   • BASE_PATH：Hudi 表的基本路径

     例如 gs://mybucket/hudi/mydataset/。

   • PARTITION_BY：分区值

     例如 EventDate。

   如需详细了解该连接器的配置，请参阅 Hudi-BigQuery 连接器。

2. 如需设置适当的精细控制，或通过启用元数据缓存来提高性能，请参阅升级 BigLake 表。

查询 Delta 工作负载

Delta 表现在获得了原生支持。我们建议为 Delta 工作负载创建 Delta BigLake 表。Delta Lake BigLake 表支持更高级的 Delta Lake 表，包括具有列重新映射和删除向量的表。此外，Delta BigLake 表会直接读取最新的快照，因此更新会立即生效。

如需查询 Delta 工作负载，请按以下步骤操作：

1. 生成清单文件。
2. 根据清单文件创建 BigLake 表。
3. 设置适当的精细控制，或通过启用元数据缓存来提高性能。如需了解如何执行此操作，请参阅升级 BigLake 表。

生成清单文件

BigQuery 支持采用 SymLinkTextInputFormat 格式的清单文件，该格式是用换行符分隔的 URI 列表。如需详细了解如何生成清单文件，请参阅设置 Presto 到 Delta Lake 集成并查询增量表。

如需生成清单文件，请向现有 Dataproc 集群提交作业：

SQL

使用 Spark，对位于位置 path-to-delta-table 的 Delta 表运行以下命令：

GENERATE symlink_format_manifest FOR TABLE delta.`<path-to-delta-table>`

Scala

使用 Spark，对位于位置 path-to-delta-table 的 Delta 表运行以下命令：

val deltaTable = DeltaTable.forPath(<path-to-delta-table>)
deltaTable.generate("symlink_format_manifest")

Java

使用 Spark，对位于位置 path-to-delta-table 的 Delta 表运行以下命令：

DeltaTable deltaTable = DeltaTable.forPath(<path-to-delta-table>);
deltaTable.generate("symlink_format_manifest");

Python

使用 Spark，对位于位置 path-to-delta-table 的 Delta 表运行以下命令：

deltaTable = DeltaTable.forPath(<path-to-delta-table>)
deltaTable.generate("symlink_format_manifest")

创建增量 BigLake 表

如需创建增量 BigLake 表，请使用将 file_set_spec_type 字段设置为 NEW_LINE_DELIMITED_MANIFEST 的 CREATE EXTERNAL TABLE 语句：

1. 转到 BigQuery 页面。

   转到 BigQuery

2. 在查询编辑器中，运行 CREATE EXTERNAL TABLE 语句：

   CREATE EXTERNAL TABLE PROJECT_ID.DATASET_NAME.TABLE_NAME
   WITH PARTITION COLUMNS(
   `PARTITION_COLUMN PARTITION_COLUMN_TYPE`,)
   WITH CONNECTION `PROJECT_IDREGION.CONNECTION_NAME`
   OPTIONS (
      format = "DATA_FORMAT",
      uris = ["URI"],
      file_set_spec_type = 'NEW_LINE_DELIMITED_MANIFEST',
      hive_partition_uri_prefix = "PATH_TO_DELTA_TABLE"
      max_staleness = STALENESS_INTERVAL,
      metadata_cache_mode = 'CACHE_MODE');
   

   请替换以下内容：

   • DATASET_NAME：您创建的数据集的名称
   • TABLE_NAME：您要为此表指定的名称
   • REGION：连接所在的位置（例如 us-east1）
   • CONNECTION_NAME：您创建的连接的名称
   • DATA_FORMAT：任何受支持的格式（例如 PARQUET）
   • URI：清单文件的路径（例如 gs://mybucket/path）
   • PATH_TO_DELTA_TABLE：分区键编码开始之前所有源 URI 的通用前缀
   • STALENESS_INTERVAL：指定对 BigLake 表执行的操作是否使用了缓存的元数据，以及操作使用的缓存的元数据的新鲜度。如需详细了解元数据缓存注意事项，请参阅启用元数据缓存以提高性能。

     如需停用元数据缓存，请指定 0。这是默认设置。

     如需启用元数据缓存，请指定 30 分钟到 7 天之间的间隔时间字面量值。例如，指定 INTERVAL 4 HOUR 表示 4 小时过时间隔时间。使用此值时，如果缓存的元数据在过去 4 小时内刷新，则对表执行的操作会使用缓存的元数据。如果缓存的元数据早于该值，则操作会改为从 Delta Lake 中检索元数据。

   • CACHE_MODE：指定元数据缓存是自动刷新还是手动刷新。如需详细了解元数据缓存注意事项，请参阅启用元数据缓存以提高性能。

     如果设置为 AUTOMATIC，元数据缓存会按系统定义的间隔时间刷新，通常在 30 到 60 分钟之间。

     如果要根据您确定的时间表刷新元数据缓存，请设置为 MANUAL。在这种情况下，您可以调用 BQ.REFRESH_EXTERNAL_METADATA_CACHE 系统过程来刷新缓存。

     如果 STALENESS_INTERVAL 设置为大于 0 的值，您必须设置 CACHE_MODE。

   示例：

   CREATE EXTERNAL TABLE mydataset.mytable
   WITH CONNECTION `us-east1.myconnection`
   OPTIONS (
       format="PARQUET",
       uris=["gs://mybucket/path/partitionpath=*"],
       file_set_spec_type = 'NEW_LINE_DELIMITED_MANIFEST'
       hive_partition_uri_prefix = "gs://mybucket/path/"
       max_staleness = INTERVAL 1 DAY,
       metadata_cache_mode = 'AUTOMATIC'
   );

升级 BigLake 表

您还可以通过利用元数据缓存和具体化视图来提高工作负载的性能。如果您想使用元数据缓存，则可以同时对此指定设置。如需获取表的详细信息，例如源格式和源 URI，请参阅获取表信息。

如需将外部表更新为 BigLake 表或更新现有 BigLake，请选择以下选项之一：

SQL

使用 CREATE OR REPLACE EXTERNAL TABLE DDL 语句更新表：

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

   转到 BigQuery

2. 在查询编辑器中，输入以下语句：

   CREATE OR REPLACE EXTERNAL TABLE
     `PROJECT_ID.DATASET.EXTERNAL_TABLE_NAME`
     WITH CONNECTION `REGION.CONNECTION_ID`
     OPTIONS(
       format ="TABLE_FORMAT",
       uris = ['BUCKET_PATH'],
       max_staleness = STALENESS_INTERVAL,
       metadata_cache_mode = 'CACHE_MODE'
       );

   请替换以下内容：

   • PROJECT_ID：包含表的项目的名称
   • DATASET：包含表的数据集的名称
   • EXTERNAL_TABLE_NAME：表的名称
   • REGION：包含连接的区域
   • CONNECTION_ID：要使用的连接的名称
   • TABLE_FORMAT：表使用的格式

     更新表时，您无法对此进行更改。

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

     您可以通过在路径中指定一个星号 (*) 通配符从存储桶中选择多个文件。例如 ['gs://mybucket/file_name*']。如需了解详情，请参阅 Cloud Storage URI 的通配符支持。

     您可以通过提供多个路径来为 uris 选项指定多个存储桶。

     以下示例展示了有效的 uris 值：

     • ['gs://bucket/path1/myfile.csv']
     • ['gs://bucket/path1/*.csv']
     • ['gs://bucket/path1/*', 'gs://bucket/path2/file00*']

     如果指定以多个文件为目标的 uris 值，则所有这些文件都必须共享一个兼容的架构。

     如需详细了解如何在 BigQuery 中使用 Cloud Storage URI，请参阅 Cloud Storage 资源路径。

   • STALENESS_INTERVAL：指定对表执行的操作是否使用缓存的元数据，以及操作使用的缓存元数据的新鲜度

     如需详细了解元数据缓存注意事项，请参阅启用元数据缓存以提高性能。

     如需停用元数据缓存，请指定 0。这是默认设置。

     如需启用元数据缓存，请指定 30 分钟到 7 天之间的间隔时间字面量值。例如，指定 INTERVAL 4 HOUR 表示 4 小时过时间隔时间。使用此值时，如果缓存的元数据在过去 4 小时内刷新，则对表执行的操作会使用缓存的元数据。如果缓存的元数据早于该值，则操作会从 Cloud Storage 检索元数据。

   • CACHE_MODE：指定元数据缓存是自动刷新还是手动刷新

     如需详细了解元数据缓存注意事项，请参阅启用元数据缓存以提高性能。

     如果设置为 AUTOMATIC，元数据缓存会按系统定义的间隔时间刷新，通常在 30 到 60 分钟之间。

     如果要根据您确定的时间表刷新元数据缓存，请设置为 MANUAL。在这种情况下，您可以调用 BQ.REFRESH_EXTERNAL_METADATA_CACHE 系统过程来刷新缓存。

     如果 STALENESS_INTERVAL 设置为大于 0 的值，您必须设置 CACHE_MODE。

3. 点击 运行。

如需详细了解如何运行查询，请参阅运行交互式查询。

bq

使用 bq mkdef 和 bq update 命令更新表：

1. 生成外部表定义，用于描述要更改的表方面：

   bq mkdef --connection_id=PROJECT_ID.REGION.CONNECTION_ID \
   --source_format=TABLE_FORMAT \
   --metadata_cache_mode=CACHE_MODE \
   "BUCKET_PATH" > /tmp/DEFINITION_FILE

   请替换以下内容：

   • PROJECT_ID：包含连接的项目的名称
   • REGION：包含连接的区域
   • CONNECTION_ID：要使用的连接的名称
   • TABLE_FORMAT：表使用的格式更新表时，您无法对此进行更改。

   • CACHE_MODE：指定元数据缓存是自动刷新还是手动刷新。如需详细了解元数据缓存注意事项，请参阅启用元数据缓存以提高性能。

     如果设置为 AUTOMATIC，元数据缓存会按系统定义的间隔时间刷新，通常在 30 到 60 分钟之间。

     如果要根据您确定的时间表刷新元数据缓存，请设置为 MANUAL。在这种情况下，您可以调用 BQ.REFRESH_EXTERNAL_METADATA_CACHE 系统过程来刷新缓存。

     如果 STALENESS_INTERVAL 设置为大于 0 的值，您必须设置 CACHE_MODE。

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

     您可以通过在路径中指定一个星号 (*) 通配符来限制从存储桶中选择的文件。例如 gs://mybucket/file_name*。如需了解详情，请参阅 Cloud Storage URI 的通配符支持。

     您可以通过提供多个路径来为 uris 选项指定多个存储桶。

     以下示例展示了有效的 uris 值：

     • gs://bucket/path1/myfile.csv
     • gs://bucket/path1/*.csv
     • gs://bucket/path1/*,gs://bucket/path2/file00*

     如果指定以多个文件为目标的 uris 值，则所有这些文件都必须共享一个兼容的架构。

     如需详细了解如何在 BigQuery 中使用 Cloud Storage URI，请参阅 Cloud Storage 资源路径。

   • DEFINITION_FILE：您要创建的表定义文件的名称。

2. 使用新的外部表定义更新表：

   bq update --max_staleness=STALENESS_INTERVAL \
   --external_table_definition=/tmp/DEFINITION_FILE \
   PROJECT_ID:DATASET.EXTERNAL_TABLE_NAME

   替换以下内容：

   • STALENESS_INTERVAL：指定对表执行的操作是否使用缓存的元数据，以及操作使用的缓存元数据的新鲜度。如需详细了解元数据缓存注意事项，请参阅启用元数据缓存以提高性能。

     如需停用元数据缓存，请指定 0。这是默认设置。

     如需启用元数据缓存，请使用 INTERVAL 数据类型文档中所述的 Y-M D H:M:S 格式指定 30 分钟到 7 天之间的间隔时间值。例如，指定 0-0 0 4:0:0 表示 4 小时过时间隔时间。使用此值时，如果缓存的元数据在过去 4 小时内刷新，则对表执行的操作会使用缓存的元数据。如果缓存的元数据早于该值，则操作会从 Cloud Storage 检索元数据。

   • DEFINITION_FILE：您创建或更新的表定义文件的名称。

   • PROJECT_ID：包含表的项目的名称

   • DATASET：包含表的数据集的名称

   • EXTERNAL_TABLE_NAME：表格的名称

查询 BigLake 表和外部表

创建 BigLake 表后，您可以使用 GoogleSQL 语法查询表，就如查询标准 BigQuery 表一样。例如 SELECT field1, field2 FROM mydataset.my_cloud_storage_table;。

限制

• BigQuery 仅支持查询 Delta Lake 读取器 v1 表。

• Hudi 和 BigQuery 集成仅适用于 Hive 样式的分区 copy-on-write 表。

后续步骤

• 了解如何在 BigQuery 中使用 SQL。
• 了解 BigLake 表。
• 了解 BigQuery 配额。