【采集项目-（4)业务数据采集】

[atguigu@hadoop102 module]$ mkdir db_log/

2）把g mall2020-mock–db-2021-11-14.jar和application.properties 上传到hadoop102的/opt/module/db_log 路径上。
3)创建数据库，建表

# 创建gmall数据库并初始化
create database gmall charset utf8
# 在gmall数据库执行脚本
source /opt/module/db_log/gmall.sql

4）编写生成日志脚本

#!/bin/bash
# 初始化日志
# 生成指定日期日志
DB_HOME='/opt/module/db_log'
DT="$1"
INIT="$2"
sed -i "/mock.clear=/s/.*/mock.clear=0/" $DB_HOME/application.properties
sed -i "/mock.clear.user=/s/.*/mock.clear.user=0/" $DB_HOME/application.properties

if [ "$DT" ]
then 
  sed -i "/mock.date=/s/.*/mock.date=$DT/" $DB_HOME/application.properties
else
  DT=$(cat $DB_HOME/application.properties | grep 'mock.date' | awk -F '=' '{print $2}')
fi
if [ "$INIT" = 'init' ]
then
  sed -i "/mock.clear=/s/.*/mock.clear=1/" $DB_HOME/application.properties
  sed -i "/mock.clear.user=/s/.*/mock.clear.user=1/" $DB_HOME/application.properties
fi

echo "正在生成$DT的数据"
cd $DB_HOME
java -jar gmall2020-mock-db-2021-11-14.jar 1&gt;/dev/null 2&gt;&amp;1

5）生成数据

# 生成数据
[atguigu@hadoop102 db_log]$ chmod +x db_mock.sh 
[atguigu@hadoop102 db_log]$ ll
总用量 15292
-rw-r--r--. 1 atguigu atguigu     1790 9月  18 09:10 application.properties
-rwxr-xr-x. 1 atguigu atguigu      749 9月  18 11:05 db_mock.sh
-rw-r--r--. 1 atguigu atguigu 15648406 2月   8 2022 gmall2020-mock-db-2021-11-14.jar
[atguigu@hadoop102 db_log]$ ./db_mock.sh 2020-06-10 init
正在生成2020-06-10的数据
[atguigu@hadoop102 db_log]$ ./db_mock.sh 2020-06-11
正在生成2020-06-11的数据
[atguigu@hadoop102 db_log]$ ./db_mock.sh 2020-06-12
正在生成2020-06-12的数据
[atguigu@hadoop102 db_log]$ ./db_mock.sh 2020-06-13
正在生成2020-06-13的数据
[atguigu@hadoop102 db_log]$ ./db_mock.sh 2020-06-14
正在生成2020-06-14的数据
[atguigu@hadoop102 db_log]$ ./db_mock.sh 2020-06-15
正在生成2020-06-15的数据
[atguigu@hadoop102 db_log]$ ./db_mock.sh 2020-06-16
正在生成2020-06-16的数据
[atguigu@hadoop102 db_log]$ ./db_mock.sh 2020-06-17
正在生成2020-06-17的数据
[atguigu@hadoop102 db_log]$

安装 maxwell

官网地址：http://maxwells-daemon.io/
1）下载安装包
（1）地址：https://github.com/zendesk/maxwell/releases/download/v1.29.2/maxwell-1.29.2.tar.gz
注：Maxwell-1.30.0及以上版本不再支持JDK1.8。
（2）将安装包上传到h adoop102节点的/opt/software 目录
2）将安装包解压至/opt/mod ule

[atguigu@hadoop102 maxwell]$ tar -zxvf maxwell-1.29.2.tar.gz -C /opt/module/

3）修改名称

[atguigu@hadoop102 module]$ mv maxwell-1.29.2/ maxwell

配置MySQL

1.启用MySQL Binlog

1）修改MySQL配置文件/etc/my.cnf
[atguigu@hadoop102 ~]$ sudo vim /etc/my.cnf
2）增加如下配置

[mysqld]

#数据库id
server-id = 1
#启动binlog，该参数的值会作为binlog的文件名
log-bin=mysql-bin
#binlog类型，maxwell要求为row类型
binlog_format=row
#启用binlog的数据库，需根据实际情况作出修改
binlog-do-db=gmall

注：MySQL Binlog模式
Statement-based：基于语句，Binlog会记录所有写操作的SQL语句，包括insert、update、delete等。
优点：节省空间
缺点：有可能造成数据不一致，例如insert 语句中包含 now()函数。
Row-based：基于行，Binlog会记录每次写操作后被操作行记录的变化。
优点：保持数据的绝对一致性。
缺点：占用较大空间。
mixed：混合模式，默认是Statement-based，如果SQL语句可能导致数据不一致，就自动切换到Row-based。
Maxwell要求Binlog采用Row-based模式。
3）重启MySQL服务

[atguigu@hadoop102 ~]$ sudo systemctl restart mysqld

注：Docker安装的mysql 修改配置文件的方法在【Docker安装MySQL】中有相应说明

2.创建Maxwell所需数据库和用户

Maxwell需要在MySQL中存储其运行过程中的所需的一些数据，包括binlog 同步的断点位置（Maxwell支持断点续传）等等，故需要在MySQL为Maxwell创建数据库及用户。
1）创建数据库

msyql> CREATE DATABASE maxwell;

2）调整MySQL数据库密码级别（Docker安装的Mysql省略这一步）

mysql> set global validate_password_policy=0;
mysql> set global validate_password_length=4;

3）创建Maxwell用户并赋予其必要权限

mysql> CREATE USER 'maxwell'@'%' IDENTIFIED BY 'maxwell';
mysql> GRANT ALL ON maxwell.* TO 'maxwell'@'%';
mysql> GRANT SELECT, REPLICATION CLIENT, REPLICATION SLAVE ON *.* TO 'maxwell'@'%';

3.配置Maxwell

1）修改Maxwell配置文件名称

[atguigu@hadoop102 maxwell]$ cd /opt/module/maxwell
[atguigu@hadoop102 maxwell]$ cp config.properties.example config.properties

2）修改Maxwell配置文件

[atguigu@hadoop102 maxwell]$ vim config.properties

#Maxwell数据发送目的地，可选配置有stdout|file|kafka|kinesis|pubsub|sqs|rabbitmq|redis
producer=kafka
#目标Kafka集群地址
kafka.bootstrap.servers=hadoop102:9092,hadoop103:9092
#目标Kafka topic，可静态配置，例如:maxwell，也可动态配置，例如：%{database}_%{table}
kafka_topic=maxwell

#MySQL相关配置
host=hadoop102
user=maxwell
password=maxwell
jdbc_options=useSSL=false&amp;serverTimezone=Asia/Shanghai

Maxwell使用

1.启动Zookeeper以及Kafka集群

2.Maxwell启停

1）启动Maxwell

[atguigu@hadoop102 ~]$ /opt/module/maxwell/bin/maxwell --config /opt/module/maxwell/config.properties --daemon

2）停止Maxwell

[atguigu@hadoop102 ~]$ ps -ef | grep maxwell | grep -v grep | grep maxwell | awk '{print $2}' | xargs kill -9

3）Maxwell启停脚本
（1）创建并编辑Maxwell启停脚本

[atguigu@hadoop102 bin]$ vim mxw.sh

（2）脚本内容如下

#!/bin/bash

MAXWELL_HOME=/opt/module/maxwell
status_maxwell(){
    result=`ps -ef | grep com.zendesk.maxwell.Maxwell | grep -v grep | wc -l`
    return $result
}
start_maxwell(){
    status_maxwell
    if [[ $? -lt 1 ]]; then
        echo "启动Maxwell"
        $MAXWELL_HOME/bin/maxwell --config $MAXWELL_HOME/config.properties --daemon
    else
        echo "Maxwell正在运行"
    fi
}
stop_maxwell(){
    status_maxwell
    if [[ $? -gt 0 ]]; then
        echo "停止Maxwell"
        ps -ef | grep com.zendesk.maxwell.Maxwell | grep -v grep | awk '{print $2}' | xargs kill -9
    else
        echo "Maxwell未在运行"
    fi
}
case $1 in
    start )
        start_maxwell
    ;;
    stop )
        stop_maxwell
    ;;
    restart )
       stop_maxwell
       start_maxwell
    ;;
esac

增量数据同步

1）启动Kafka消费者

[atguigu@hadoop102 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic maxwell

2）模拟生成数据

[root@hadoop102 db_log]# ./db_mock.sh 2020-06-18
正在生成2020-06-18的数据

3）观察Kafka消费者

历史数据 全量 同步

1.Maxwell提供了bootstrap 功能来进行历史数据的全量同步，命令如下：

[atguigu@hadoop102 maxwell]$ /opt/module/maxwell/bin/maxwell-bootstrap --database gmall --table spu_info --config /opt/module/maxwell/config.properties

2.观察Kafka消费者
 boostrap数据格式
采用bootstrap 方式同步的输出数据格式如下：

{
    "database": "fooDB",
    "table": "barTable",
    "type": "bootstrap-start",
    "ts": 1450557744,
    "data": {}
}
{
    "database": "fooDB",
    "table": "barTable",
    "type": "bootstrap-insert",
    "ts": 1450557744,
    "data": {
        "txt": "hello"
    }
}
{
    "database": "fooDB",
    "table": "barTable",
    "type": "bootstrap-insert",
    "ts": 1450557744,
    "data": {
        "txt": "bootstrap!"
    }
}
{
    "database": "fooDB",
    "table": "barTable",
    "type": "bootstrap-complete",
    "ts": 1450557744,
    "data": {}
}