Skip to content

README.md

Latest commit

 

History

History
693 lines (512 loc) · 13.4 KB

README.md

File metadata and controls

693 lines (512 loc) · 13.4 KB
Raw

go-mysql-postgres

English | 中文

English

go-mysql-postgres syncs MySQL table data to PostgreSQL.

It can load an initial snapshot with mysqldump, then keep PostgreSQL updated from the MySQL binlog. The current replication position is stored as master.info under data_dir, so the process can resume after restart.

Features

  • Initial snapshot sync with mysqldump.
  • Incremental binlog sync for insert, update and delete events.
  • Primary-key update handling as delete plus insert.
  • Basic table structure sync for ALTER TABLE changes.
  • Column mapping, column filtering and wildcard table rules.
  • Optional routing to named PostgreSQL targets.
  • HTTP status endpoint.

Requirements

  • Go 1.26 or newer.
  • MySQL with row-based binlog enabled.
  • PostgreSQL target tables must exist before data sync.
  • Synced MySQL tables should have a primary key unless skip_no_pk_table = true.
  • mysqldump is required for initial snapshot sync. If it is empty, the process starts from the saved binlog position instead.

Recommended MySQL settings:

[mysqld]
server-id=1
log-bin=mysql-bin
binlog-format=ROW
binlog-row-image=FULL

The MySQL user needs permission to read table metadata and consume binlogs.

Build

make

or:

go build -o bin/go-mysql-postgres ./cmd/go-mysql-postgres

Run

./bin/go-mysql-postgres -config=./etc/river.toml

Common flags:

-config      config file path
-my_addr     MySQL address
-my_user     MySQL user
-my_pass     MySQL password
-data_dir    position data directory
-server_id   MySQL replication client server id
-flavor      mysql or mariadb
-exec        mysqldump executable path
-max_procs   GOMAXPROCS

Configuration

See etc/river.toml for a complete example.

Minimal configuration:

my_addr = "127.0.0.1:3306"
my_user = "root"
my_pass = ""
my_charset = "utf8"

pg_host = "127.0.0.1"
pg_port = 5432
pg_user = "postgres"
pg_pass = ""
pg_dbname = "test"
pg_maxconn = 16

data_dir = "./var"
stat_addr = "127.0.0.1:12800"
server_id = 1001
flavor = "mysql"
mysqldump = "mysqldump"

[[source]]
schema = "test"
tables = ["orders"]

[[rule]]
schema = "test"
table = "orders"
pg_schema = "public"
pg_table = "orders"

Sources And Rules

[[source]] selects MySQL tables:

[[source]]
schema = "test"
tables = ["orders", "orders_[0-9]{4}"]

[[rule]] maps MySQL source tables to PostgreSQL target tables:

[[rule]]
schema = "test"
table = "orders"
pg_schema = "public"
pg_table = "orders"
id = ["id"]

Wildcard source tables must appear in both [[source]] and [[rule]]:

[[source]]
schema = "test"
tables = ["orders_[0-9]{4}"]

[[rule]]
schema = "test"
table = "orders_[0-9]{4}"
pg_schema = "public"
pg_table = "orders"

Field Mapping

By default, MySQL column names are written to columns with the same PostgreSQL names.

Use [rule.field] to map fields:

[[rule]]
schema = "test"
table = "order_fields"
pg_schema = "public"
pg_table = "order_fields"

[rule.field]
id = "pg_id"
tags = "pg_tags,list"
created_at = ",date"

Field mapping values use target_column[,modifier].

  • list converts comma-separated strings into arrays.
  • date converts numeric MySQL values through the date conversion path.

Use filter to sync only selected columns:

[[rule]]
schema = "test"
table = "order_filter"
pg_schema = "public"
pg_table = "order_filter"
filter = ["id", "name"]

Use skip_actions to ignore selected row events:

skip_actions = ["delete"]

Use skip_alter_actions to ignore selected table structure changes:

skip_alter_actions = ["drop"]

Multiple PostgreSQL Targets

Configure named targets with [[target]]:

[[target]]
pg_name = "archive"
pg_host = "127.0.0.1"
pg_port = 5432
pg_user = "postgres"
pg_pass = ""
pg_dbname = "archive"
pg_maxconn = 16

Select a target from a rule:

[[rule]]
schema = "test"
table = "orders"
pg_name = "archive"
pg_schema = "public"
pg_table = "orders"

Data routing is also supported:

[[rule.data_routers]]
[rule.data_routers.field_filters]
region = "east"

[rule.data_routers.target]
data_source = "archive"
schema_name = "public"
table_name = "orders_east"

Status

When stat_addr is configured, the process exposes:

/stat    current binlog and event counters
/config  current loaded configuration
/metrics Prometheus metrics (insert/update/delete counts, binlog position, sync lag)
/healthz Health check endpoint (200 OK / 503 Service Unavailable)

Example:

curl http://127.0.0.1:12800/stat

Testing

Run unit tests:

go test ./...

Build the binary:

go build -o bin/go-mysql-postgres ./cmd/go-mysql-postgres

Run the Docker schema-change smoke test:

make smoke-schema-change
make smoke-field-type
make smoke-regression

The current Docker smoke tests used during development covered:

  • initial snapshot sync
  • insert, update and delete binlog events
  • primary-key update
  • ALTER TABLE ADD COLUMN
  • restart from master.info

Smoke Test Data

Use a clean MySQL database named smoke, a clean PostgreSQL database named smoke, and an empty data_dir.

MySQL source table:

CREATE TABLE orders (
  id INT NOT NULL PRIMARY KEY,
  name VARCHAR(64),
  amount INT,
  note VARCHAR(128)
);

INSERT INTO orders (id, name, amount, note) VALUES
  (1, 'initial', 10, 'dump row');

PostgreSQL target table:

CREATE TABLE orders (
  id INTEGER PRIMARY KEY,
  name CHARACTER VARYING(64),
  amount INTEGER,
  note CHARACTER VARYING(128)
);

After the initial snapshot sync succeeds, apply these MySQL changes:

INSERT INTO orders (id, name, amount, note)
VALUES (2, 'binlog insert', 20, 'insert ok');

UPDATE orders
SET name = 'binlog update', amount = 15, note = 'update ok'
WHERE id = 1;

DELETE FROM orders WHERE id = 2;

UPDATE orders
SET id = 3, name = 'pk moved', amount = 25, note = 'pk update ok'
WHERE id = 1;

ALTER TABLE orders ADD COLUMN extra VARCHAR(32);

UPDATE orders SET extra = 'ddl ok' WHERE id = 3;

INSERT INTO orders (id, name, amount, note, extra)
VALUES (4, 'after ddl', 40, 'insert after ddl', 'new column');

To test resume from master.info, stop the sync process, insert one more row in MySQL, then start the sync process again with the same data_dir:

INSERT INTO orders (id, name, amount, note, extra)
VALUES (5, 'after restart', 50, 'resume event', 'resume ok');

The final MySQL and PostgreSQL data should match:

id | name          | amount | note             | extra
3  | pk moved      | 25     | pk update ok     | ddl ok
4  | after ddl     | 40     | insert after ddl | new column
5  | after restart | 50     | resume event     | resume ok

License

MIT. See LICENSE.

中文

go-mysql-postgres 用于将 MySQL 表数据同步到 PostgreSQL。

它可以通过 mysqldump 加载初始快照，然后继续消费 MySQL binlog，让 PostgreSQL 保持更新。当前同步位点会保存到 data_dir 下的 master.info，进程重启后可以继续同步。

功能

  • 使用 mysqldump 进行初始快照同步。
  • 通过 binlog 增量同步 insert、update、delete。
  • 主键变更按 delete 加 insert 处理。
  • 支持基础 ALTER TABLE 表结构同步。
  • 支持字段映射、字段过滤和通配表规则。
  • 支持路由到命名 PostgreSQL 目标。
  • 提供 HTTP 状态接口。

运行要求

  • Go 1.26 或更新版本。
  • MySQL 必须开启 row-based binlog。
  • PostgreSQL 目标表需要在数据同步前存在。
  • 被同步的 MySQL 表建议有主键，除非设置 skip_no_pk_table = true。
  • 初始快照同步需要 mysqldump。如果 mysqldump 配置为空，进程会从已保存的 binlog 位点开始。

推荐 MySQL 配置：

[mysqld]
server-id=1
log-bin=mysql-bin
binlog-format=ROW
binlog-row-image=FULL

MySQL 用户需要具备读取表结构和消费 binlog 的权限。

构建

make

或者：

go build -o bin/go-mysql-postgres ./cmd/go-mysql-postgres

运行

./bin/go-mysql-postgres -config=./etc/river.toml

常用参数：

-config      配置文件路径
-my_addr     MySQL 地址
-my_user     MySQL 用户名
-my_pass     MySQL 密码
-data_dir    位点数据目录
-server_id   MySQL 复制客户端 server id
-flavor      mysql 或 mariadb
-exec        mysqldump 可执行文件路径
-max_procs   GOMAXPROCS

配置

完整示例见 etc/river.toml。

最小配置：

my_addr = "127.0.0.1:3306"
my_user = "root"
my_pass = ""
my_charset = "utf8"

pg_host = "127.0.0.1"
pg_port = 5432
pg_user = "postgres"
pg_pass = ""
pg_dbname = "test"
pg_maxconn = 16

data_dir = "./var"
stat_addr = "127.0.0.1:12800"
server_id = 1001
flavor = "mysql"
mysqldump = "mysqldump"

[[source]]
schema = "test"
tables = ["orders"]

[[rule]]
schema = "test"
table = "orders"
pg_schema = "public"
pg_table = "orders"

数据源与规则

[[source]] 用来选择 MySQL 表：

[[source]]
schema = "test"
tables = ["orders", "orders_[0-9]{4}"]

[[rule]] 用来把 MySQL 源表映射到 PostgreSQL 目标表：

[[rule]]
schema = "test"
table = "orders"
pg_schema = "public"
pg_table = "orders"
id = ["id"]

通配表必须同时出现在 [[source]] 和 [[rule]] 中：

[[source]]
schema = "test"
tables = ["orders_[0-9]{4}"]

[[rule]]
schema = "test"
table = "orders_[0-9]{4}"
pg_schema = "public"
pg_table = "orders"

字段映射

默认情况下，MySQL 字段会写入同名的 PostgreSQL 字段。

使用 [rule.field] 可以配置字段映射：

[[rule]]
schema = "test"
table = "order_fields"
pg_schema = "public"
pg_table = "order_fields"

[rule.field]
id = "pg_id"
tags = "pg_tags,list"
created_at = ",date"

字段映射值格式为 target_column[,modifier]。

  • list 将逗号分隔的字符串转换为数组。
  • date 将 MySQL 数值按日期转换路径处理。

使用 filter 可以只同步部分字段：

[[rule]]
schema = "test"
table = "order_filter"
pg_schema = "public"
pg_table = "order_filter"
filter = ["id", "name"]

使用 skip_actions 可以忽略指定行事件：

skip_actions = ["delete"]

使用 skip_alter_actions 可以忽略指定表结构变更：

skip_alter_actions = ["drop"]

多 PostgreSQL 目标

使用 [[target]] 配置命名目标：

[[target]]
pg_name = "archive"
pg_host = "127.0.0.1"
pg_port = 5432
pg_user = "postgres"
pg_pass = ""
pg_dbname = "archive"
pg_maxconn = 16

在规则里选择目标：

[[rule]]
schema = "test"
table = "orders"
pg_name = "archive"
pg_schema = "public"
pg_table = "orders"

也支持按数据内容路由：

[[rule.data_routers]]
[rule.data_routers.field_filters]
region = "east"

[rule.data_routers.target]
data_source = "archive"
schema_name = "public"
table_name = "orders_east"

状态接口

配置 stat_addr 后，进程会暴露：

/stat    当前 binlog 与事件计数
/config  当前加载的配置
/metrics Prometheus 指标（insert/update/delete 计数、binlog 位点、同步延迟）
/healthz 健康检查端点（200 OK / 503 Service Unavailable）

示例：

curl http://127.0.0.1:12800/stat

测试

运行单元测试：

go test ./...

构建二进制：

go build -o bin/go-mysql-postgres ./cmd/go-mysql-postgres

运行 Docker 表结构变更 smoke test：

make smoke-schema-change
make smoke-field-type
make smoke-regression

当前开发过程中的 Docker smoke tests 覆盖了：

  • 初始快照同步
  • insert、update、delete binlog 事件
  • 主键变更
  • ALTER TABLE ADD COLUMN
  • 从 master.info 重启续传

Smoke Test 测试数据

使用干净的 MySQL smoke 数据库、干净的 PostgreSQL smoke 数据库，并使用空的 data_dir。

MySQL 源表：

CREATE TABLE orders (
  id INT NOT NULL PRIMARY KEY,
  name VARCHAR(64),
  amount INT,
  note VARCHAR(128)
);

INSERT INTO orders (id, name, amount, note) VALUES
  (1, 'initial', 10, 'dump row');

PostgreSQL 目标表：

CREATE TABLE orders (
  id INTEGER PRIMARY KEY,
  name CHARACTER VARYING(64),
  amount INTEGER,
  note CHARACTER VARYING(128)
);

初始快照同步成功后，在 MySQL 执行这些变更：

INSERT INTO orders (id, name, amount, note)
VALUES (2, 'binlog insert', 20, 'insert ok');

UPDATE orders
SET name = 'binlog update', amount = 15, note = 'update ok'
WHERE id = 1;

DELETE FROM orders WHERE id = 2;

UPDATE orders
SET id = 3, name = 'pk moved', amount = 25, note = 'pk update ok'
WHERE id = 1;

ALTER TABLE orders ADD COLUMN extra VARCHAR(32);

UPDATE orders SET extra = 'ddl ok' WHERE id = 3;

INSERT INTO orders (id, name, amount, note, extra)
VALUES (4, 'after ddl', 40, 'insert after ddl', 'new column');

测试从 master.info 续传时，先停止同步进程，在 MySQL 插入一行，再使用相同 data_dir 重新启动同步进程：

INSERT INTO orders (id, name, amount, note, extra)
VALUES (5, 'after restart', 50, 'resume event', 'resume ok');

最终 MySQL 和 PostgreSQL 数据应保持一致：

id | name          | amount | note             | extra
3  | pk moved      | 25     | pk update ok     | ddl ok
4  | after ddl     | 40     | insert after ddl | new column
5  | after restart | 50     | resume event     | resume ok

许可证

MIT，见 LICENSE。