wal2json

wal2json简介

wal2json是逻辑解码插件，使用该插件可以访问由INSERT和UPDATE生成的元组，解析WAL中的内容。

wal2json插件会在每个事务中生成一个JSON对象。JSON对象中提供了所有新/旧元组，额外选项还可以包括事务时间戳、限定架构、数据类型、事务ID等属性。

wal2json安装

访问MogDB下载页面，下载所需版本的wal2json插件。

解压插件包，例如：

tar -xzvf wal2json-2.3-3.1.0-01-CentOS-x86_64.tar.gz

进入插件所在目录下，执行make install命令。
```
cd wal2json/
make install
```

修改数据库参数配置

修改数据库数据目录下postgresql.conf（主备都需要修改）。
```
wal_level = logical
max_replication_slots = 10
max_wal_senders = 10
```

修改pg_hba.conf（主备都需要修改）。

host   replication   all     127.0.0.1/32       trust

执行完成后重启数据库。

wal2json使用

重新开一个终端执行使用wal2json连接数据库。

pg_recvlogical -d postgres --slot test2_slot --create -P wal2json -h 127.0.0.1 -p 5001 -U test
pg_recvlogical -d postgres -h 127.0.0.1 -p 5001 -U test --slot test2_slot --start  -f -

sql端进行数据库增删改查操作。

create table ff(id int, name int);
insert into ff values(1,2);
insert into ff values(2,4);
select * from ff;

wal2json端输出操作过程。

{"change":[]}
{"change":[]}
{"change":[{"kind":"insert","schema":"test","table":"ff","columnnames":["id","name"],"columntypes":["integer","integer"],"columnvalues":[1,10]}]}
{"change":[{"kind":"insert","schema":"test","table":"ff","columnnames":["id","name"],"columntypes":["integer","integer"],"columnvalues":[2,2]}]}

wal2json参数

在数据库数据目录下执行：

pg_recvlogical --help
 -f, --file=FILE     receive log into this file. - for stdout
 -n, --no-loop      do not loop on connection lost
 -v, --verbose      output verbose messages
 -V, --version      output version information, then exit
 -?, --help       show this help, then exit

Connection options:
 -d, --dbname=DBNAME   database to connect to
 -h, --host=HOSTNAME   database server host or socket directory
 -p, --port=PORT     database server port number
 -U, --username=NAME   connect as specified database user
 -w, --no-password    never prompt for password
 -W, --password     force password prompt (should happen automatically)

Replication options:
 -F  --fsync-interval=INTERVAL
            frequency of syncs to the output file (in seconds, defaults to 10)
 -o, --option=NAME[=VALUE]
            Specify option NAME with optional value VAL, to be passed
            to the output plugin
 -P, --plugin=PLUGIN   use output plugin PLUGIN (defaults to mppdb_decoding)
 -s, --status-interval=INTERVAL
            time between status packets sent to server (in seconds, defaults to 10)
 -S, --slot=SLOT     use existing replication slot SLOT instead of starting a new one
 -I, --startpos=PTR   Where in an existing slot should the streaming start

Action to be performed:
   --create      create a new replication slot (for the slotname see --slot)
   --start       start streaming in a replication slot (for the slotname see --slot)
   --drop       drop the replication slot (for the slotname see --slot)

反馈