跳至主要内容

PostgreSQL 表引擎

PostgreSQL 引擎允许对存储在远程 PostgreSQL 服务器上的数据进行 SELECT 和 INSERT 查询。

注意

目前，仅支持 PostgreSQL 12 及更高版本。

使用 PeerDB 复制或迁移 Postgres 数据

除了 Postgres 表引擎之外，您还可以使用 ClickHouse 的 PeerDB 从 Postgres 设置持续数据管道到 ClickHouse。PeerDB 是一个专门设计的工具，用于使用更改数据捕获 (CDC) 将数据从 Postgres 复制到 ClickHouse。

创建表​

CREATE TABLE [IF NOT EXISTS] [db.]table_name [ON CLUSTER cluster]
(
    name1 type1 [DEFAULT|MATERIALIZED|ALIAS expr1] [TTL expr1],
    name2 type2 [DEFAULT|MATERIALIZED|ALIAS expr2] [TTL expr2],
    ...
) ENGINE = PostgreSQL({host:port, database, table, user, password[, schema, [, on_conflict]] | named_collection[, option=value [,..]]})

查看 CREATE TABLE 查询的详细说明。

表结构可能与原始 PostgreSQL 表结构不同

  • 列名应与原始 PostgreSQL 表中的列名相同，但您可以使用这些列中的一部分，并且可以按任何顺序使用。
  • 列类型可能与原始 PostgreSQL 表中的列类型不同。ClickHouse 会尝试 转换 值为 ClickHouse 数据类型。
  • external_table_functions_use_nulls 设置定义如何处理可为空列。默认值：1。如果为 0，则表函数不会创建可为空列，而是插入默认值而不是空值。这同样适用于数组中的空值。

引擎参数

  • host:port — PostgreSQL 服务器地址。
  • database — 远程数据库名称。
  • table — 远程表名称。
  • user — PostgreSQL 用户。
  • password — 用户密码。
  • schema — 非默认表模式。可选。
  • on_conflict — 冲突解决策略。例如：ON CONFLICT DO NOTHING。可选。注意：添加此选项会降低插入效率。

命名集合（从 21.11 版本开始可用）建议在生产环境中使用。以下是一个示例

<named_collections>
    <postgres_creds>
        <host>localhost</host>
        <port>5432</port>
        <user>postgres</user>
        <password>****</password>
        <schema>schema1</schema>
    </postgres_creds>
</named_collections>

某些参数可以通过键值参数覆盖

SELECT * FROM postgresql(postgres_creds, table='table1');

实现细节​

PostgreSQL 侧的 SELECT 查询在只读 PostgreSQL 事务内作为 COPY (SELECT ...) TO STDOUT 运行，每个 SELECT 查询之后都会提交。

简单的 WHERE 子句，如 =、!=、>、>=、<、<= 和 IN 在 PostgreSQL 服务器上执行。

所有连接、聚合、排序、IN [ array ] 条件和 LIMIT 采样约束仅在对 PostgreSQL 的查询完成之后在 ClickHouse 中执行。

PostgreSQL 侧的 INSERT 查询在 PostgreSQL 事务内作为 COPY "table_name" (field1, field2, ... fieldN) FROM STDIN 运行，每个 INSERT 语句之后会自动提交。

PostgreSQL Array 类型会转换为 ClickHouse 数组。

注意

小心 - 在 PostgreSQL 中，像 type_name[] 这样的数组数据可能包含在同一列不同表行中不同维度的多维数组。但在 ClickHouse 中，仅允许在同一列所有表行中具有相同维度数量的多维数组。

支持多个副本，这些副本必须通过 | 列出。例如

CREATE TABLE test_replicas (id UInt32, name String) ENGINE = PostgreSQL(`postgres{2|3|4}:5432`, 'clickhouse', 'test_replicas', 'postgres', 'mysecretpassword');

支持 PostgreSQL 字典源的副本优先级。映射中数字越大，优先级越低。最高优先级为 0。

在下面的示例中，副本 example01-1 具有最高优先级

<postgresql>
    <port>5432</port>
    <user>clickhouse</user>
    <password>qwerty</password>
    <replica>
        <host>example01-1</host>
        <priority>1</priority>
    </replica>
    <replica>
        <host>example01-2</host>
        <priority>2</priority>
    </replica>
    <db>db_name</db>
    <table>table_name</table>
    <where>id=10</where>
    <invalidate_query>SQL_QUERY</invalidate_query>
</postgresql>
</source>

使用示例​

PostgreSQL 中的表​

postgres=# CREATE TABLE "public"."test" (
"int_id" SERIAL,
"int_nullable" INT NULL DEFAULT NULL,
"float" FLOAT NOT NULL,
"str" VARCHAR(100) NOT NULL DEFAULT '',
"float_nullable" FLOAT NULL DEFAULT NULL,
PRIMARY KEY (int_id));

CREATE TABLE

postgres=# INSERT INTO test (int_id, str, "float") VALUES (1,'test',2);
INSERT 0 1

postgresql> SELECT * FROM test;
  int_id | int_nullable | float | str  | float_nullable
 --------+--------------+-------+------+----------------
       1 |              |     2 | test |
 (1 row)

在 ClickHouse 中创建表，并连接到上面创建的 PostgreSQL 表​

此示例使用 PostgreSQL 表引擎 将 ClickHouse 表连接到 PostgreSQL 表，并使用 SELECT 和 INSERT 语句对 PostgreSQL 数据库进行操作

CREATE TABLE default.postgresql_table
(
    `float_nullable` Nullable(Float32),
    `str` String,
    `int_id` Int32
)
ENGINE = PostgreSQL('localhost:5432', 'public', 'test', 'postges_user', 'postgres_password');

使用 SELECT 查询将初始数据从 PostgreSQL 表插入到 ClickHouse 表中​

postgresql 表函数 将数据从 PostgreSQL 复制到 ClickHouse，这通常用于通过在 ClickHouse 而不是在 PostgreSQL 中查询或执行分析来提高数据的查询性能，也可以用于将数据从 PostgreSQL 迁移到 ClickHouse。由于我们将从 PostgreSQL 复制数据到 ClickHouse，因此我们将在 ClickHouse 中使用 MergeTree 表引擎，并将其称为 postgresql_copy

CREATE TABLE default.postgresql_copy
(
    `float_nullable` Nullable(Float32),
    `str` String,
    `int_id` Int32
)
ENGINE = MergeTree
ORDER BY (int_id);
INSERT INTO default.postgresql_copy
SELECT * FROM postgresql('localhost:5432', 'public', 'test', 'postges_user', 'postgres_password');

将增量数据从 PostgreSQL 表插入到 ClickHouse 表中​

如果在初始插入之后执行 PostgreSQL 表和 ClickHouse 表之间的持续同步，则可以在 ClickHouse 中使用 WHERE 子句，根据时间戳或唯一序列 ID 仅插入添加到 PostgreSQL 的数据。

这将需要跟踪之前添加的最大 ID 或时间戳，例如以下内容

SELECT max(`int_id`) AS maxIntID FROM default.postgresql_copy;

然后插入 PostgreSQL 表中大于 max 的值

INSERT INTO default.postgresql_copy
SELECT * FROM postgresql('localhost:5432', 'public', 'test', 'postges_user', 'postgres_password');
WHERE int_id > maxIntID;

从生成的 ClickHouse 表中选择数据​

SELECT * FROM postgresql_copy WHERE str IN ('test');
┌─float_nullable─┬─str──┬─int_id─┐
│           ᴺᵁᴸᴸ │ test │      1 │
└────────────────┴──────┴────────┘

使用非默认模式​

postgres=# CREATE SCHEMA "nice.schema";

postgres=# CREATE TABLE "nice.schema"."nice.table" (a integer);

postgres=# INSERT INTO "nice.schema"."nice.table" SELECT i FROM generate_series(0, 99) as t(i)
CREATE TABLE pg_table_schema_with_dots (a UInt32)
        ENGINE PostgreSQL('localhost:5432', 'clickhouse', 'nice.table', 'postgrsql_user', 'password', 'nice.schema');

另请参阅

相关内容​