使用 amazonriver将postgres 数据实时同步 至elasticsearch （基于pg的逻辑复制功能）
使用 amazonriver将postgres 数据实时同步 至elasticsearch （基于pg的逻辑复制功能）
amazonriver
一、 简介
amazonriver 是一个将postgresql的增、删、改实时同步到es或kafka的服务（基于pg的逻辑复制功能），go语言实现，哈罗单车开源的一个项目。
git地址https://github.com/hellobike/amazonriver
也有java版本https://github.com/hellobike/tunnel（需依赖zookeeper）
版本支持
- Postgresql 9.4 or later
- Kafka 0.8 or later
- ElasticSearch 5.x
amazonriver 利用pg内部的逻辑复制功能,通过在pg创建逻辑复制槽,接收数据库的逻辑变更,通过解析test_decoding特定格式的消息,得到逻辑数据，与es的通讯是发布订阅的方式，
效果要比触发器好的多，并且配置简易
安装
下载最新版本
https://github.com/hellobike/amazonriver/releases
压缩包中amazonriver是可执行文件，放到/opt/amazonriver目录下
二、 配置
amazonriver 配置
在该目录下新增一个配置文件 config.json
config.json配置内容为
{
“pg_dump_path”: “”,
“subscribes”: [{
#历史数据导入的功能，测试发现无法使用
“dump”: false,
#若有多个amazonriver实例，这里要配置成不一样的
“slotName”: “slot_for_es1”,
“pgConnConf”: {
#pg数据库配置
“host”: “10.168.4.91”,
“port”: 5321,
“database”: “xzdssituation”,
“schema”: “public”,
“user”: “postgres”,
“password”: “postgres”
},
“rules”: [
{
#表名
“table”: “ds_command”,
#表的主键
“pks”: [“command_id”],
#es中的主键
“esid”: [“command_id”],
“index”: “ds_command”,
#固定为data, 禁止以下划线开头
“type”: “data”
},
{
“table”: “ds_alarm”,
“pks”: [“command_id”],
“esid”: [“command_id”],
“index”: “ds_alarm”,
“type”: “data”
}
],
“esConf”: {
#elasticsearch 地址
“addrs”: “http://10.168.4.60:9200/”,
“user”: “”,
“password”: “”
},
#失败重试0为不重试,-1会一直重试直到成功
“retry”: -1
}],
# 填”:8080″会开启端口，供普罗米修斯监控，这里不配置
“prometheus_address”: “”
}
PostgreSQL 配置
1.修改/app/pgsql_data/postgresql.conf
wal_level = 'logical';max_replication_slots = 5; #该值要大于1
修改后需要重启才能生效
2.创建有replication权限的用户
CREATE ROLE syn_rep LOGIN ENCRYPTED PASSWORD 'postgres' REPLICATION;
GRANT CONNECT ON DATABASE xzdssituation to syn_rep;
COMMIT;
3.修改白名单配置
在 /app/pgsql_data/pg_hba.conf 中增加配置:
host replication syn_rep all md5
修改后需要reload才能生效
4.重启pg
su – postgres
./stop-pgsql.sh
./start-pgsql.sh
三、 启停
先给amazonriver加可执行权限
chmod +x /opt/amazonriver/amazonriver
启动（指定配置 和 日志）
已写成脚本：/opt/amazonriver/start.sh
nohup /opt/amazonriver/amazonriver -config /opt/amazonriver/config.json -level debug > /opt/amazonriver/logfile.log 2>&1 & echo $! > /opt/amazonriver/pid
停止
已写成脚本：/opt/amazonriver/stop.sh
kill `cat /opt/amazonriver/pid`
四、 日志查看
tail -200f /opt/amazonriver/logfile.log
附录：
/app/pgsql/bin/pg_ctl 控制pg启停
pg_ctl start [-w] [-s] [-D datadir] [-l filename] [-o options] [-p path]
pg_ctl stop
[-W] [-s] [-D datadir]
[-m s[mart] | f[ast] | i[mmediate] ]
pg_ctl restart
[-w] [-s] [-D datadir]
[-m s[mart] | f[ast] | i[mmediate] ] [-o options]
pg_ctl reload
[-s] [-D datadir]
pg_ctl status
[-D datadir]
pg_ctl kill
[signal_name]
[process_id]
pg_ctl register
[-N servicename]
[-U username]
[-P password]
[-D datadir]
[-w] [-o options]
pg_ctl unregister
[-N servicename]