go-mysql-postgresql 配置文档解析

go-mysql-postgresql 增量同步工具
下载地址 https://github.com/frainmeng/go-mysql-elasticsearch/releases
下载go-mysql-postgresql3.0.0-linux-amd64.tar.gz
解压后，里面有启动脚本 start.sh stop.sh sync.log 是输出日志
建议配置一个logrotate 定时切割sync.log日志
这里有一个logrotate模板

/var/lib/pgsql/go-mysql-postgresql/sync.log {
        hourly
        rotate 24 
        missingok
        notifempty
        copytruncate
}

然后加入到定时任务

0 * * * * /usr/sbin/logrotate -f /etc/logrotate.d/sync
var/master.info binlog 日志位点
bin_name = "mysql-bin.007340"
bin_pos = 26694040

river.toml 配置文件详解

# MySQL address, user and password
# user must have replication privilege in MySQL.
my_addr = "192.168.1.1:3306"
my_user = "dts"
my_pass = "dts"
my_charset = "utf8"
#mysql 数据库源地址
# Set true when elasticsearch use https
#es_https = false
# postgresql address
#pg_addr = "127.0.0.1:5432"
# Elasticsearch user and password, maybe set by shield, nginx, or x-pack
#es_user = ""
#es_pass = ""
pg_host = "localhost"
pg_port = 5432
pg_user = "postgres"
pg_pass = "postgres"
pg_dbname = "testdb"

postgresql 目标数据库地址
# Path to store data, like master.info, if not set or empty,
# we must use this to support breakpoint resume syncing. 
# TODO: support other storage, like etcd. 
data_dir = "./var"

# Inner Http status address
stat_addr = "192.168.1.1:12800"
#监控地址用网页打开192.168.1.1:12800
# statsd monitor
statsd_host = "127.0.0.1"
statsd_port = 8125
statsd_prefix = "test"

# pseudo server id like a slave 
server_id = 12363
#mysql server_id
# mysql or mariadb
flavor = "mysql"

# mysqldump execution path
# if not set or empty, ignore mysqldump.
#mysqldump = "/usr/sbin/mysqldump"
#这是mysqldump路径地址，现在同步全量，通过mysqldump将数据同步到pg上，支持不怎么好，自己用mysql2pgsql同步表结构，用dts工具同步一份全量数据
# if we have no privilege to use mysqldump with --master-data,
# we must skip it.
#skip_master_data = false

# minimal items to be inserted in one bulk
bulk_size = 1

# force flush the pending requests if we don't have enough items >= bulk_size
flush_bulk_time = "500ms"

# Ignore table without primary key
skip_no_pk_table = false

# 
concurrent_size = 6
concurrent_ack_win = 2048
#上面几个是优化参数，根据自己环境进行优化
#下面是目标和源映射关系
# MySQL data source
[[source]]
schema = "testdb1"

# Only below tables will be synced into Elasticsearch.
# "t_[0-9]{4}" is a wildcard table format, you can use it if you have many sub tables, like table_0000 - table_1023
# I don't think it is necessary to sync all tables in a database.
tables=["t1"]
# Below is for special rule mapping

[[source]]
schema = "testdb2"
tables=["t2"]

# Very simple example
# 
# desc t;
# +-------+--------------+------+-----+---------+-------+
# | Field | Type         | Null | Key | Default | Extra |
# +-------+--------------+------+-----+---------+-------+
# | id    | int(11)      | NO   | PRI | NULL    |       |
# | name  | varchar(256) | YES  |     | NULL    |       |
# +-------+--------------+------+-----+---------+-------+
# 
# The table `t` will be synced to ES index `coupons` and type `t`.
[[target]]
pg_name = "localhost_testdb1"  这个是用来区分pg多个库
pg_host = "localhost"
pg_port = 5432
pg_user = "postgres"
pg_pass = "postgres"
pg_dbname = "testdb1"

[[target]]
pg_name = "localhost_testdb2"
pg_host = "localhost"
pg_port = 5432
pg_user = "postgres"
pg_pass = "postgres"
pg_dbname = "report"


[[rule]]
schema = "testdb1"
table = "t1"
pg_schema = "public"
pg_table = "t1"
pg_name="localhost_testdb1"
# skip_actions available values ["insert", "delete","update"]
# # skip_actions = ["insert", "delete","update"]
#
# # skip_alter_actions available values ["ADD", "DROP","ALTER"]
skip_alter_actions = ["ALTER","ADD","DROP"]



[[rule]]
schema = "testdb2"
table = "t2"
pg_schema = "public"
pg_table = "t2"
pg_name="localhost_testdb2"
# skip_actions available values ["insert", "delete","update"]
# # skip_actions = ["insert", "delete","update"]
#
# # skip_alter_actions available values ["ADD", "DROP","ALTER"]
skip_alter_actions = ["ALTER","ADD","DROP"]
上面某些ddl,dml可以跳过