将 BigQuery 数据同步到 AlloyDB

预览

此功能受服务专用条款的“通用服务条款”部分中的“非正式版产品条款”的约束。 非正式版功能“按原样”提供，可能只能获得有限的支持。 如需了解详情，请参阅 发布阶段说明。

如需了解如何访问此版本，请参阅 访问 申请页面。

注意： 同步数据功能仅支持 PostgreSQL 18。

本页面介绍了如何将 BigQuery 中的表同步到 AlloyDB for PostgreSQL 实例。

通过将 BigQuery 中的分析数据同步到 AlloyDB，您可以构建运营系统，这些系统可以从对数据湖的低延迟事务访问中受益。与就地查询数据的 外部数据封装容器 (FDW) 不同，同步表会将数据移至 AlloyDB 存储空间，以实现最佳性能。

AlloyDB 提供了以下方法将 BigQuery 数据移至您的实例：

• 一次性同步 ：创建 BigQuery 表的可写独立副本。

• 定期同步（镜像） ：创建只读本地表，该表会按时间表自动刷新，例如每 6 小时或每天刷新一次。

性能和运营注意事项

使用 BigQuery 同步表时，请注意以下事项：

• 资源用量：数据移动会消耗 CPU 和内存。对于非常大的表，请考虑在非高峰时段安排同步，以避免影响主事务工作负载。
• 数据可见性：在替换操作期间，系统会预先删除并重新创建现有目标表。导入期间的查询最初会看到一个空表，然后随着批量事务提交，新导入的数据会逐步显示。

准备工作

1. 熟悉 bigquery_fdw 如何处理 BigQuery 数据类型和列映射， 因为 alloydb_sync 扩展程序使用 bigquery_fdw 连接到 BigQuery。

2. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

   Roles required to select or create a project

   • Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
   • Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.
   Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

   Go to project selector

3. Verify that billing is enabled for your Google Cloud project.

4. 启用创建和连接到 AlloyDB 所需的 Cloud API。

   启用 API

5. 如需确认您要更改的项目的名称，请在确认项目 步骤中，点击下一步 。

6. 在启用 API 步骤中，点击启用 以启用以下内容：

   • AlloyDB API
   • Compute Engine API
   • Cloud Resource Manager API
   • Service Networking API
   • BigQuery Storage API

   如果您计划使用与 AlloyDB 位于同一项目中的 VPC 网络配置与 AlloyDB 的网络连接，则需要使用 Service Networking API。 Google Cloud

   如果您计划使用位于其他项目中的 VPC 网络配置与 AlloyDB 的网络连接，则需要使用 Compute Engine API 和 Cloud Resource Manager API。 Google Cloud

7. 确保您有一个现有的 BigQuery 表，用于从中同步数据。如需了解详情，请参阅创建和使用 BigQuery 表。

所需的角色

如需向 AlloyDB 集群服务帐号授予 BigQuery 数据集访问权限，您需要以下权限：

• BigQuery Data Viewer (roles/bigquery.dataViewer) 或具有 bigquery.tables.get 和 bigquery.tables.getData 权限的任何自定义角色。针对服务帐号授予此角色后，可提供从表或视图中读取数据和元数据的权限。
• BigQuery Read Session User (roles/bigquery.readSessionUser) 或具有 bigquery.readsessions.create 和 bigquery.readsessions.getData 权限的任何自定义角色。提供创建和使用读取会话的功能。
• BigQuery Job User (roles/bigquery.jobUser) 或具有 bigquery.jobs.create 权限的任何自定义角色。提供创建和运行作业（包括查询作业）的功能。

配置扩展程序

在同步 BigQuery 中的表之前，请启用所需的扩展程序并配置与 BigQuery 的连接。如果您使用 控制台 Google Cloud ，AlloyDB 会自动执行这些步骤 。

1. 创建扩展程序。

   1. 按照将 psql 客户端连接到实例中的说明，使用 psql 客户端连接到 AlloyDB 实例。

   2. 运行以下命令：

      CREATE EXTENSION IF NOT EXISTS alloydb_sync;
      

      注意： 您必须在要同步 BigQuery 数据的每个数据库中创建这些扩展程序。

2. 如需让 AlloyDB 向 BigQuery 进行身份验证，请创建用户映射。

   CREATE EXTENSION IF NOT EXISTS bigquery_fdw;
   CREATE SERVER IF NOT EXISTS BIGQUERY_SERVER_NAME FOREIGN DATA WRAPPER bigquery_fdw;
   CREATE USER MAPPING IF NOT EXISTS FOR USER SERVER BIGQUERY_SERVER_NAME;
   

   替换以下内容：

   • USER：访问 BigQuery 表的数据库用户名或 IAM 用户。
   • BIGQUERY_SERVER_NAME：BigQuery 服务器的唯一标识符。在给定数据库中定义一次。 您可以将 BIGQUERY_SERVER_NAME 替换为您的服务器名称。

同步 BigQuery 表以进行一次性导出

您可以使用 Google Cloud 控制台或使用 psql 同步 BigQuery 表以进行一次性导出。

使用 Google Cloud 控制台

如需使用 Google Cloud 控制台将 BigQuery 表同步到 AlloyDB，请执行以下操作：

1. 在 Google Cloud 控制台中打开 BigQuery 页面。

   转到 BigQuery 页面

2. 在左侧窗格中，点击 Explorer:

   如果您没有看到左侧窗格，请点击 展开左侧窗格 以打开该窗格。

3. 在 Explorer 窗格中，展开您的项目，点击 数据集，然后 点击您的数据集。

4. 点击概览 > 表，然后选择一个表。

5. 在详细信息窗格中，依次点击 upload 导出 / 同步 > AlloyDB（导出一次或同步）。

6. 在选择目标集群下，选择以下选项之一：

   • 选择使用现有集群 ，将 BigQuery 表导出到现有 AlloyDB 集群。之后，执行以下操作：

     1. 选择主 AlloyDB 集群。

     2. 选择目标 AlloyDB 数据库。

     3. 为目标 AlloyDB 表选择架构。

     4. 为目标 AlloyDB 表指定名称。

     5. 对于同步频率 ，选择仅一次 以创建 BigQuery 表的副本。

     6. 点击设置导出 。

     同步设置完成后，您可以使用提供的 SQL 语句跟踪导入作业并查询导入的表。点击查询 ，系统会将您定向到 AlloyDB Studio 中导入的表。

   • 选择创建新集群 ，将 BigQuery 表导出到新的 AlloyDB 集群。之后，执行以下操作：

     1. 点击设置导出 。

     2. 在重定向到 AlloyDB 对话框中，选择重定向 。

     3. 选择集群类型，可以是免费试用集群 或预配的集群 。

     4. 点击继续 。

     5. 在 Sync Configuration 中，选择默认的 postgres 目标 AlloyDB 数据库、默认的 public 目标 AlloyDB 表架构，并为目标 AlloyDB 表指定名称。

        对于同步频率 ，选择仅一次 以创建 BigQuery 表的副本。

     6. 点击继续 。

     7. 配置集群。如需详细了解每个字段，请参阅 创建新集群和主实例。

     8. 点击创建集群 。

     同步设置完成后，前往 AlloyDB Studio 查询导入的表。

使用 psql 一次性同步 BigQuery 表

**注意**： alloydb_sync.import_bq_table() 函数会创建一个可写的 AlloyDB 表，该表不会自动刷新 BigQuery 中的数据。

如需创建 BigQuery 数据的可修改副本，请使用 psql运行 alloydb_sync.import_bq_table 函数。

SELECT alloydb_sync.import_bq_table(
  'PROJECT_ID.DATASET_ID.TABLE_ID',
  'ALLOYDB_DESTINATION_TABLE_NAME',
  'ON_EXISTS',
  ARRAY['PRIMARY_KEY_COLUMN']
);

替换以下内容：

• PROJECT_ID：BigQuery 数据集所在项目的 ID。
• DATASET_ID：表的 BigQuery 数据集的名称。对于具有 4 部分名称的 Iceberg 表，这是 Catalog.Namespace。
• TABLE_ID：BigQuery 表或视图的名称。
• ALLOYDB_DESTINATION_TABLE_NAME：AlloyDB 数据库中要创建并导入数据的本地表的名称。 您可以添加架构名称，例如 public.local_sales。
• ON_EXISTS：如果目标表已存在，则使用的策略。
• PRIMARY_KEY_COLUMN：要用作主键的列名称的可选列表。

示例

以下示例展示了如何将 BigQuery 数据集中的名为 transactions 的表同步到名为 public.local_sales 的新 AlloyDB 表：

SELECT alloydb_sync.import_bq_table(
    'my-gcp-project.sales_data.transactions',
    'public.local_sales',
    'replace'
);

on_exists 参数

如果目标表已存在于 AlloyDB 中，on_exists 参数会确定函数如何处理同步：

• error：默认选项。如果目标表已存在，则停止同步。
• skip：如果目标表已存在，则跳过同步。
• replace：使用 BigQuery 中的新数据替换现有本地表。

主键支持

如果您以文本数组的形式提供可选的 primary_key 参数，AlloyDB 会使用指定的列作为主键创建表。

SELECT alloydb_sync.import_bq_table(
    'my-gcp-project.sales_data.transactions',
    'public.local_sales',
    ARRAY['transaction_id']
);

同步 BigQuery 表以进行定期导出

您可以使用 Google Cloud 控制台或使用 psql 同步 BigQuery 表以进行定期导出。

使用 Google Cloud 控制台

如需使用 Google Cloud 控制台将 BigQuery 表同步到 AlloyDB，请执行以下操作：

1. 在 Google Cloud 控制台中打开 BigQuery 页面。

   转到 BigQuery 页面

2. 在左侧窗格中，点击 探索器：

   

   如果您没有看到左侧窗格，请点击 展开左侧窗格以打开该窗格。

3. 在 Explorer 窗格中，展开您的项目，点击 数据集，然后 点击您的数据集。

4. 点击概览 > 表，然后选择一个表。

5. 在详细信息窗格中，依次点击 upload 导出 / 同步 > AlloyDB（导出一次或同步）。

6. 在选择目标集群下，选择以下选项之一：

   • 选择使用现有集群 ，将 BigQuery 表导出到现有 AlloyDB 集群。之后，执行以下操作：

     1. 选择主 AlloyDB 集群。

     2. 选择目标 AlloyDB 数据库。

     3. 为目标 AlloyDB 表选择架构。

     4. 为目标 AlloyDB 表指定名称。

     5. 对于同步频率，选择一个时段以创建 BigQuery 表的定期同步，例如每小时、每六小时。

     6. 点击设置导出 。

     同步设置完成后，您可以使用提供的 SQL 语句跟踪导入作业并查询导入的表。点击查询 ，系统会将您定向到 AlloyDB Studio 中导入的表。

   • 选择创建新集群 ，将 BigQuery 表导出到新的 AlloyDB 集群。之后，执行以下操作：

     1. 点击设置导出 。

     2. 在重定向到 AlloyDB 对话框中，选择重定向 。

     3. 选择集群类型，可以是免费试用集群 或预配的集群 。

     4. 点击继续 。

     5. 在 Sync Configuration 中，选择默认的 postgres 目标 AlloyDB 数据库、默认的 public 目标 AlloyDB 表架构，并为目标 AlloyDB 表指定名称。

        对于同步频率 ，选择仅一次 以创建 BigQuery 表的副本。

     6. 点击继续 。

     7. 配置集群。如需详细了解每个字段，请参阅 创建新集群和主实例。

     8. 点击创建集群 。

     同步设置完成后，前往 AlloyDB Studio 查询导入的表。

创建定期同步

如需维护与 BigQuery 数据保持同步的只读表，请使用 psql 运行 alloydb_sync.create_bq_sync_table 函数。

SELECT alloydb_sync.create_bq_sync_table(
    'PROJECT_ID.DATASET_ID.TABLE_ID',
    'ALLOYDB_DESTINATION_TABLE_NAME',
    'REFRESH_INTERVAL',
    'ON_EXISTS',
    ARRAY['PRIMARY_KEY_COLUMN']
);

替换以下内容：

• PROJECT_ID.DATASET_ID.TABLE_ID：BigQuery 表或视图的完全限定名称，包括项目 ID、数据集 ID 和表 ID，以英文句点分隔。对于具有 4 部分名称的 Iceberg 表，DATASET_ID 表示为 Catalog.Namespace。 例如，my-gcp-project.sales_data.transactions。
• ALLOYDB_DESTINATION_TABLE_NAME：AlloyDB 数据库中要创建并同步数据的本地表的名称。
• REFRESH_INTERVAL：AlloyDB 定期从 BigQuery 刷新数据的时间间隔，例如 12 hours。
• ON_EXISTS：如果目标表已存在，则使用的策略。
• PRIMARY_KEY_COLUMN：要用作主键的列名称的可选列表。

示例

以下示例展示了如何创建每 12 小时刷新一次的客户个人资料镜像：

SELECT alloydb_sync.create_bq_sync_table(
    'my-gcp-project.crm_data.profiles',
    'public.customer_mirror',
    '12 hours',
    'replace'
);

监控和管理作业

启动同步后，您可以监控其进度并管理作业。

检查作业状态

大型同步可能需要一些时间。您可以通过查询 job_status 视图来监控进度，包括已处理的记录和预计完成时间：

**注意**： alloydb_sync.import_bq_table() 函数会返回一个作业 UUID，您可以将其与 import_id 在 alloydb_sync.job_status 视图中的内容进行匹配。

SELECT
    import_id,
    status,
    records_processed,
    total_records,
    error
FROM alloydb_sync.job_status;

例如，如需取消作业，请运行以下命令：

SELECT alloydb_sync.cancel_import_job('85bb5dfa-dfb9-4017-9153-738f55abe4b1');

停止和删除同步作业

如需停止镜像 BigQuery 表并删除本地表，请使用 alloydb_sync.delete_bq_sync_table 函数：

SELECT alloydb_sync.delete_bq_sync_table('public.customer_mirror');

重要提示： 您必须使用此函数而不是标准 PostgreSQL DROP TABLE 命令来删除复制的表。如果不使用此函数手动删除同步表，则不会移除后台调度器钩子，这可能会导致调度器不断遇到错误并耗尽后台工作器槽。

数据类型映射

当您使用 alloydb_sync 扩展程序将 BigQuery 中的数据同步或导入到 AlloyDB 时，AlloyDB 会将 BigQuery 数据类型映射到目标表中相应的 PostgreSQL 数据类型。

验证源 BigQuery 表列是否使用以下受支持的数据类型。

下表列出了 BigQuery 和 AlloyDB 之间的数据类型映射。

BigQuery 表 数据类型	推荐的 PostgreSQL 外部表 数据类型
BOOLEAN	BOOLEAN
INTEGER (INT64)	BIGINT
FLOAT (FLOAT64)	DOUBLE PRECISION
STRING	VARCHAR
NUMERIC	NUMERIC(38, 9)
NUMERIC(P[, S])	NUMERIC(P, S)
BIGNUMERIC	NUMERIC(77, 38)
BIGNUMERIC(P[, S])	NUMERIC(P, S)
DATE	DATE
TIMESTAMP	TIMESTAMPTZ
TIME	TIME
JSON	JSONB
BYTES	BYTEA
GEOGRAPHY	GEOGRAPHY(POINT), ... 如需了解详情，请参阅 PostGIS_Geography。
DATETIME	TIMESTAMP
ARRAY	VECTOR(N) N 是向量的维度。您必须在会话中设置 bigquery_fdw.enable_vector_downcasting 标志。 由于 AlloyDB 中的 VECTOR 类型使用 float4 类型，因此在此转换中可能会出现精度损失。 如需了解详情，请参阅 pgvector 扩展程序。

限制

从 BigQuery 同步表时，适用以下限制：

• 此功能仅支持 PostgreSQL 18。
• 如果您 DROP alloydb_sync 扩展程序，则必须先重启实例，然后才能再次创建该扩展程序。
• 同步在事务中运行。如果导入作业中断或失败，系统会回滚导入的数据。
• 如果两个用户同时启动具有相同目标表的同步作业，则这些表可能会相互覆盖。
• 如果新注册的同步表的初始后台导入期间发生任何中断，则该表将保持不完整状态，直到下一个计划的刷新时间间隔为止。如需解决此问题，您可以使用 alloydb_sync.delete_bq_sync_table() 函数删除同步表并重新创建该表。
• 同步不支持复杂的 BigQuery 类型，例如 ARRAY、BYTES、VECTOR 和 GEOGRAPHY。如需查看完整列表，请参阅 受支持的 BigQuery 数据类型和列映射。
• 请勿手动删除复制的表。使用 alloydb_sync.delete_bq_sync_table() API 函数安全地删除表和刷新。
• 如需删除使用 alloydb_sync 扩展程序的数据库，您必须使用 DROP DATABASE ... WITH (FORCE)。
• 如果在导入运行时 Postgres 数据库崩溃，元数据可能会停留在 RUNNING 状态，从而阻止将来的导入。您必须 手动运行 UPDATE alloydb_sync.import_job_status SET status = 'FAILED' WHERE status = 'RUNNING'; 才能解除阻止。

价格

当您将 BigQuery 中的数据同步到 AlloyDB 时，系统会按照 BigQuery 流式读取（Storage Read API）价格向您收费。

导出数据后，如果您将数据存储在 AlloyDB 中，则需要为此付费。如需了解详情， 请参阅 AlloyDB for PostgreSQL 价格。

后续步骤

• 详细了解将数据导出到 AlloyDB（反向 ETL）。
• 详细了解如何使用 FDW 就地查询 BigQuery 数据。