Hadoop实战：使用Maxwell把MySQL数据实时同步到HDFS

- 日理万妓 2022-02-04 23:19 535阅读 0赞

### Maxwell介绍 ###

Maxwell是一个守护程序，一个应用程序，能够读取MySQL Binlogs然后解析输出为json。支持数据输出到Kafka中，支持表和库过滤。

→ Reference：[http://maxwells-daemon.io][http_maxwells-daemon.io]

→ Download: [https://github.com/zendesk/maxwell/releases/download/v1.10.3/maxwell-1.10.3.tar.gz][https_github.com_zendesk_maxwell_releases_download_v1.10.3_maxwell-1.10.3.tar.gz]

→ Source: [https://github.com/zendesk/maxwell][https_github.com_zendesk_maxwell]

### 配置MySQL->Maxwell->Kafka->Flume->HDFS ###

1）MySQL配置要求

配置要求

<table> 
 <tbody> 
 <tr> 
 <td> 
 <hr>[mysqld] server-id=1 log-bin=master binlog_format=row binlog_row_image=FULL &nbsp; 
 <hr>&nbsp; </td> 
 </tr> 
 </tbody> 
</table>

权限要求

--------------------

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 </td> 
 <td> GRANT ALL on maxwell.* to 'maxwell'@'%' identified by 'maxwell'; GRANT ALL on maxwell.* to 'maxwell'@'localhost' identified by 'maxwell'; GRANT SELECT, REPLICATION CLIENT, REPLICATION SLAVE on *.* to 'maxwell'@'%'; GRANT SELECT, REPLICATION CLIENT, REPLICATION SLAVE on *.* to 'maxwell'@'localhost'; &nbsp; </td> 
 </tr> 
 </tbody> 
</table>

--------------------

2）安装配置Kafka

确认已安装java运行环境，直接解压Kafka即可使用。

<table> 
 <tbody> 
 <tr> 
 <td> 1 </td> 
 <td> $ tar xvf kafka_2.10-0.10.2.1.tgz -C /usr/local/elk </td> 
 </tr> 
 </tbody> 
</table>

解压后，编辑配置文件：

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 </td> 
 <td> $ cat /usr/local/elk/kafka_2.10-0.10.2.1/config/server.properties&nbsp;&nbsp;&nbsp;&nbsp; ############################# Server Basics ############################# broker.id=0 delete.topic.enable=true &nbsp; ############################# Socket Server Settings ############################# listeners=PLAINTEXT://0.0.0.0:9092 num.network.threads=3 num.io.threads=8 socket.send.buffer.bytes=102400 socket.receive.buffer.bytes=102400 socket.request.max.bytes=104857600 &nbsp; ############################# Log Basics ############################# log.dirs=/tmp/kafka-logs num.partitions=1 num.recovery.threads.per.data.dir=1 &nbsp; ############################# Log Flush Policy ############################# log.flush.interval.messages=10000 log.flush.interval.ms=1000 &nbsp; ############################# Log Retention Policy ############################# log.retention.hours=168 log.segment.bytes=1073741824 log.retention.check.interval.ms=300000 &nbsp; ############################# Zookeeper ############################# zookeeper.connect=localhost:2181 zookeeper.connection.timeout.ms=6000 </td> 
 </tr> 
 </tbody> 
</table>

kafka需要依赖zookeeper，所以需要先启动zookeeper。

<table> 
 <tbody> 
 <tr> 
 <td> 1 </td> 
 <td> $ nohup /usr/local/elk/kafka_2.10-0.10.2.1/bin/zookeeper-server-start.sh /usr/local/elk/kafka_2.10-0.10.2.1/config/zookeeper.properties &amp; </td> 
 </tr> 
 </tbody> 
</table>

启动Kafka Server：(指定JMX\_PORT端口，可以通过Kafka-manager获取统计信息)

<table> 
 <tbody> 
 <tr> 
 <td> 1 </td> 
 <td> $ nohup /usr/local/elk/kafka_2.10-0.10.2.1/bin/kafka-server-start.sh /usr/local/elk/kafka_2.10-0.10.2.1/config/server.properties &amp; </td> 
 </tr> 
 </tbody> 
</table>

3）安装配置Flume

去Apache官网下载Flume二进制安装包，然后解压即可。

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 </td> 
 <td> tar xvf apache-flume-1.7.0-bin.tar.gz -C /usr/local/ ln -sv /usr/local/apache-flume-1.7.0-bin/ /usr/local/flume </td> 
 </tr> 
 </tbody> 
</table>

设置环境变量

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 </td> 
 <td> $ cat /etc/profile.d/flume.sh export FLUME_HOME=/usr/local/flume export FLUME_CONF_DIR=$FLUME_HOME/conf export PATH=$PATH:$FLUME_HOME/bin </td> 
 </tr> 
 </tbody> 
</table>

查看Flume版本

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 </td> 
 <td> $ flume-ng version Flume 1.7.0 Source code repository: https://git-wip-us.apache.org/repos/asf/flume.git Revision: 511d868555dd4d16e6ce4fedc72c2d1454546707 Compiled by bessbd on Wed Oct 12 20:51:10 CEST 2016 </td> 
 </tr> 
 </tbody> 
</table>

创建配置文件和环境变量

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 </td> 
 <td> $ cp -fr /usr/local/flume/conf/flume-conf.properties.template /usr/local/flume/conf/flume.conf $ cp -fr /usr/local/flume/conf/flume-env.sh.template /usr/local/flume/conf/flume-env.sh </td> 
 </tr> 
 </tbody> 
</table>

如果上面的JAVA\_HOME设置好了，这里其实不需要设置flume-env.sh，也可以选择配置。

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 </td> 
 <td> $ cat /usr/local/flume/conf/flume-env.sh # Enviroment variables can be set here. export JAVA_HOME=/usr/lib/jvm/java-1.8.0-openjdk-1.8.0.131-3.b12.el7_3.x86_64/jre &nbsp; # Give Flume more memory and pre-allocate, enable remote monitoring via JMX export JAVA_OPTS="-Xms100m -Xmx2000m -Dcom.sun.management.jmxremote" &nbsp; # as it may result in logging sensitive user information or encryption secrets. export JAVA_OPTS="$JAVA_OPTS -Dorg.apache.flume.log.rawdata=true -Dorg.apache.flume.log.printconfig=true " &nbsp; # Note that the Flume conf directory is always included in the classpath. #FLUME_CLASSPATH="" </td> 
 </tr> 
 </tbody> 
</table>

4）安装配置Maxwell

Maxwell存储在MySQL服务器本身所需要的所有状态，在schema\_database选项指定的数据库中。默认情况下, 数据库被命名为maxwell。

<table> 
 <tbody> 
 <tr> 
 <td> 1 </td> 
 <td> $ cd /usr/local/maxwell/;./bin/maxwell --user='maxwell' --password='maxwell' --host='127.0.0.1' --port='3306' --producer=stdout </td> 
 </tr> 
 </tbody> 
</table>

MySQL创造点数据

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 </td> 
 <td> mysql&gt; create database hadoop charset utf8; Query OK, 1 row affected (0.02 sec) &nbsp; mysql&gt; use hadoop; Database changed mysql&gt; create table test(id int,name varchar(10),address varchar(20)); Query OK, 0 rows affected (0.00 sec) &nbsp; mysql&gt; insert into test values(1,'dkey','ShangHai'); Query OK, 1 row affected (0.01 sec) </td> 
 </tr> 
 </tbody> 
</table>

然后可以看到Maxwell的输出信息：

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 </td> 
 <td> 04:16:48,341 INFO OpenReplicator - starting replication at mysql-bin.000004:6777 04:18:18,654 INFO AbstractSchemaStore - storing schema @Position[BinlogPosition[mysql-bin.000004:136974], lastHeartbeat=1497601097500] after applying "create database hadoop charset utf8" to hadoop, new schema id is 2 04:20:24,430 INFO AbstractSchemaStore - storing schema @Position[BinlogPosition[mysql-bin.000004:255163], lastHeartbeat=1497601224355] after applying "create table test(id int,name varchar(10),address varchar(20))" to hadoop, new schema id is 3 {"database":"hadoop","table":"test","type":"insert","ts":1497601280,"xid":929,"commit":true,"data":{"id":1,"name":"dkey","address":"ShangHai"}} </td> 
 </tr> 
 </tbody> 
</table>

5）数据输出到HDFS

Kafka创建topic

<table> 
 <tbody> 
 <tr> 
 <td> 1 </td> 
 <td> $ /usr/local/kafka/bin/kafka-topics.sh --zookeeper 127.0.0.1:2181 --create --topic maxwell --partitions 20 --replication-factor 1 </td> 
 </tr> 
 </tbody> 
</table>

查看主题

<table> 
 <tbody> 
 <tr> 
 <td> 1 </td> 
 <td> $ /usr/local/kafka/bin/kafka-topics.sh --list --zookeeper=127.0.0.1:2181 maxwell </td> 
 </tr> 
 </tbody> 
</table>

查看主题详情

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 </td> 
 <td> $ /usr/local/kafka/bin/kafka-topics.sh --zookeeper=127.0.0.1:2181 --describe --topic maxwell Topic:maxwell PartitionCount:1 ReplicationFactor:1 Configs: Topic: maxwell Partition: 0 Leader: 0 Replicas: 0 Isr: 0 </td> 
 </tr> 
 </tbody> 
</table>

提供一份Flume配置文件（从Kafka收集日志到HDFS）

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 </td> 
 <td> $ cat /usr/local/flume/conf/mysql-flume-hdfs.conf a1.sources = r1 a1.sinks = k1 a1.channels = c1 &nbsp; # Describe/configure the source a1.sources.r1.type = org.apache.flume.source.kafka.KafkaSource a1.sources.r1.zookeeperConnect = 127.0.0.1:2181 a1.sources.r1.topic = maxwell a1.sources.r1.groupId = flume a1.sources.r1.channels = c1 a1.sources.r1.interceptors = i1 a1.sources.r1.interceptors.i1.type = timestamp a1.sources.r1.kafka.consumer.timeout.ms = 100 &nbsp; # Describe the sink a1.sinks.k1.type = hdfs #a1.sinks.k1.hdfs.path = /mysql/%{topic}/%y-%m-%d a1.sinks.k1.hdfs.path = hdfs://10.10.0.186:8020/mysql/%{topic}/%y-%m-%d a1.sinks.k1.hdfs.rollInterval = 5 a1.sinks.k1.hdfs.rollSize = 0 a1.sinks.k1.hdfs.rollCount = 0 a1.sinks.k1.hdfs.fileType = DataStream a1.sinks.k1.channel = c1 &nbsp; # Use a channel which buffers events in memory a1.channels.c1.type = memory a1.channels.c1.capacity = 10000 a1.channels.c1.transactionCapacity = 1000 &nbsp; # Bind the source and sink to the channel a1.sources.r1.channels = c1 a1.sinks.k1.channel = c1 </td> 
 </tr> 
 </tbody> 
</table>

启动Flume

<table> 
 <tbody> 
 <tr> 
 <td> 1 </td> 
 <td> $ nohup flume-ng agent --conf /usr/local/flume/conf --conf-file /usr/local/flume/conf/mysql-flume-hdfs.conf --name a1 -Dflume.root.logger=INFO,console &amp; </td> 
 </tr> 
 </tbody> 
</table>

如果启动Flume时报错：ERROR – org.apache.flume.node.PollingPropertiesFileConfigurationProvider$FileWatcherRunnable.run(PollingPropertiesFileConfigurationProvider.java:146)\] Failed to start agent because dependencies were not found in classpath. Error follows.java.lang.NoClassDefFoundError: org/apache/hadoop/io/SequenceFile$CompressionType.

可能是因为你的Flume是独立部署，需要依赖Hadoop HDFS的jar包，解决方法也很简单，就是在Flume主机上解压好Hadoop的二进制安装包，然后输出Hadoop环境变量即可，Flume会根据环境变量自动找到相关的依赖jar包。具体可以看：[Hadoop实战：Flume输入日志到HDFS报错解决][Hadoop_Flume_HDFS]

另外，当Flume-ng正常运行后，写入HDFS时报错：java.lang.NoClassDefFoundError: org/apache/hadoop/io/SequenceFile$CompressionType  
Caused by: org.apache.hadoop.ipc.RemoteException(org.apache.hadoop.security.AccessControlException): Permission denied: user=root, access=WRITE, inode=”/”:hadoop:supergroup:drwxr-xr-x.

这个提示很明显，就是没有写入权限（因为你当前运行flume-ng的用户不是Hadoop用户），解决方案也很简单，就是切换到Hadoop用户执行flume-ng命令即可。或者开启HDFS允许所有用户进行文件写入，默认可能你没有开启。具体可以看：[Hadoop实战：Flume输入日志到HDFS报错解决][Hadoop_Flume_HDFS]

启动Maxwell

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 </td> 
 <td> $ cd /usr/local/maxwell/;./bin/maxwell \ --user='maxwell' \ --password='maxwell' \ --host='127.0.0.1' \ --port='3306' \ --producer=kafka \ --kafka.bootstrap.servers=127.0.0.1:9092 </td> 
 </tr> 
 </tbody> 
</table>

### 测试MySQL->Maxwell->Kafka->Flume->HDFS ###

相关组件现在都已经跑通了，接下来就是测试了，我们在MySQL插入一条数据：

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 </td> 
 <td> mysql&gt; insert into hadoop.test values(5,'dkey5','Shanghai'); Query OK, 1 row affected (0.00 sec) </td> 
 </tr> 
 </tbody> 
</table>

查看Kafka队列

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 </td> 
 <td> $ /usr/local/kafka/bin/kafka-console-consumer.sh -zookeeper=127.0.0.1:2181 --from-beginning --topic maxwell Using the ConsoleConsumer with old consumer is deprecated and will be removed in a future major release. Consider using the new consumer by passing [bootstrap-server] instead of [zookeeper]. {"database":"hadoop","table":"test","type":"insert","ts":1497607783,"xid":2414,"commit":true,"data":{"id":5,"name":"dkey5","address":"Shanghai"}} </td> 
 </tr> 
 </tbody> 
</table>

然后去HDFS查看：

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 </td> 
 <td> [hadoop@hadoop-nn ~]$ hdfs dfs -ls /mysql/maxwell/17-06-19/ Found 1 items -rw-r--r--&nbsp;&nbsp; 3 hadoop supergroup&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;148 2017-06-19 03:57 /mysql/maxwell/17-06-19/FlumeData.1497859019506 </td> 
 </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 </td> 
 <td> [hadoop@hadoop-nn ~]$ hdfs dfs -cat /mysql/maxwell/17-06-19/FlumeData.1497859019506 {"database":"hadoop","table":"test","type":"insert","ts":1497859014,"xid":372064,"commit":true,"data":{"id":5,"name":"dkey5","address":"Shanghai"}} </td> 
 </tr> 
 </tbody> 
</table>

会自动创建相关目录，并生成一个文件。

总结

整个MySQL->Maxwell->Flume->HDFS流程算是跑通了，但是此时也仅限于玩一玩而已，包括Flume和Kakfa都得深入学习一下。另外，我们可以看到写入HDFS的数据时json的，可能还需要提取只需要的数据，另外，对于update或delete操作目前还不知道要怎么处理。生产使用难度很大。

[http_maxwells-daemon.io]: http://maxwells-daemon.io/
[https_github.com_zendesk_maxwell_releases_download_v1.10.3_maxwell-1.10.3.tar.gz]: https://github.com/zendesk/maxwell/releases/download/v1.10.3/maxwell-1.10.3.tar.gz
[https_github.com_zendesk_maxwell]: https://github.com/zendesk/maxwell
[Hadoop_Flume_HDFS]: http://www.ywnds.com/?p=10227