1.maxwell基础
参考http://seanlook.com/2018/01/13/maxwell-binlog/

这里写下我的基础配置

2.python 消费者伪代码

import time, json
from pykafka import KafkaClient
client = KafkaClient(hosts="10.1.2.175:9092")
topic = client.topics['test']
balanced_consumer = topic.get_balanced_consumer(consumer_group='goods_group',auto_commit_enable=True,zookeeper_connect='localhost:2181')
for message in balanced_consumer:
print message.offset, message.value

启动
[root@ip-10-1-2-175 sh]# python cus.py

3.启动maxwel

4.模拟数据库操作

5.查看消费者

[root@ip-10-1-2-175 sh]# python cus.py
67 {"database":"test","table":"testwell","type":"insert","ts":1526290515,"xid":208818,"commit":true,"data":{"id":3050,"name":"fuck"}}
68 {"database":"test","table":"testwell","type":"insert","ts":1526290515,"xid":208819,"commit":true,"data":{"id":3051,"name":"fuck"}}
69 {"database":"test","table":"testwell","type":"insert","ts":1526290517,"xid":208820,"commit":true,"data":{"id":3052,"name":"fuck"}}
70 {"database":"test","table":"testwell","type":"insert","ts":1526290519,"xid":208822,"commit":true,"data":{"id":3053,"name":"fuck"}}

6.根据上面的思路可以写出增量同步的伪代码

[root@ip-10-1-2-175 sh]#
import time, json
from pykafka import KafkaClient
client = KafkaClient(hosts="10.1.2.175:9092")
topic = client.topics['test']
balanced_consumer = topic.get_balanced_consumer(consumer_group='goods_group',auto_commit_enable=True,zookeeper_connect='localhost:2181')
for message in balanced_consumer:
print message.offset, message.value
# 对数据库进行操作
if message.value ['type'] == 'insert':
mysql_insert()
elif message.value['type'] == 'update':
mysql_update()
elif message.value['type'] == 'delete':
mysql_delete()
else:
order_option()

ITPub博客

【Maxwell】使用maxwell+kafka+python做binlog增量解析消费