使用maxwell实时同步mysql数据到kafka

一、软件环境：

操作系统：CentOS release 6.5 (Final)

java版本: jdk1.8

zookeeper版本: zookeeper-3.4.11

kafka 版本: kafka_2.11-1.1.0.tgz

maxwell版本：maxwell-1.16.0.tar.gz

注意 ： 关闭所有机器的防火墙，同时注意启动可以相互telnet ip 端口

二、环境部署

1、安装jdk

export JAVA_HOME=/usr/java/jdk1.8.0_181

export PATH=$JAVA_HOME/bin:$JAVA_HOME/jre/bin:$PATH

export CLASSPATH=.:$JAVA_HOME/lib:$JAVA_HOME/jre/lib:$CLASSPATH

2、安装maven

参考：https://www.cnblogs.com/wcwen1990/p/7227278.html

3、安装zookeeper

1）下载软件：

wget http://101.96.8.157/archive.apache.org/dist/zookeeper/zookeeper-3.4.11/zookeeper-3.4.11.tar.gz

tar zxvf zookeeper-3.4.11.tar.gz

mv zookeeper-3.4.11 /usr/local/zookeeper

2）修改环境变量

编辑 /etc/profile 文件, 在文件末尾添加以下环境变量配置:

# ZooKeeper Env

export ZOOKEEPER_HOME=/usr/local/zookeeper

export PATH=$PATH:$ZOOKEEPER_HOME/bin

运行以下命令使环境变量生效: source /etc/profile

3）重命名配置文件

初次使用 ZooKeeper 时,需要将$ZOOKEEPER_HOME/conf 目录下的 zoo_sample.cfg 重命名为 zoo.cfg

mv  $ZOOKEEPER_HOME/conf/zoo_sample.cfg $ZOOKEEPER_HOME/conf/zoo.cfg

4）单机模式--修改配置文件

创建目录/usr/local/zookeeper/data 和/usr/local/zookeeper/logs 修改配置文件

tickTime=2000

initLimit=10

syncLimit=5

dataDir=/usr/local/zookeeper/data

dataLogDir=/usr/local/zookeeper/logs

clientPort=2181

5）启动zookeeper

# bin/zkServer.sh start

ZooKeeper JMX enabled by default

Using config: /usr/local/zookeeper/bin/../conf/zoo.cfg

Starting zookeeper ... STARTED

6）验证zukeeper服务

# telnet chavin.king 2181

Trying 192.168.72.130...

Connected to chavin.king.

Escape character is ‘^]‘.

stat

Zookeeper version: 3.4.11-37e277162d567b55a07d1755f0b31c32e93c01a0, built on 11/01/2017 18:06 GMT

Clients:
  /192.168.72.130:44054[0](queued=0,recved=1,sent=0)

Latency min/avg/max: 0/0/0

Received: 1

Sent: 0

Connections: 1

Outstanding: 0

Zxid: 0x1a4

Mode: standalone

Node count: 147

Connection closed by foreign host.

4、安装zkui

git clone https://github.com/DeemOpen/zkui.git

cd zkui

mvn clean install

修改配置文件默认值

#vim config.cfg
     serverPort=9090     #指定端口
     zkServer=192.168.1.110:2181
     sessionTimeout=300000

启动程序至后台

2.0-SNAPSHOT 会随软件的更新版本不同而不同，执行时请查看target 目录中真正生成的版本

nohup java -jar target/zkui-2.0-SNAPSHOT-jar-with-dependencies.jar &

用浏览器访问：

http://chavin.king:9090/

5、安装kafka

wget http://archive.apache.org/dist/kafka/1.1.0/kafka_2.11-1.1.0.tgz

tar -zxvf kafka_2.11-1.1.0.tgz -C /usr/local/kafka

mkdir -p /usr/local/kafka/data-logs

修改配置文件

vim server.properties

log.dirs=/usr/local/kafka/data-logs

zookeeper.connect=chavin.king:2181

启动kafka

bin/kafka-server-start.sh -daemon config/server.properties &

创建topic

bin/kafka-topics.sh --create --zookeeper chavin.king:2181 --replication-factor 1 --partitions 1 --topic maxwell

查看所有topic

bin/kafka-topics.sh --list --zookeeper chavin.king:2181

启动producer

bin/kafka-console-producer.sh --broker-list chavin.king:9092 --topic maxwell

启动consumer

bin/kafka-console-consumer.sh --zookeeper chavin.king:2181 --topic maxwell --from-beginning

或者

bin/kafka-console-consumer.sh --bootstrap-server chavin.king:9092  --from-beginning --topic maxwell

6、开启mysql binlog

more /etc/my.cnf

[client]

default_character_set = utf8

[mysqld]

basedir = /usr/local/mysql-5.6.24

datadir = /usr/local/mysql-5.6.24/data

port = 3306

#skip-grant-tables

character_set_server = utf8

log_error = /usr/local/mysql-5.6.24/data/mysql.err

binlog_format = row

log-bin = /usr/local/mysql-5.6.24/logs/mysql-bin

sync_binlog = 2

max_binlog_size = 16M

expire_logs_days = 10

server_id = 1

sql_mode=NO_ENGINE_SUBSTITUTION,STRICT_TRANS_TABLES

7、安装maxwell

wget https://github.com/zendesk/maxwell/releases/download/v1.16.0/maxwell-1.16.0.tar.gz

tar -zxvf maxwell-1.16.0.tar.gz -C /usr/local/maxwell

启动maxwell

nohup bin/maxwell --user=‘canal‘ --password=‘canal‘ --host=‘chavin.king‘ --producer=kafka --kafka.bootstrap.servers=chavin.king:9092 > maxwell.log &

8、开发kafka消费程序

import org.apache.kafka.clients.consumer.ConsumerRecord;

import org.apache.kafka.clients.consumer.ConsumerRecords;

import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Arrays;

import java.util.Properties;

public class KafkaTest {

public static void main(String[] args){

String topicName = "maxwell";
         String groupID = "example-group";

Properties props = new Properties();
         props.put("bootstrap.servers","192.168.72.130:9092");
         props.put("group.id",groupID);
         props.put("auto.offset.reset","earliest");
         props.put("serializer.encoding","utf-8");
         props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
         props.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer<String,String > consumer = new KafkaConsumer<String, String>(props);
        
         consumer.subscribe(Arrays.asList(topicName));

try{
             while(true){
                 ConsumerRecords<String,String> records = consumer.poll(1000);
             for (ConsumerRecord<String, String> record : records)
                 System.out.printf("offset = %d, key = %s, value = %s\n",
                         record.offset(), record.key(), record.value());
             }
         }finally{
             consumer.close();
         }
     }

}

ideal启动以上消费程序

9、测试

offset = 3428, key = {"database":"chavin","table":"dept","_uuid":"0b195622-e7c7-4cf6-8203-5576752f9024"}, value = {"database":"chavin","table":"dept","type":"insert","ts":1499944326,"xid":121,"xoffset":2276,"data":{"deptno":10,"dname":"ACCOUNTING","loc":"NEW YORK"}}

offset = 3429, key = {"database":"chavin","table":"dept","_uuid":"333b98e3-a597-47fc-95ad-6e59ee0dadf6"}, value = {"database":"chavin","table":"dept","type":"insert","ts":1499944326,"xid":121,"xoffset":2277,"data":{"deptno":20,"dname":"RESEARCH","loc":"DALLAS"}}

offset = 3430, key = {"database":"chavin","table":"dept","_uuid":"cf9fa656-ed13-4cb0-b909-d1218e402e96"}, value = {"database":"chavin","table":"dept","type":"insert","ts":1499944326,"xid":121,"xoffset":2278,"data":{"deptno":30,"dname":"SALES","loc":"CHICAGO"}}

offset = 3431, key = {"database":"chavin","table":"dept","_uuid":"7f2f683a-39bc-498b-9a4e-920697b3da18"}, value = {"database":"chavin","table":"dept","type":"insert","ts":1499944326,"xid":121,"xoffset":2279,"data":{"deptno":40,"dname":"OPERATIONS","loc":"BOSTON"}}

offset = 3432, key = {"database":"chavin","table":"dept","_uuid":"ef639cd1-9206-4145-8608-372bbaaaa14a"}, value = {"database":"chavin","table":"dept","type":"insert","ts":1499944326,"xid":121,"xoffset":2280,"data":{"deptno":10,"dname":"ACCOUNTING","loc":"NEW YORK"}}

offset = 3433, key = {"database":"chavin","table":"dept","_uuid":"ebdf15ad-7149-4ac4-b567-627dd910182c"}, value = {"database":"chavin","table":"dept","type":"insert","ts":1499944326,"xid":121,"xoffset":2281,"data":{"deptno":20,"dname":"RESEARCH","loc":"DALLAS"}}

offset = 3434, key = {"database":"chavin","table":"dept","_uuid":"1bc667f4-15f0-438c-8139-6f1cbe8b4db3"}, value = {"database":"chavin","table":"dept","type":"insert","ts":1499944326,"xid":121,"xoffset":2282,"data":{"deptno":30,"dname":"SALES","loc":"CHICAGO"}}

offset = 3435, key = {"database":"chavin","table":"dept","_uuid":"1613b695-284a-49e3-9793-74fb2cf8dc5b"}, value = {"database":"chavin","table":"dept","type":"insert","ts":1499944326,"xid":121,"xoffset":2283,"data":{"deptno":40,"dname":"OPERATIONS","loc":"BOSTON"}}

offset = 3436, key = {"database":"chavin","table":"dept","_uuid":"f72a800c-92cc-4494-9438-bc61c58b5cb9"}, value = {"database":"chavin","table":"dept","type":"insert","ts":1499944326,"xid":121,"xoffset":2284,"data":{"deptno":10,"dname":"ACCOUNTING","loc":"NEW YORK"}}

offset = 3437, key = {"database":"chavin","table":"dept","_uuid":"9887d144-d75d-46f8-96ba-ad7c3adf45fd"}, value = {"database":"chavin","table":"dept","type":"insert","ts":1499944326,"xid":121,"xoffset":2285,"data":{"deptno":20,"dname":"RESEARCH","loc":"DALLAS"}}

至此数据同步已经可以正常进行了，是不是很简单。

原文地址：https://www.cnblogs.com/wcwen1990/p/9817906.html

时间： 2024-11-03 21:53:42

使用maxwell实时同步mysql数据到kafka的相关文章

solr5.5.4 增量索引 自动同步mysql数据

前言:之前测试了solr好几个版本都没有成功,solr比较好下载的版本5.0,5.5,6.6.solr6.6结合tomcat8做自动同步mysql数据一直报错,然后果断测试5.5版本的,成功了. 环境:windows10, jdk1.8,tomcat8,solr5.5.4 实现功能:通过前端输入关键字查询,同时查询多表多字段.当数据库mysql添加或者更新数据的时候,数据自动更新到solr引擎中. 正文: 第一步:下载solr 1.solr5.5下载:http://apache.fayea.co

logstash增量同步mysql数据到es

本篇本章地址:https://www.cnblogs.com/Thehorse/p/11601013.html 今天我们来讲一下logstash同步mysql数据到es 我认为呢,logstash是众多同步mysql数据到es的插件中,最稳定并且最容易配置的一个. input { stdin { } jdbc { type => "xxx" jdbc_connection_string => "jdbc:mysql://127.0.0.1:3306/yinhela

logstash同步mysql数据到Elasticsearch

安装logstash查看我的另一篇文章  Docker 部署 logstash 同步数据我们首先需要安装好对应的插件,然后下载对应的数据库链接jar包,下面是具体的步骤 1.进入容器中 docker exec it logstash bash 2.进入到bin 目录下,我这里是/usr/share/logstash/bin,可以看到logstash-plugin文件,然后安装插件 logstash-plugin install logstash-input-jdbc 3.看到如下输出,则表示安装

Elasticsearch-jdbc批量同步mysql数据失败

最近公司系统中全模糊查询很多,数据量又大,多表连接查询时会很影响性能.于是考虑使用搜索引擎来做全模糊查询,思路: mysql数据库数据同步至ES类型,同步采用全量同步+定时增量方式,应用直接从ES中去查询想要的结果. 通过一番查找,决定使用elasticsearch-jdbc进行数据的同步,五六张表的连接结果,在数据量小的开发与测试环境运行正常,但在数据量比较大的性能测试环境做数据同步的话就会出现问题,以下为同步时报的一些错,github上也未找到相关有用的东西.群里问也都没人这样使用. 一种报

InfoSphere CDC 实时同步本地数据到云端的BigInsights

IBM InfoSphere CDC 是一款功能强大的数据实时复制软件,不仅广泛应用于传统ODS.数据仓库.数据集市及BI系统的异构平台集成,同时也提供对Cloud的全面支持:对于各种 Cloud 应用场景,CDC 不仅能提供低影响.近实时的海量数据复制,同时还可确保数据在传输过程中的完整性和安全性. 作为IBM重金打造的旗舰品牌,Bluemix 公有云平台是一个基于 Cloud Foundry 开源项目的平台即服务(PaaS)产品,它使得组织和开发人员能够快速而又轻松地在云上创建.部署和管理应

WebSocket 实时更新mysql数据到页面

使用websocket的初衷是,要实时更新mysql中的报警信息到web页面显示 没怎么碰过web,代码写的是真烂,不过也算是功能实现了,放在这里也是鞭策自己,web也要多下些功夫 准备 引入依赖 <dependency> <groupId>mysql</groupId> <artifactId>mysql-connector-java</artifactId> <version>5.1.38</version> <

【记录】ELK之logstash同步mysql数据到Elasticsearch ，配置文件详解

本文出处:https://my.oschina.net/xiaowangqiongyou/blog/1812708#comments 截取部分内容以便学习 input { jdbc { # mysql相关jdbc配置 jdbc_connection_string => "jdbc:mysql://10.112.76.30:3306/jack_test?useUnicode=true&characterEncoding=utf-8&useSSL=false" jdb

10.Solr4.10.3数据导入(DIH全量增量同步Mysql数据)

转载请出自出处:http://www.cnblogs.com/hd3013779515/ 1.创建MySQL数据 create database solr; use solr; DROP TABLE IF EXISTS student; CREATE TABLE student ( id char(10) NOT NULL, stu_name varchar(50) DEFAULT NULL, stu_sex int(1) DEFAULT NULL, stu_address varchar(20

sqoop同步mysql数据到hive中

一.sqoop 在同步mysql表结构到hive sqoop create-hive-table --connect jdbc:mysql://ip:3306/sampledata --table t1--username dev --password 1234 --hive-table t1; 执行到这一步就退出了,但是在hadoop的hdfs上的/hive/warehouse/的目录下是找不到t1表的目录, 但是正常执行完成是下面这样的: 错误就是hive的jar包有缺失 全部的jar包该是