PostgreSQL 迁移同步到 ClickHouse

NineData 数据复制支持将自建 PostgreSQL 的数据复制到 ClickHouse，支持结构和全量数据复制。

功能介绍

NineData 数据复制支持数据源之间的结构、全量数据、增量数据的高性能复制，对于部分数据源，还提供双向复制功能，实现快捷构建异地多活业务架构。

结构：支持同构及异构数据源之间的对象结构复制，很大程度上降低了两个数据源之间的数据复制门槛。
全量数据：通过智能数据分片实现行级并发批量复制能力，有效保障复制性能。自主研发的新型断点续传技术，保证无主键表的数据准确性。
增量数据：支持全对象类型的 DML|DDL 增量数据复制，结合行级并发、热点合并等技术，提供强劲复制性能。
双向数据实时复制：直接多个节点之间的数据双向复制，保证所有节点的数据均保持最新状态。

通过以上功能，可以轻松高效地实现全量|增量数据复制、全量|增量数据迁移、全量|增量数据同步、数据集成、不停机无缝迁移等场景，为企业提供灵活和可靠的数据复制解决方案。

前提条件

已将源数据源和目标数据源添加至 NineData。如何添加，请参见添加数据源。
源和目标数据源的版本见下表。
源数据源目标数据源
PostgreSQL 9 及以上 ClickHouse 20.8 及以上
用于登录 ClickHouse 的账号的 access_management 参数值需为 1。
配置方式：
通过编辑器打开位于 /etc/clickhouse-server 目录下的 User.xml 配置文件，找到目标用户，加入参数：<access_management>1</access_management>。
已拥有源数据源和目标数据源的如下权限。
复制类型
源数据源目标数据源
全量复制 CONNECT、SELECT TABLE的相关权限（CREATE、CREATE TABLE、ALTER、ALTER COLUMN、DROP、SELECT、INSERT)

源数据源	目标数据源
PostgreSQL 9 及以上	ClickHouse 20.8 及以上

复制类型	源数据源	目标数据源
全量复制	CONNECT、SELECT	TABLE的相关权限（CREATE、CREATE TABLE、ALTER、ALTER COLUMN、DROP、SELECT、INSERT)

注意事项

源 PostgreSQL 数据源中，如果有无符号整形类型字段，则 ClickHouse 中会被映射为 Uint 类型。

通常 PostgreSQL 到 ClickHouse 的复制场景需要包含结构复制，系统会自动迁移 PostgreSQL 的表结构到 ClickHouse，并在 ClickHouse 涉及复制的表中插入两个系统列。如果复制类型不包含结构复制，则您在需要手动确保下列事项：

ClickHouse 数据源中涉及复制的数据对象，其库表结构必须和 PostgreSQL 数据源一致。

ClickHouse 数据源涉及复制的表中都需要手动插入如下两个系统列用于记录 DML 操作以及 Binlog 的时间。

列名称	数据类型	列默认值	说明
_jz_data_sign	Int8	DEFAULT 1	记录 DML 操作的类型，保证 ClickHouse 端的数据和 PostgreSQL 端一致。 INSERT 操作：该字段记录为 1。 DELETE 操作：该字段记录为 -1。 UPDATE 操作：拆分成 INSERT、DELETE 两条记录。
_jz_data_time	String	DEFAULT now()	记录 Binlog 的时间。

示例：

CREATE TABLE [IF NOT EXISTS] [db.]table_name [ON CLUSTER cluster]
(
    ...
    _jz_data_sign Int8 DEFAULT 1 COMMENT 'replication data update sign: add = 1, remove = -1',
    _jz_data_time String DEFAULT now() COMMENT 'replication data update time'
) ENGINE = engine

使用限制

执行数据同步前需评估源数据源和目标数据源的性能，同时建议业务低峰期执行数据同步。否则全量数据初始化时将占用源数据源和目标数据源一定的读写资源，导致数据库负载上升。
建议确保同步对象中的每张表都有主键或唯一约束、列名具有唯一性，否则可能会重复同步相同数据。

操作步骤

登录 NineData 控制台。
在左侧导航栏单击数据复制 > 数据复制。
在数据复制页面，单击右上角的创建复制。

在数据源与目标页签，按照下表进行配置，并单击下一步。

参数	说明
任务名称	输入数据同步任务的名称，为了方便后续查找和管理，请尽量使用有意义的名称。最多支持 64 个字符。
源数据源	同步对象所在的数据源。
目标数据源	接收同步对象的数据源。
目标对象命名规则	选择对象名称从源端迁移到目标端后的大小写转换规则。全部转小写：无论源端的命名规则如何，目标端的命名规则全部为小写。保持与源一致：沿用源端的命名规则。全部转大写：无论源端的命名规则如何，目标端的命名规则全部为大写。
复制类型	选择复制类型。结构复制：只同步源数据源的库表结构，不同步数据。全量复制：同步源数据源的所有对象和数据，即全量数据复制。右侧的开关为周期性全量复制的开关，更多信息，请参见周期性全量复制。
复制规格	复制任务的规格，规格越大复制的速率越高，将鼠标悬浮于图标上即可查看每个规格对应的速率和配置信息。如果先配置数据复制任务再购买资源，可以在此选择需要的规格；如果先购买资源再配置任务，系统会选中购买资源时指定的规格，且无法在任务配置中变更。
目标库同名对象处理策略（选中结构复制时需要选择）	预检查报错并停止任务：预检查阶段检测到同名表时，停止任务。跳过并继续任务：预检查阶段检测到同名表时，发送提示并继续任务。结构复制时，忽略该同名表。如果您同时进行了数据复制，则数据会在同名表中追加，而不会覆盖原有数据。删除对象并重建：预检查阶段检测到同名表时，发送提示并继续任务。结构复制时，删除目标库同名表，并基于源库重新复制表结构。如果您同时进行了数据复制，则数据会在表结构复制完成后写入。
目标表存量数据处理策略（选中全量复制时需要选择）	预检查报错并停止任务：预检查阶段检测到目标表中存在数据时，停止任务。忽略目标存量数据，追加写入：预检查阶段检测到目标表中存在数据时，忽略该部分数据，追加写入其他数据。清空目标存量数据，重新写入：预检查阶段检测到目标表中存在数据时，删除该部分数据，重新写入。

在选择复制对象页签，配置下列参数，然后单击下一步。

参数	说明
复制对象	选择需要复制的内容，您可以选择全部实例复制源库所有内容，也可以选择自定义对象，在源对象列表中选中需要复制的内容，单击>添加到右侧目标对象列表。

如果您需要创建多条相同复制对象的复制链路，可以创建一个配置文件，在新建任务的时候导入即可。单击右上角的导入配置，再单击下载模板，将配置文件模版下载到本地，编辑完成后单击上传文件上传该配置文件即可实现批量导入。字段和示例请参见导入复制对象配置模板。配置文件说明：

参数	说明	示例
`source_table_name`	需同步的源表名。填写源数据源中实际存在的表名。	`example_tbl_name`
`target_table_name`	接收同步对象的目标表名。	`example_tbl_name`
`source_database_name`	需同步的对象所在的源库名。	`test_db`
`target_database_name`	接收同步对象的目标库名。	`test_db_bak`
`column_list`	需要同步的字段列表。	`["col1", "clo2", "col3"]`
`extra_configuration`	额外配置的 JSON 字符串。`column_rules` 用于配置字段映射与取值规则，其中 `column_name` 为源列名，`target_column_name` 为目标列名，`target_column_type` 为目标列类型，`column_value` 为字段值；`filter_condition` 用于配置行级过滤条件。	`{"extra_config": {"column_rules": [...]}}`

提示

以下 JSON 表示下载的 Excel 模板中的两行示例数据。实际上传时，请保留 Excel 模板的列名，并将 extra_configuration 单元格填写为 JSON 字符串；以下代码块不是可直接上传的 JSON 文件。

[
  {
    "source_table_name": "example_tbl_name",
    "target_table_name": "example_tbl_name",
    "source_database_name": "test_db",
    "target_database_name": "test_db_bak",
    "column_list": ["col1", "clo2", "col3"],
    "extra_configuration": {
      "extra_config": {
        "column_rules": [
          {
            "column_name": "example_column_name",
            "target_column_name": "example_target_column_name",
            "target_column_type": "int",
            "column_value": "current_timestamp()"
          },
          {
            "column_name": "example_column_name",
            "target_column_name": "example_target_column_name",
            "target_column_type": "int"
          }
        ],
        "filter_condition": "example_condition_expression"
      }
    }
  },
  {
    "source_table_name": "example_order_tbl",
    "target_table_name": "example_order_tbl",
    "source_database_name": "test_db",
    "target_database_name": "test_db_bak",
    "column_list": ["col1", "clo2", "col3"],
    "extra_configuration": {
      "extra_config": {
        "column_rules": [
          {
            "column_name": "example_column_name",
            "target_column_name": "example_target_column_name",
            "target_column_type": "int",
            "column_value": "current_timestamp()"
          },
          {
            "column_name": "example_column_name",
            "target_column_name": "example_target_column_name",
            "target_column_type": "int"
          }
        ],
        "filter_condition": "example_condition_expression"
      }
    }
  }
]

在配置映射页签，根据所选的复制类型选择不同操作，然后单击保存并预检查。如果在配置映射阶段，源和目标数据源中有更新，可以单击页面右上角的刷新元数据按钮，重新获取源和目标数据源的信息。
- 包含结构复制：配置目标表同步到目标数据源之后的表名。
- 不包含结构复制：系统默认选择目标数据源中的同名数据库，如果不存在，则需要手动选择目标库。目标库中的表名、列名需要和同步对象一致。如果不一致，您也可以手动进行表名和列名的映射。
您还可以进行如下操作：
- 单击页面右侧的映射与过滤，自定义列名同步到目标数据源之后的名称。
- 在映射与过滤页面，在数据过滤条件下方的文本框中输入比较表达式作为过滤条件，仅符合过滤条件的数据会同步到目标数据源。例如，将过滤条件设置为 emp_no>=10005，则 emp_no 列中小于 10005 的数据均不会同步到目标数据源。
- 单击"目标表名"右侧的图标，可搜索表名，并替换成目标名称。
- 在右上角的搜索表文本框中输入表名以快速定位到目标表。
- 单击右上角的批量配置，可以批量定义通用规则（如表名、字段名大小写转换、前后缀添加、替换等），实现对大量表和字段的统一映射配置。

在预检查页签等待系统完成预检查。预检查通过后，在复制任务列表中打开已保存任务的详情页并启动任务。
- 您可以勾选开启数据一致性对比。系统会根据您选择的复制类型，在对应的启动时机自动开启基于源数据源的数据一致性对比，用于检查源端和目标端的数据是否一致。启动时机如下：
  - 结构复制：结构复制完成后启动。
  - 结构复制+全量复制：全量复制完成后启动。
  - 全量复制：全量复制完成后启动。
- 如果预检查未通过，需要单击目标检查项右侧操作列的详情，排查失败的原因，手动修复后单击重新检查重新执行预检查，直到通过。
- 检查结果为警告的检查项，可视具体情况修复或忽略。
复制任务启动后，您可以进行以下操作：
- 单击查看详情查看同步任务各个阶段的执行情况。
  - 单击返回列表可以返回数据复制任务列表页面。

查看同步结果

登录 NineData 控制台。
在左侧导航栏单击数据复制 > 数据复制。

在数据复制页面单击目标同步任务的任务 ID，页面说明如下。

result_no_incre

序号	功能	说明
1	配置告警	配置告警后，系统会在任务失败时通过您选择的方式通知您。更多信息，请参见运维监控简介。
2	更多	暂停：暂停任务，仅状态为运行中的任务可选。类似创建：创建一个和当前任务配置相同的新复制任务。终止：结束未完成或监听中（即增量同步中）的任务，终止任务后无法重启任务，请谨慎操作。如果复制对象中包含触发器或事件，系统会显示相应对象的终止行为选项：进行触发器迁移后终止或进行事件迁移后终止：先将相应对象迁移至目标数据源，再终止任务。不迁移触发器，立即终止或不迁移事件，立即终止：不迁移相应对象，立即终止任务。删除：删除任务，任务删除后无法恢复，请谨慎操作。
3	结构复制（包含结构复制的场景下显示）	展示结构复制的进度和详细信息。单击页面右侧的日志：查看结构复制的执行日志。单击页面右侧的：查看最新的信息。单击列表中目标对象右侧操作列的查看 DDL：可以查看 SQL 回放。
4	全量复制（包含全量复制的场景下显示）	展示全量复制的进度和详细信息。单击页面右侧的监控：查看全量复制过程中的各监控指标。全量复制过程中，还可以单击监控指标页面右侧的限流设置，限制每秒写入到目标数据源的速率。单位为行/秒。单击页面右侧的日志：查看全量复制的执行日志。单击页面右侧的：查看最新的信息。
5	数据对比	展示源数据源和目标数据源之间对比的结果。如果您未开启数据对比，请单击页面中的开启数据对比。单击页面右侧的重新对比：对当前源和目标两端数据重新发起对比。单击页面右侧的停止：对比任务开始后，可单击该按钮立即停止对比任务。单击页面右侧的日志：查看一致性对比的执行日志。单击页面右侧的监控（仅数据对比显示）：查看对比 RPS（每秒对比的记录数）的走势图。单击详情可以查看更早之前的记录。在对比列表右侧操作列单击（数据页签下只在不一致情况下显示）：查看源端和目标端的对比详情。在对比列表右侧操作列单击（不一致情况下显示)：生成用于修正不一致的 SQL。该 SQL 可能修改目标端数据。执行前，请人工审查语句、目标对象和影响范围，并确认目标端备份或适用的回滚方案。按审查后的执行顺序运行；不要未经审查批量执行。
6	展开	展示当前复制任务的详细信息。常用选项：导出表配置：导出当前任务的库表配置，可在新建复制任务时快速导入，以快速创建多条相同复制对象的复制链路。告警规则：配置当前任务的告警策略。

附录一：PostgreSQL 15.2 → ClickHouse 23.8.3.48 数据类型映射

以下映射仅适用于 PostgreSQL 15.2 → ClickHouse 23.8.3.48。源类型为 PostgreSQL 15.2 源端目录 format_type 的实际返回名称；目标类型为 ClickHouse system.columns 的逐列读回结果，包含实际的 Nullable 包装。别名以源目录返回的规范名称表示。

本表覆盖 PostgreSQL 15 官方通用内置类型、目录/系统类型、数组类型，以及本轮创建的用户定义类型样例。用户定义类型样例仅表示表中列出的具体类型，不扩展到其他用户定义类型。PostgreSQL 伪类型不能作为普通表列类型，因此不纳入列映射。映射表描述列类型元数据，不代表非空值转换或内容保真。

PostgreSQL 数据类型（源端目录名称）	ClickHouse 列类型（目标端读回）
`smallint`	`Nullable(Int16)`
`integer`	`Nullable(Int32)`
`bigint`	`Nullable(Int64)`
`numeric`	`Nullable(Decimal(76, 38))`
`numeric(12,4)`	`Nullable(Decimal(12, 4))`
`real`	`Nullable(Float64)`
`double precision`	`Nullable(Float64)`
`boolean`	`Nullable(Bool)`
`character(3)`	`Nullable(FixedString(9))`
`character varying(20)`	`Nullable(String)`
`text`	`Nullable(String)`
`bytea`	`Nullable(String)`
`date`	`Nullable(Date32)`
`time(3) without time zone`	`Nullable(String)`
`time(3) with time zone`	`Nullable(String)`
`timestamp(3) without time zone`	`Nullable(DateTime64(3))`
`timestamp(3) with time zone`	`Nullable(DateTime64(3))`
`interval`	`Nullable(String)`
`interval(3)`	`Nullable(String)`
`interval year to month`	`Nullable(String)`
`interval day to second(3)`	`Nullable(String)`
`bit(4)`	`Nullable(String)`
`bit varying(8)`	`Nullable(String)`
`json`	`Nullable(String)`
`jsonb`	`Nullable(String)`
`uuid`	`Nullable(UUID)`
`xml`	`Nullable(String)`
`money`	`Nullable(Decimal(19, 2))`
`box`	`Nullable(String)`
`circle`	`Nullable(String)`
`line`	`Nullable(String)`
`lseg`	`Nullable(String)`
`path`	`Nullable(String)`
`point`	`Nullable(String)`
`polygon`	`Nullable(String)`
`cidr`	`Nullable(String)`
`inet`	`Nullable(String)`
`macaddr`	`Nullable(String)`
`macaddr8`	`Nullable(String)`
`tsquery`	`Nullable(String)`
`tsvector`	`Nullable(String)`
`jsonpath`	`Nullable(String)`
`oid`	`Nullable(String)`
`regclass`	`Nullable(String)`
`regcollation`	`Nullable(String)`
`regconfig`	`Nullable(String)`
`regdictionary`	`Nullable(String)`
`regnamespace`	`Nullable(String)`
`regoper`	`Nullable(String)`
`regoperator`	`Nullable(String)`
`regproc`	`Nullable(String)`
`regprocedure`	`Nullable(String)`
`regrole`	`Nullable(String)`
`regtype`	`Nullable(String)`
`xid`	`Nullable(String)`
`xid8`	`Nullable(String)`
`cid`	`Nullable(String)`
`tid`	`Nullable(String)`
`pg_lsn`	`Nullable(String)`
`pg_snapshot`	`Nullable(String)`
`txid_snapshot`	`Nullable(String)`
`refcursor`	`Nullable(String)`
`name`	`Nullable(String)`
`bpchar`	`Nullable(String)`
`"char"`	`Nullable(String)`
`aclitem`	`Nullable(String)`
`gtsvector`	`Nullable(String)`
`int2vector`	`Nullable(String)`
`oidvector`	`Nullable(String)`
`int4range`	`Nullable(String)`
`int8range`	`Nullable(String)`
`numrange`	`Nullable(String)`
`tsrange`	`Nullable(String)`
`tstzrange`	`Nullable(String)`
`daterange`	`Nullable(String)`
`int4multirange`	`Nullable(String)`
`int8multirange`	`Nullable(String)`
`nummultirange`	`Nullable(String)`
`tsmultirange`	`Nullable(String)`
`tstzmultirange`	`Nullable(String)`
`datemultirange`	`Nullable(String)`
`本轮定义的用户枚举类型`	`Nullable(String)`
`本轮定义的用户域类型`	`Nullable(Int32)`
`本轮定义的用户复合类型`	`Nullable(String)`
`本轮定义的用户范围类型`	`Nullable(String)`
`本轮定义的用户多重范围类型`	`Nullable(String)`
`smallint`	`Int16`
`integer`	`Int32`
`bigint`	`Int64`
`aclitem[]`	`Nullable(String)`
`bit(1)[]`	`Nullable(String)`
`boolean[]`	`Nullable(String)`
`box[]`	`Nullable(String)`
`character(1)[]`	`Nullable(String)`
`bytea[]`	`Nullable(String)`
`"char"[]`	`Nullable(String)`
`cid[]`	`Nullable(String)`
`cidr[]`	`Nullable(String)`
`circle[]`	`Nullable(String)`
`date[]`	`Nullable(String)`
`datemultirange[]`	`Nullable(String)`
`daterange[]`	`Nullable(String)`
`real[]`	`Nullable(String)`
`double precision[]`	`Nullable(String)`
`gtsvector[]`	`Nullable(String)`
`inet[]`	`Nullable(String)`
`smallint[]`	`Nullable(String)`
`integer[]`	`Nullable(String)`
`int4multirange[]`	`Nullable(String)`
`int4range[]`	`Nullable(String)`
`bigint[]`	`Nullable(String)`
`int8multirange[]`	`Nullable(String)`
`int8range[]`	`Nullable(String)`
`interval[]`	`Nullable(String)`
`json[]`	`Nullable(String)`
`jsonb[]`	`Nullable(String)`
`jsonpath[]`	`Nullable(String)`
`line[]`	`Nullable(String)`
`lseg[]`	`Nullable(String)`
`macaddr[]`	`Nullable(String)`
`macaddr8[]`	`Nullable(String)`
`money[]`	`Nullable(String)`
`name[]`	`Nullable(String)`
`numeric[]`	`Nullable(String)`
`nummultirange[]`	`Nullable(String)`
`numrange[]`	`Nullable(String)`
`oid[]`	`Nullable(String)`
`oidvector[]`	`Nullable(String)`
`path[]`	`Nullable(String)`
`pg_lsn[]`	`Nullable(String)`
`pg_snapshot[]`	`Nullable(String)`
`point[]`	`Nullable(String)`
`polygon[]`	`Nullable(String)`
`refcursor[]`	`Nullable(String)`
`regclass[]`	`Nullable(String)`
`regcollation[]`	`Nullable(String)`
`regconfig[]`	`Nullable(String)`
`regdictionary[]`	`Nullable(String)`
`regnamespace[]`	`Nullable(String)`
`regoper[]`	`Nullable(String)`
`regoperator[]`	`Nullable(String)`
`regproc[]`	`Nullable(String)`
`regprocedure[]`	`Nullable(String)`
`regrole[]`	`Nullable(String)`
`regtype[]`	`Nullable(String)`
`text[]`	`Nullable(String)`
`tid[]`	`Nullable(String)`
`time without time zone[]`	`Nullable(String)`
`timestamp without time zone[]`	`Nullable(String)`
`timestamp with time zone[]`	`Nullable(String)`
`time with time zone[]`	`Nullable(String)`
`tsmultirange[]`	`Nullable(String)`
`tsquery[]`	`Nullable(String)`
`tsrange[]`	`Nullable(String)`
`tstzmultirange[]`	`Nullable(String)`
`tstzrange[]`	`Nullable(String)`
`tsvector[]`	`Nullable(String)`
`txid_snapshot[]`	`Nullable(String)`
`uuid[]`	`Nullable(String)`
`bit varying[]`	`Nullable(String)`
`character varying[]`	`Nullable(String)`
`xid[]`	`Nullable(String)`
`xid8[]`	`Nullable(String)`
`xml[]`	`Nullable(String)`
`int2vector[]`	`Nullable(String)`
`本轮定义的用户枚举类型数组`	`Nullable(String)`
`本轮定义的用户域类型数组`	`Nullable(String)`
`本轮定义的用户复合类型数组`	`Nullable(String)`
`本轮定义的用户范围类型数组`	`Nullable(String)`
`本轮定义的用户多重范围类型数组`	`Nullable(String)`

类型说明：

本次 character(3) 列的目标类型为 Nullable(FixedString(9))；numeric(12,4) 映射为 Nullable(Decimal(12, 4))，无精度/标度声明的 numeric 映射为 Nullable(Decimal(76, 38))。
PostgreSQL int2vector、oidvector 两种标量列的自动生成结构在本次任务中未能直接用于 ClickHouse；将相应目标列结构修正为 Nullable(String) 后，结构复制和全量复制完成。因此这两项反映修正后的目标结构，不代表自动生成结构可直接使用。
目录中以 smallint、integer 和 bigint 返回的列同时出现可空和非空目标类型；本表保留目标端实际的 Nullable 包装，不据此说明序列或默认值的迁移行为。
类型候选范围参考 PostgreSQL 15 数据类型、数组类型、伪类型及 pg_type 系统目录。

附录二：预检查项一览表

检查项	检查内容
目标库数据存在性检查	检查待复制对象在目标数据库中是否已存在数据
目标库同名对象存在性检查	检查待复制对象在目标数据库中是否已存在
源数据源连接检查	检查源数据源网关状态、实例是否可达、用户名及密码准确性
目标数据源连接检查	检查目标数据源网关状态、实例是否可达、用户名及密码准确性
目标库权限检查	检查目标数据库的账号权限是否满足要求
源库权限检查	检查源数据库的账号权限是否满足要求

PostgreSQL 迁移同步到 ClickHouse

功能介绍​

前提条件​

注意事项​

使用限制​

操作步骤​

查看同步结果​

附录一：PostgreSQL 15.2 → ClickHouse 23.8.3.48 数据类型映射​

附录二：预检查项一览表​

相关文档​