mysql增量同步到greenplum
采用工具:maxwell+kafka+bireme
maxwell:maxwell能实时解析mysql的binlog,并输出json格式的数据发送到kafka(还支持其它的消息中间件),具体参见:
kafka: 一种消息中间件,在该方案中主要用于消息中转,具体参见kafka官网
bireme:支持greenplum的数据增量同步工具,在写入greenplum的过程中,由于采用copy模式,所以性能较高,具体参见bireme官网
大致原理就是:利用maxwell把mysql binlog解析成json,然后用kafka创建topic,然后用bireme消费,从而达到增量,增量的前提是先把数据全量同步一次,然后再增量。
全量同步初始化个人推荐dbswitch工具,个人测试使用体验最佳,可以自动创建表结构,同步速度也很快。
操作步骤:
1.下载并搭建kafka服务
2.下载并搭建maxwell服务,修改配置使其能够连接mysql并能向kafka写入数据
3.下载并搭建bireme服务,修改配置使其能读取kafka的数据并能向greenplum写入数据
kafka:
(1)下载安装:
wget http://mirrors.hust.edu.cn/apache/kafka/2.5.0/kafka_2.12-2.5.0.tgz
tar -xzf kafka_2.12-2.5.0.tgz -c /usr/local
(2)配置server.properties,我的简单配置如下:
[root@szwpldb1080 config]# cat server.properties |grep -ve '^#|^$'
broker.id=0
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/tmp/kafka-logs
num.partitions=1
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=16
log.cleanup.policy=delete
log.segment.bytes=1073741824
log.retention.check.interval.ms=3000
delete.topic.enable = true
zookeeper.connect=localhost:2181
zookeeper.connection.timeout.ms=6000
delete.topic.enable=true
group.initial.rebalance.delay.ms=0
advertised.host.name=172.18.1.150
(3)制作kafka启停脚本(提前安装好java):
#!/usr/bin/env bash # chkconfig: 2345 20 80 #description: start and stop server zoop_home=/usr/local/kafka_2.12-2.5.0/bin java_home=/usr/java/jdk1.8.0_221/ path=$java_home/bin:$path classpath=.:$java_home/lib/dt.jar:$java_home/lib/tools.jar
svr_name=kafka case $1 in start) echo "starting $zoop_home/$svr_name ..." $zoop_home/kafka-server-start.sh /usr/local/kafka_2.12-2.5.0/config/server.properties > /tmp/kafka.logs.out& ;; stop) echo "stopping $pro_home/$svr_name ..." ps -ef|grep *.$svr_name* |grep -v grep |awk '{print $2}' | sed -e "s/^/kill -9 /g" | sh - ;; restart) "$0" stop sleep 3 "$0" start ;; status) ps -ef|grep *.$svr_name* ;; logs) tail -f /tmp/zookeeper.logs.out ;; *) echo "example: server-$svr_name [start|stop|restart|status|logs]" ;; esac
添加到 /etc/rc.d/init.d ,然后就可以直接service kafka xxx来管理,或者添加到systemd下面,确保服务正常启动安装zookerper,然后启动kafka。
我的zookeeper简单配置如下:
[root@szwpldb1080 config]# cat zookeeper.properties |grep -ev '^$|^#' datadir=/tmp/zookeeper clientport=2181 maxclientcnxns=0
zookeeper启停脚本
#!/usr/bin/env bash # chkconfig: 2345 20 80 #description: start and stop server zoop_home=/usr/local/kafka_2.12-2.5.0/bin java_home=/usr/java/jdk1.8.0_221/ path=$java_home/bin:$path classpath=.:$java_home/lib/dt.jar:$java_home/lib/tools.jar svr_name=zookeeper case $1 in start) echo "starting $zoop_home/$svr_name ..." $zoop_home/zookeeper-server-start.sh /usr/local/kafka_2.12-2.5.0/config/zookeeper.properties > /tmp/zookeeper.logs.out& ;; stop) echo "stopping $pro_home/$svr_name ..." ps -ef|grep *.$svr_name* |grep -v grep |awk '{print $2}' | sed -e "s/^/kill -9 /g" | sh - ;; restart) "$0" stop sleep 3 "$0" start ;; status) ps -ef|grep *.$svr_name* ;; logs) tail -f /tmp/zookeeper.logs.out ;; *) echo "example: server-$svr_name [start|stop|restart|status|logs]" ;; esac
(4)检查状态:
[root@szwpldb1080 config]# jps 1762 kafka 18521 quorumpeermain 30383 jps
maxwell:
(1)docker镜像下载
docker pull zendesk/maxwell
(2)在源端mysql建好用户设置好权限,测试maxwell:
docker run -ti --rm zendesk/maxwell bin/maxwell --user='xxxx' --password='xxxx' --host='x.x.x.x' --producer=stdout
看到日志输出正常,可以放后台运行:
docker run -d --rm zendesk/maxwell bin/maxwell --user='xx' \ --password='xx' --host='x.x.x.x' --port=3306\ --producer=kafka --kafka.bootstrap.servers='x.x.x.x:9092'\ --kafka_topic=syncdb --log_level=debug --output_ddl
然后创建kafka topic:
bin/kafka-topics.sh --create --topic syncdb --zookeeper localhost:2181 --partitions 1 --replication-factor 2
topic名字与maxwell创建的topic一致,并且由于maxwell可以解析所有binlog,但是bireme工具只能同步dml,因此没有加 --output_ddl
具体maxwell参数用法可以参考:
检查kafka消费情况:
[root@szwpldb1080 bin]# ./kafka-topics.sh --zookeeper localhost:2181 --describe --topic syncdb topic:syncdb partitioncount:1 replicationfactor:1 configs: topic: syncdb partition: 0 leader: 0 replicas: 0 isr: 0 [root@szwpldb1080 bin]#./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic syncdb --from-beginning
看到kafka能接收maxwell产生的json文件,表示没问题。
bireme
安装配置都很简单,此处略过。
主要是修改以下2个配置文件。
config.properties maxwell1.properties
然后监控 http://x.x.x.x:8080或者监控bireme日志就可以了。
总结:
只能同步dml语句,无法处理ddl,对比几款开源的工具已经同步方式,我觉得我这种是最舒服的。
上一篇: 【Python初阶汇总】
下一篇: css 自定义滚动条样式
推荐阅读