MongoDB -> kafka 高性能实时同步(sync 采集)mongodb数据到kafka解决方案

提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

写这篇博客的目的

让更多的人了解阿里开源的MongoShake可以很好满足mongodb到kafka高性能高可用实时同步需求（项目地址：https://github.com/alibaba/MongoShake，下载地址：https://github.com/alibaba/MongoShake/releases）。至此博客就结束了，你可以愉快地啃这个项目了。还是一起来看一下官方的描述：

MongoShake is a universal data replication platform based on MongoDB's oplog. Redundant replication and active-active replication are two most important functions. 基于mongodb oplog的集群复制工具，可以满足迁移和同步的需求，进一步实现灾备和多活功能。

没有标题的标题

哈哈，有兴趣听我啰嗦的可以往下。最近，有个实时增量采集mongodb数据(数据量在每天10亿条左右)的需求，需要先调研一下解决方案。我分别百度、google了mongodb kafka sync 同步采集实时等关键词，写这篇博客的时候排在最前面的当属kafka-connect(官方有实现https://github.com/mongodb/mongo-kafka，其实也有非官方的实现)那一套方案，我对kafka-connect相对熟悉一点(不熟悉的话估计编译部署都要花好一段时间)，没测之前就感觉可能不满足我的采集性能需求，测下来果然也是不满足需求。后来，也看到了https://github.com/rwynn/route81，编译部署也较为麻烦，同样不满足采集性能需求。我搜索东西的时候一般情况下不会往下翻太多，没找到所需的，大多会尝试换关键词(包括中英文)搜搜，这次可能也提醒我下次要多往下找找，说不定有些好东西未必排在最前面几个。

之后在github上搜in:readme mongodb kafka sync，让我眼前一亮。

点进去快速读了一下readme，正是我想要的(后面自己实际测下来确实高性能、高可用，满足我的需求)，官方也提供了MongoShake的性能测试报告。

这篇博客不讲(也很大可能是笔者技术太渣，无法参透领会(●´ω｀●))MongoShake的架构、原理、实现，如何高性能的，如何高可用的等等。就一个目的，希望其他朋友在搜索mongodb kafka时候，MongoShake的解决方案可以排在最前面。

初次使用MongoShake值得注意的地方

数据处理流程

v2.2.1之前的MongoShake版本处理数据的流程：

MongoDB(数据源端，待同步的数据)

-->MongoShake(对应的是collector.linux进程，作用是采集)

-->Kafka(raw格式，未解析的带有header+body的数据)

-->receiver(对应的是receiver.linux进程，作用是解析，这样下游组件就能拿到比如解析好的一条一条的json格式的数据)

-->下游组件(拿到mongodb中的数据用于自己的业务处理)

v2.2.1之前MongoShake的版本解析入kafka，需要分别启collector.linux和receiver.linux进程，而且receiver.linux需要自己根据你的业务逻辑填充完整,然后编译出来，默认只是把解析出来的数据打个log而已

src/mongoshake/receiver/replayer.go中的代码如图：

详情见：https://github.com/alibaba/MongoShake/wiki/FAQ#q-how-to-connect-to-different-tunnel-except-direct

v2.2.1版本MongoShake的collector.conf有一个配置项tunnel.message

# the message format in the tunnel, used when tunnel is kafka. # "raw": batched raw data format which has good performance but encoded so that users # should parse it by receiver. # "json": single oplog format by json. # "bson": single oplog format by bson. # 通道数据的类型，只用于kafka和file通道类型。 # raw是默认的类型，其采用聚合的模式进行写入和 # 读取，但是由于携带了一些控制信息，所以需要专门用receiver进行解析。 # json以json的格式写入kafka，便于用户直接读取。 # bson以bson二进制的格式写入kafka。 tunnel.message = json

如果选择的raw格式，那么数据处理流程和上面之前的一致(MongoDB->MongoShake->Kafka->receiver->下游组件)
如果选择的是json、bson,处理流程为MongoDB->MongoShake->Kafka->下游组件

v2.2.1版本设置为json处理的优点就是把以前需要由receiver对接的格式，改为直接对接，从而少了一个receiver，也不需要用户额外开发，降低开源用户的使用成本。

简单总结一下就是：

raw格式能够最大程度的提高性能，但是需要用户有额外部署receiver的成本。json和bson格式能够降低用户部署成本，直接对接kafka即可消费，相对于raw来说，带来的性能损耗对于大部分用户是能够接受的。

高可用部署方案

我用的是v2.2.1版本，高可用部署非常简单。collector.conf开启master的选举即可：

# high availability option. # enable master election if set true. only one mongoshake can become master # and do sync, the others will wait and at most one of them become master once # previous master die. The master information stores in the `mongoshake` db in the source # database by default. # 如果开启主备mongoshake拉取同一个源端，此参数需要开启。 master_quorum = true# checkpoint存储的地址，database表示存储到MongoDB中，api表示提供http的接口写入checkpoint。 context.storage = database

同时我checkpoint的存储地址默认用的是database，会默认存储在mongoshake这个db中。我们可以查询到checkpoint记录的一些信息。

rs0:PRIMARY> use mongoshake switched to db mongoshake rs0:PRIMARY> show collections; ckpt_default ckpt_default_oplog election rs0:PRIMARY> db.election.find() { "_id" : ObjectId("5204af979955496907000001"), "pid" : 6545, "host" : "192.168.31.175", "heartbeat" : NumberLong(1582045562) }

我在192.168.31.174，192.168.31.175，192.168.31.176上总共启了3个MongoShake实例，可以看到现在工作的是192.168.31.175机器上进程。自测过程，高速往mongodb写入数据，手动kill掉192.168.31.175上的collector进程，等192.168.31.174成为master之后，我又手动kill掉它，最终只保留192.168.31.176上的进程工作，最后统计数据发现，有重采数据现象，猜测有实例还没来得及checkpoint就被kill掉了。

MongoDB -> kafka 高性能实时同步(sync 采集)mongodb数据到kafka解决方案的更多相关文章

MongoDB -&gt&semi; kafka 高性能实时同步(采集)mongodb数据到kafka解决方案
写这篇博客的目的让更多的人了解阿里开源的MongoShake可以很好满足mongodb到kafka高性能高可用实时同步需求(项目地址:https://github.com/alibaba/Mong ...
flume实时采集mysql数据到kafka中并输出
环境说明 centos7(运行于vbox虚拟机) flume1.9.0(flume-ng-sql-source插件版本1.5.3) jdk1.8 kafka(版本忘了后续更新) zookeeper(版 ...
Mongodb 和 Solr 实时同步
一.安装前准备 1.mongo-connector(基于python)中间件 2.python-3.4.3.msi 3.Mongodb 4.Solr 二.配置Mongodb集群 1).配置replic ...
使用脚本+kafka自带命令行工具统计数据写入kafka速率
思路每隔一段时间(比如说10秒)统计一次某topic的所有partition的最大offset值之和,这便是该topic的message总数. 然后除以间隔时间就可以粗略但方便得出某topic的数 ...
基于Canal和Kafka实现MySQL的Binlog近实时同步
前提近段时间,业务系统架构基本完备,数据层面的建设比较薄弱,因为笔者目前工作重心在于搭建一个小型的数据平台.优先级比较高的一个任务就是需要近实时同步业务系统的数据(包括保存.更新或者软删除)到一个另 ...
linux运维、架构之路-实时同步方案
一.inotify+rsync实时同步 1.介绍 inotify-tools是一种强大的.细粒度的.异步的文件系统事件监控机制,可以用来监控文件系统的事件.inotify-tools是 ...
maxwell实时同步mysql中binlog
概述 Maxwell是一个能实时读取MySQL二进制日志binlog,并生成 JSON 格式的消息,作为生产者发送给 Kafka,Kinesis.RabbitMQ.Redis.Google Cloud ...
5、Sersync实时同步实战
1.实时同步概述 1.什么是实时同步, 只要当前目录发生变化则会触发一个事件,事件触发后将变化的目录同步至远程服务器. 2.为什么要实时同步, 保证数据的连续性, 减少人力维护成本, 解决nfs单点故 ...
storm消费kafka实现实时计算
大致架构 * 每个应用实例部署一个日志agent * agent实时将日志发送到kafka * storm实时计算日志 * storm计算结果保存到hbase storm消费kafka 创建实时计算项 ...

随机推荐

CMS
一.任务简介: 开发简单的CMS.在数据库中创建新闻数据库表news,包含(题目.作者.日期.正文等字段):创建HTML模板文件:读取数据库所有数据的信息,并使用新闻信息替换模板文件中的占位符,从而 ...
iOS项目重构日记
如何重构首先,要对程序的一般架构烂熟于心,尤其是MVC,这是基本.还有就是分离存储和网络请求的逻辑. 对于一些常用的控件尽量分离复用,设置开关函数,适当的时候开启,不要的时候关闭,有必要的话还可以 ...
【测试】模拟一个全表扫描的sql，对其进行优化走索引，并且将执行计划稳定到baseLine。
①创建表t3: SQL> create table t3 (id int); Table created. SQL; rows created. ②开启自动捕获并修改时间格式: SQL> ...
css3 边框记
css3 边框 border属性在css1中就已经定义了,使用它可以设置元素的边框风格,边框颜色以及边框粗细. border-width:设置元素边框的粗细. border-color:设置元素边框的 ...
UITableView中复用cell显示信息错乱
UITableView继承自UIScrollview,是苹果为我们封装好的一个基于scroll的控件.上面主要是一个个的 UITableViewCell,可以让UITableViewCell响应一些点 ...
Azure上A/D系列虚拟机到DS系列迁移(1)
有一些用户在刚开始的时候创建了A7,D14等虚拟机来跑一些IO要求比较高的应用,比如Oracle数据库,MySQL数据库,等到用户量上来之后,性能不够,需要升级数据磁盘到SSD固态硬盘,但是问题是: ...
RabbitMQ入门-Topic模式
上篇<RabbitMQ入门-Routing直连模式>我们介绍了可以定向发送消息,并可以根据自定义规则派发消息.看起来,这个Routing模式已经算灵活的了,但是,这还不够,我们还有更加多样 ...
Elastic Stack之FileBeat使用实战
Elastic Stack之FileBeat使用实战作者:尹正杰版权声明:原创作品,谢绝转载!否则将追究法律责任. 本篇博客数据流走向:FileBeat ===>logstash == ...
VirtualBox 导入虚拟机时的注意事项 VDI 与VMDK
1. 建议不要勾选 as vdi vmdk 最好不过了.. 长个经验教训以后尽量不勾选 vmdk 有很多工具能进行转换 vdi的要麻烦一些.
iOS中Date和NString的相互转换
必须知道的内容 G: 公元时代,例如AD公元 yy: 年的后2位 yyyy: 完整年 MM: 月,显示为1-12 MMM: 月,显示为英文月份简写,如 Jan ...