Bottled Water: 实时集成postgresql与kafka

前端之家收集整理的这篇文章主要介绍了Bottled Water: 实时集成postgresql与kafka前端之家小编觉得挺不错的,现在分享给大家,也给大家做个参考。
摘要: Bottled Water是Confluent公司开发的一款可以将postgresql数据库转换为kafka events的工具。

Bottled Water简介

Bottled Water是Confluent公司开发的一款可以将postgresql数据库转换为kafka events的工具。

Bottled Water可以实时地将postgresql的变化推送至kafka中。有以下几个主要特性:

  • 支持postgresql 9.4及以后版本。不限数据库的schema。
  • 无需schema changes,无需触发器及额外表。(但是需要添加一个扩展)
  • 几乎不影响数据库本身性能
  • 事务一致性的输出。即: 只有写入操作成功commit数据库之后才会将变化写入kafka,顺序与commit顺序一致。
  • 容错。在进程崩溃、机器故障、网络中断等情况下不会丢失数据。

为什么使用Bottled Water

在某些场景下,可能需要将数据存储在不止一个数据存储中。比如,为了搜索方便快速,可能会考虑冗余存储一份数据到elasticsearch或某些缓存数据库中;为了迁移到不同的数据库,在迁移过程中也可能会同时存储多种不同的数据库

这种需求下,常见的一种实现方式是“dual writes”,即应用程序内部做这样的处理 —— 将变更的数据同时并行写入不同的数据库中。如下图所示:

"dual writes"的做法是有隐患的,会受到竞态条件(race conditions)和可靠性问题的影响,造成不同的数据库之间的差异越来越大,最终变得难以修复。

数据库快照中重建缓存或索引有利于消除从dump重建的不一致性。但是,在一个大型数据库这个过程将是十分缓慢低效的,需要找到一种方案使其更快。

如果能模仿数据库主从之间复制的过程,将数据库修改变成流,携带每一次修改的消息,并且完全按照写入顺序重放这些修改,那么将得到一份一模一样的数据库,并且这种做法比dual writes效率好,可靠性更高。

一个更加理想的工作流程如下图所示:

Bottled Water就是按照上述工作流程设计的工具,将postgresql的变更存入kafka中。

Bottled Water工作原理

Bottled Water就是模拟数据库主从复制的过程,将数据库的变化写入kafka中。

其中,这个过程使用了postgresql 9.4版本引入的新特性logical decoding,可以从一个数据库提取出consistent snapshot和连续的change events流,提取出的数据按照行级别推送至kafka中。默认情况下,会在kafka中对每个表创建一个同名topic,以存储change events。

有关逻辑解码(logical decoding)的概念和过程描述,可以参考postgresql的文档:http://www.postgres.cn/docs/9.4/logicaldecoding-explanation.html

Bottled Water项目在使用中包含了两个部分:

  • postgresql的扩展bottledwater。
  • 一个可执行程序bottledwater,负责从postgresql中将变更内容推送至kafka中。

bottledwater这个可执行程序在启动时会创建一个名为bottledwater的复制槽(通过命令行参数可修改复制槽的名字,复制槽如果已存在就不会新建),以存储数据库的更改,并且追踪已消耗的diff。由于复制槽会保留最后被消耗掉的diff位置,因此,即使bottledwater正常或异常退出,也不会丢失数据库diff追踪,下次启动会接着上次的break point继续执行。由于复制槽本身不保留消费者的状态,每个消费者只能得到最后一个消费者停止消费之后的修改,所以特别注意不要启动多个bottledwater连接同一个复制槽。

Bottled Water使用方法

安装

按照官方文档的说明,有三种方法可以上手使用: docker、源码编译、ubuntu ppa。本文将使用第三种方案,使用的环境为Ubuntu 14.04,postgresql 9.4讲解。

添加Bottled Water的ppa源,然后安装两个软件包即可:

$ sudo apt-add-repository -y ppa:stub/bottledwater
$ sudo apt-get update
$ sudo apt-get install postgresql-9.4-bottledwater bottledwater

使用上述命令安装postgresql-9.4-bottledwater扩展和bottledwater可执行程序。

使用前配置

Bottled Water会连接到postgresql获取相关数据,连接的账户需要有replication权限,pg中数据库的变化存储在WAL中(类似于MysqLbinlog),至少需要replication权限才能读取WAL。

需要修改WAL的日志级别为logical。本例中,我们将创建一个repl账户,具有REPLICATION权限。

打开/etc/postgresql/9.4/main/postgresql.conf文件修改以下属性:

wal_level = logical             # 默认WAL级别minimal,不符合要求,需要调整为logical
max_wal_senders = 8
wal_keep_segments = 4
max_replication_slots = 4

打开/etc/postgresql/9.4/main/pg_hba.conf增加以下内容:

local   replication     repl                 trust
host    replication     repl  127.0.0.1/32   trust
host    replication     repl  ::1/128        trust

重启postgresql服务,让配置生效:

sudo service postgresql restart

管理员身份(postgres)登录psql,进行创建角色操作:

$ sudo -u postgres psql
psql (9.4.8)
Type "help" for help.

postgres=#
CREATE ROLE repl WITH REPLICATION PASSWORD 'password' LOGIN; 'password' LOGIN;

至此使用前的配置已全部完成。

Bottled Water使用演示

我们创建一个数据库mybw,包含一个表test,让Bottled Water追踪mybw.test这张表的变化。

DATABASE mybw; \c mybw CREATE EXTENSION bottledwater; -- 创建bottledwater扩展 TABLE test (id serial primary key,value text); text);

在另一个终端,启动bottledwater可执行程序:

$ bottledwater -d postgres://repl:password@127.0.0.1/mybw -b 172.16.250.10:9092 -f json

解释一下上述命令行:

  • -d参数指定postgresql的URL,这里使用repl账户,密码password连接至mybw数据库
  • -b参数指定kafka broker,以逗号分割多个broker
  • -f参数指定输出格式为json,默认格式为avro。为了演示简单这里选择了json

执行成功后,终端输出如下,应该无任何报错:

Writing messages to Kafka in JSON format
Created replication slot "bottledwater",capturing consistent snapshot "0049D495-1".
INFO:  bottledwater_export: Table public.test is keyed by index test_pkey
Snapshot complete,streaming changes from 2/44DAB970.

现在向mybw.test这张表插入一条记录:

INSERT INTO value) values('hello world!'); 'hello world!');

此时从kafka中可以看到如下的消息:

$ kafka-console-consumer --zookeeper localhost --topic test --from-beginning --property print.key=true {"id": {"int": 1}} {1},"value": {"string": "hello world!"}} $ kafka-"hello world!"}}

继续执行以下会修改数据库sql:

'hello'); UPDATE test SET value = 'world' WHERE id = 2; DELETE FROM 2; 2;

那么可以观察到kafka中多了以下几条消息:

{2}}	{2},147);">"hello"}}
{"world"}}
{2}}	null

官方文档中,对kafka中消息格式的描述如下:

  • Bottled Water为每个数据库表在kafka中创建一个同名的topic(可以通过-p参数设置topic前缀)
  • 每条消息使用表的主键作为kafka message的key,整行记录作为message value
  • insert和update操作,message value将会是行的新值
  • delete操作,message value将会是null
  • 如果表没有主键,Bottled Water将不会处理此表。可以通过--allow-unkeyed参数强行使用

最后,只需要使用一个kafka consumer连接kafka就可以实时获取数据库表的变更了。

总结

Bottled Water总体来说使用还是很方便的,并且响应速度十分灵敏。即使停掉bottledwater进程,下次启动时也会继续上次的断点,不会丢失数据。需要使用postgresql与kafka结合方案的可以考虑。

参考资料

猜你在找的Postgre SQL相关文章