数据库

BigQuery

BigQuery 是一种完全无服务器、经济高效的企业级数据仓库，可在云端运行并随您的数据扩展，内置 BI、机器学习和人工智能。

BigQuery Wrapper 允许您在 Postgres 数据库中读取和写入 BigQuery 数据。

准备#

在查询 BigQuery 之前，您需要启用 Wrappers 扩展并在 Postgres 中存储您的凭据。

启用 Wrappers#

确保 wrappers 扩展已安装在您的数据库上

1
create extension if not exists wrappers with schema extensions;

启用 BigQuery Wrapper#

启用 bigquery_wrapper FDW

1
create foreign data wrapper bigquery_wrapper
2
  handler big_query_fdw_handler
3
  validator big_query_fdw_validator;

存储您的凭据（可选)#

默认情况下，Postgres 将 FDW 凭据以明文形式存储在 pg_catalog.pg_foreign_server 中。任何访问此表的人都可以查看这些凭据。Wrappers 旨在与 Vault 配合使用，Vault 为存储凭据提供额外的安全级别。我们建议使用 Vault 存储您的凭据。

1
-- Save your BigQuery service account json in Vault and retrieve the created `key_id`
2
select vault.create_secret(
3
  '
4
    {
5
      "type": "service_account",
6
      "project_id": "your_gcp_project_id",
7
      "private_key_id": "your_private_key_id",
8
      "private_key": "-----BEGIN PRIVATE KEY-----\n...\n-----END PRIVATE KEY-----\n",
9
      ...
10
    }
11
  ',
12
  'bigquery',
13
  'BigQuery service account json for Wrappers'
14
);

连接到 BigQuery#

我们需要向 Postgres 提供连接到 BigQuery 的凭据以及任何其他选项。我们可以使用 create server 命令来执行此操作

1
create server bigquery_server
2
  foreign data wrapper bigquery_wrapper
3
  options (
4
    sa_key_id '<key_ID>', -- The Key ID from above.
5
    project_id 'your_gcp_project_id',
6
    dataset_id 'your_gcp_dataset_id'
7
  );

创建模式#

我们建议创建一个模式来保存所有外部表

1
create schema if not exists bigquery;

选项#

创建 BigQuery 外表时，可用的选项如下

  • table - BigQuery 中的源表或视图名称，必需
  • location - 源表位置（默认值：'US'）
  • timeout - 查询请求超时时间（毫秒），默认值为 30000
  • rowid_column - 主键列名称（数据修改必需）

您还可以将子查询用作表选项

1
table '(select * except(props), to_json_string(props) as props from `my_project.my_dataset.my_table`)'

注意：使用子查询时，必须使用完全限定的表名。

实体#

表#

BigQuery Wrapper 支持从 BigQuery 表和视图读取和写入数据。

操作#

对象选择插入更新删除截断
表✅✅✅✅❌

用法#

1
create foreign table bigquery.my_bigquery_table (
2
  id bigint,
3
  name text,
4
  ts timestamp
5
)
6
  server bigquery_server
7
  options (
8
    table 'people',
9
    location 'EU'
10
  );

说明#

  • 支持 where、order by 和 limit 子句下推
  • 使用 rowid_column 时，必须为数据修改操作指定它
  • 流缓冲区中的数据在刷新缓冲区之前（最多 90 分钟）无法更新或删除

查询下推支持#

此 FDW 支持 where、order by 和 limit 子句下推。

插入行和流缓冲区#

此外部数据包装器使用 BigQuery 的 insertAll API 方法创建一个具有关联分区时间的 streamingBuffer。 在该分区时间内，数据无法更新、删除或完全导出。 只有在时间流逝后（根据 BigQuery 的文档，最多 90 分钟），才能执行操作。

如果您尝试在 streamingBuffer 中对行执行 UPDATE 或 DELETE 语句，您将收到一个错误，提示 UPDATE 或 DELETE 语句作用于数据集名称 - 请注意，表名称会影响流缓冲区中的行，这不受支持。

支持的数据类型#

Postgres 类型BigQuery 类型
booleanBOOL
bigintINT64
double precisionFLOAT64
numericNUMERIC
textSTRING
varcharSTRING
dateDATE
timestampDATETIME
timestampTIMESTAMP
timestamptzTIMESTAMP
jsonbJSON

限制#

本节描述了在使用此 FDW 时需要注意的重要限制和注意事项

  • 大型结果集在数据传输期间可能会遇到网络延迟
  • 流缓冲区中的数据最多 90 分钟内无法修改
  • 仅支持 Postgres 和 BigQuery 之间特定的数据类型映射
  • 使用外表的物化视图在逻辑备份期间可能会失败

示例#

一些关于如何使用 BigQuery 外表的示例。

让我们先准备 BigQuery 中的源表

1
-- Run below SQLs on BigQuery to create source table
2
create table your_project_id.your_dataset_id.people (
3
  id int64,
4
  name string,
5
  ts timestamp,
6
  props jsonb
7
);
8
9
-- Add some test data
10
insert into your_project_id.your_dataset_id.people values
11
  (1, 'Luke Skywalker', current_timestamp(), parse_json('{"coordinates":[10,20],"id":1}')),
12
  (2, 'Leia Organa', current_timestamp(), null),
13
  (3, 'Han Solo', current_timestamp(), null);

基本示例#

此示例将在您的 Postgres 数据库中创建一个名为 people 的“外表”并查询其数据

1
create foreign table bigquery.people (
2
  id bigint,
3
  name text,
4
  ts timestamp,
5
  props jsonb
6
)
7
  server bigquery_server
8
  options (
9
    table 'people',
10
    location 'EU'
11
  );
12
13
select * from bigquery.people;

数据修改示例#

此示例将修改 Postgres 数据库中名为 people 的“外表”中的数据，请注意，rowid_column 选项是必需的

1
create foreign table bigquery.people (
2
  id bigint,
3
  name text,
4
  ts timestamp,
5
  props jsonb
6
)
7
  server bigquery_server
8
  options (
9
    table 'people',
10
    location 'EU',
11
    rowid_column 'id'
12
  );
13
14
-- insert new data
15
insert into bigquery.people(id, name, ts, props)
16
values (4, 'Yoda', '2023-01-01 12:34:56', '{"coordinates":[10,20],"id":1}'::jsonb);
17
18
-- update existing data
19
update bigquery.people
20
set name = 'Anakin Skywalker', props = '{"coordinates":[30,40],"id":42}'::jsonb
21
where id = 1;
22
23
-- delete data
24
delete from bigquery.people
25
where id = 2;