Maxwell同步mysql增量表数据到kafka，并消费到hd
1 数据通道
image.png
2 Maxwell配置
默认情况下，Maxwell会同步binlog中的所有表的数据变更记录，按照规划，有cart_info、order_info等共计11张表需进行增量同步，按理我们应对Maxwell进行配置，令其只同步这特定的11张表，但为了与实时数仓架构保持一致，此处不做相应配置，而令 Maxwell 对 binlog 中所有表的数据变更记录进行同步，并将数据全部发往 topic_db 主题。
Maxwell最终配置如下：
1）修改Maxwell配置文件config.properties
[yobhel@hadoop101 maxwell]$ vim /opt/module/maxwell/config.properties
2）全部配置参数如下
log_level=info
producer=kafka
kafka.bootstrap.servers=hadoop101:9092,hadoop101:9092,hadoop101:9092
#kafka topic配置，业务数据发往的目标主题
kafka_topic=topic_db
# mysql login info
host=hadoop101
user=maxwell
password=maxwell
jdbc_options=useSSL=false&serverTimezone=Asia/Shanghai
3）重新启动Maxwell
[yobhel@hadoop101 bin]$ mxw.sh restart
4）通道测试
（1）启动Zookeeper以及Kafka集群
（2）启动一个Kafka Console Consumer，消费 topic_db 主题的数据
[yobhel@hadoop101 kafka]$ kafka-console-consumer.sh --bootstrap-server hadoop101:9092 --topic topic_db
（3）生成模拟数据
[yobhel@hadoop101 bin]$ cd /opt/module/data_mocker/
[yobhel@hadoop101 data_mocker]$ java -jar edu2021-mock-2022-06-18.jar
（4）观察Kafka消费者是否能消费到数据
{"database":"edu2077","table":"order_info","type":"update","ts":1645425636,"xid":37606,"commit":true,"data":{"id":23899,"user_id":16,"origin_amount":800.00,"coupon_reduce":0.00,"final_amount":800.00,"order_status":"1002","out_trade_no":"211814417714292","trade_body":"大数据技术之Zookeeper（2021最新版）等4件商品","session_id":"3a96bddb-7f94-4a0f-9a5b-1aa6fadd718c","province_id":30,"create_time":"2022-02-21 15:15:14","expire_time":"2022-02-21 15:30:14","update_time":"2022-02-21 15:15:42"},"old":{"order_status":"1001","update_time":null}}
{"database":"edu2077","table":"order_info","type":"update","ts":1645425636,"xid":37589,"commit":true,"data":{"id":23900,"user_id":473,"origin_amount":200.00,"coupon_reduce":0.00,"final_amount":200.00,"order_status":"1003","out_trade_no":"462573352988853","trade_body":"尚硅谷大数据技术之Azkaban等1件商品","session_id":"d78dd675-5a38-4e33-b431-b1ef68a89089","province_id":29,"create_time":"2022-02-21 11:26:30","expire_time":"2022-02-21 11:41:30","update_time":"2022-02-21 11:41:47"},"old":{"order_status":"1001","update_time":null}}
{"database":"edu2077","table":"order_info","type":"update","ts":1645425636,"xid":37694,"commit":true,"data":{"id":23901,"user_id":70,"origin_amount":400.00,"coupon_reduce":0.00,"final_amount":400.00,"order_status":"1002","out_trade_no":"677577676596486","trade_body":"尚硅谷大数据技术之Shell等2件商品","session_id":"9b842bcc-3288-49da-8ec2-0e00d743b783","province_id":33,"create_time":"2022-02-21 19:45:13","expire_time":"2022-02-21 20:00:13","update_time":"2022-02-21 19:45:33"},"old":{"order_status":"1001","update_time":null}}
3 Flume配置
1）Flume配置概述
Flume需要将Kafka中各topic的数据传输到HDFS，故其需选用KafkaSource以及HDFSSink，Channe选用FileChanne。
需要注意的是，KafkaSource需订阅Kafka中的11个topic，HDFSSink需要将不同topic的数据写到不同的路径，并且路径中应当包含一层日期，用于区分每天的数据。关键配置如下：
image.png
具体数据示例如下：
image.png
2）Flume配置实操
（1）创建Flume配置文件
在hadoop103节点的Flume的job目录下创建kafka_to_hdfs_db.conf
[yobhel@hadoop103 job]$ vim kafka_to_hdfs_db.conf
（2）配置文件内容如下
a1.sources = r1
a1.channels = c1
a1.sinks = k1
a1.sources.r1.type = org.apache.flume.source.kafka.KafkaSource
a1.sources.r1.batchSize = 5000
a1.sources.r1.batchDurationMillis = 2000
a1.sources.r1.kafka.bootstrap.servers = hadoop101:9092,hadoop102:9092
a1.sources.r1.kafka.topics = topic_db
a1.sources.r1.kafka.consumer.group.id = flume
a1.sources.r1.setTopicHeader = true
a1.sources.r1.topicHeader = topic
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = com.yobhel.flume.interceptors.TimestampAndTableNameInterceptor$Builder
a1.channels.c1.type = file
a1.channels.c1.checkpointDir = /opt/data/flume/checkpoint/behavior2
a1.channels.c1.dataDirs = /opt/data/flume/data/behavior2
a1.channels.c1.maxFileSize = 2146435071
a1.channels.c1.capacity = 1000000
a1.channels.c1.keep-alive = 6
## sink1
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = /origin_data/edu/db/%{tableName}_inc/%Y-%m-%d
a1.sinks.k1.hdfs.filePrefix = db
a1.sinks.k1.hdfs.round = false
a1.sinks.k1.hdfs.rollInterval = 10
a1.sinks.k1.hdfs.rollSize = 134217728
a1.sinks.k1.hdfs.rollCount = 0
a1.sinks.k1.hdfs.fileType = CompressedStream
a1.sinks.k1.hdfs.codeC = gzip
## 拼装
a1.sources.r1.channels = c1
a1.sinks.k1.channel= c1
（3）编写Flume拦截器
代码：https://github.com/Yobhel121/edu-flume-interceptor
将打好的包放入到hadoop103的/opt/module/flume/lib文件夹下
3）通道测试
（1）启动Zookeeper、Kafka集群
（2）启动hadoop103的Flume
[yobhel@hadoop103 flume]$ bin/flume-ng agent -n a1 -c conf/ -f job/kafka_to_hdfs_db.conf -Dflume.root.logger=INFO,console
（3）反注释mock.sh中关于Maxwell的内容, 执行脚本生成模拟数据
#!/bin/bash
DATA_HOME=/opt/module/data_mocker
MAXWELL_HOME=/opt/module/maxwell
function mock_data() {
if [ $1 ]
then
sed -i "/mock.date/s/.*/mock.date: \"$1\"/" $DATA_HOME/application.yml
echo "正在生成 $1 当日的数据"
fi
cd $DATA_HOME
nohup java -jar "edu2021-mock-2022-03-14.jar" >/dev/null 2>&1
}
case $1 in
"init")
[ $2 ] && do_date=$2 || do_date='2022-02-21'
sed -i "/mock.clear.busi/s/.*/mock.clear.busi: 1/" $DATA_HOME/application.yml
sed -i "/mock.clear.user/s/.*/mock.clear.user: 1/" $DATA_HOME/application.yml
mock_data $(date -d "$do_date -5 days" +%F)
sed -i "/mock.clear.busi/s/.*/mock.clear.busi: 0/" $DATA_HOME/application.yml
sed -i "/mock.clear.user/s/.*/mock.clear.user: 0/" $DATA_HOME/application.yml
for ((i=4;i>=0;i--));
do
mock_data $(date -d "$do_date -$i days" +%F)
done
;;
[0-9][0-9][0-9][0-9]-[0-1][0-9]-[0-3][0-9])
sed -i "/mock_date/s/.*/mock_date=$1/" $MAXWELL_HOME/config.properties
mxw.sh restart
sleep 1
mock_data $1
;;
esac
执行脚本生成数据
[yobhel@hadoop101 bin]$ mock.sh 2022-02-22
（4）观察HDFS上的目标路径是否有数据出现
若HDFS上的目标路径已有增量表的数据出现了，就证明数据通道已经打通。
（5）数据目标路径的日期说明
仔细观察，会发现目标路径中的日期，并非模拟数据的业务日期，而是当前日期。这是由于Maxwell输出的JSON字符串中的ts字段的值，是数据的变动日期。而真实场景下，数据的业务日期与变动日期应当是一致的。
此处为了模拟真实环境，对Maxwell源码进行了改动，增加了一个参数mock_date，该参数的作用就是指定Maxwell输出JSON字符串的ts时间戳的日期，接下来进行测试。
mock.sh脚本在生成数据时会修改Maxwell的配置信息
4）编写Flume启停脚本
为方便使用，此处编写一个Flume的启停脚本
（1）在hadoop101节点的/home/yobhel/bin目录下创建脚本f3.sh
[yobhel@hadoop101 bin]$ vim f3.sh
在脚本中填写如下内容
#!/bin/bash
case $1 in
"start")
echo " --------启动 hadoop103 业务数据flume-------"
ssh hadoop103 "nohup /opt/module/flume/bin/flume-ng agent -n a1 -c /opt/module/flume/conf -f /opt/module/flume/job/kafka_to_hdfs_db.conf >/dev/null 2>&1 &"
;;
"stop")
echo " --------停止 hadoop103 业务数据flume-------"
ssh hadoop103 "ps -ef | grep kafka_to_hdfs_db | grep -v grep |awk '{print \$2}' | xargs -n1 kill"
;;
esac
（2）增加脚本执行权限
[yobhel@hadoop101 bin]$ chmod +x f3.sh
（3）f3启动
[yobhel@hadoop101 module]$ f3.sh start
（4）f3停止
[yobhel@hadoop101 module]$ f3.sh stop
4 增量表首日全量同步
通常情况下，增量表需要在首日进行一次全量同步，后续每日再进行增量同步，首日全量同步可以使用Maxwell的bootstrap功能，方便起见，下面编写一个增量表首日全量同步脚本。
1）在~/bin目录创建mysql_to_kafka_inc_init.sh
[yobhel@hadoop101 bin]$ vim mysql_to_kafka_inc_init.sh
脚本内容如下
#!/bin/bash
# 该脚本的作用是初始化所有的增量表，只需执行一次
MAXWELL_HOME=/opt/module/maxwell
import_data() {
$MAXWELL_HOME/bin/maxwell-bootstrap --database edu2077 --table $1 --config $MAXWELL_HOME/config.properties
}
case $1 in
cart_info | comment_info | favor_info | order_detail | order_info | payment_info | review_info | test_exam | test_exam_question | user_info | vip_change_detail)
import_data $1
;;
"all")
for tmp in cart_info comment_info favor_info order_detail order_info payment_info review_info test_exam test_exam_question user_info vip_change_detail
do
import_data $tmp
done
;;
esac
2）为mysql_to_kafka_inc_init.sh all增加执行权限
[yobhel@hadoop101 bin]$ chmod +x ~/bin/mysql_to_kafka_inc_init.sh
3）测试同步脚本
（1）清理历史数据
为方便查看结果，现将HDFS上之前同步的增量表数据删除
[yobhel@hadoop101 ~]$ hadoop fs -ls /origin_data/edu/db | grep _inc | awk '{print $8}' | xargs hadoop fs -rm -r -f
（2）执行同步脚本
[yobhel@hadoop101 bin]$ mysql_to_kafka_inc_init.sh all
4）检查同步结果
观察HDFS上是否重新出现增量表数据。
5 增量表同步总结
增量表同步，需要在首日进行一次全量同步，后续每日才是增量同步。首日进行全量同步时，需先启动数据通道，包括Maxwell、Kafka、Flume，然后执行增量表首日同步脚本mysql_to_kafka_inc_init.sh进行同步。后续每日只需保证采集通道正常运行即可，Maxwell便会实时将变动数据发往Kafka。
