kafka-maxwell-kafka信息采集通道

忘是亡心i 2023-09-26 11:25 12阅读 0赞

**目录**

一：安装maxwell及配置

1.1 maxwell简介

1.2 maxwell输出数据格式

1.3 maxwell原理

1.3.1 MySQL二进制日志

1.3.2 MySQL主从复制

1.4 maxwell部署

1.5 配置MYSQL

1.5.1 启用MySQL Binlog

1.5.2 创建Maxwell所需数据库和用户

1.5.3 配置Maxwell

1.6 maxwell的使用

1.6.1 maxwell启停

二：采集信息通道

2.1 采集通道 maxwell配置

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

# 一：安装maxwell及配置 #

## 1.1 maxwell简介 ##

> Maxwell 是由美国Zendesk公司开源，用Java编写的**MySQL变更数据抓取软件**。它会**实时监控Mysql数据库的数据变更操作**（包括insert、update、delete），并将变更数据**以 JSON 格式**发送给 Kafka、Kinesi等**流数据处理平台**。

## 1.2 maxwell输出数据格式 ##

![45e21fa1bf954fcaa2ff0304aeeaa252.png][]

**注：Maxwell输出的json字段说明：**

![9e8b2b23b8a54bb1a4201d30a1f7612f.png][]

## 1.3 maxwell原理 ##

> Maxwell的工作原理是实时**读取MySQL数据库的二进制日志**（Binlog），从中获取变更数据，再将变更数据以JSON格式发送至Kafka等流处理平台。

**很简单，就是将自己伪装成slave，并遵循MySQL主从复制的协议，从master同步数据。**

### **1.3.1** **MySQL二进制日志** ###

> 二进制日志（Binlog）是MySQL服务端非常重要的一种日志，它会**保存MySQL数据库的所有数据变更记录**。Binlog的主要作用包括**主从复制**和**数据恢复**。Maxwell的工作原理和**主从复制**密切相关。

### 1.3.2 MySQL主从复制 ###

> MySQL的主从复制，就是用来建立一个和主数据库完全一样的数据库环境，这个数据库称为从数据库。

**1）主从复制的应用场景如下：**

> （1）做数据库的热备：主数据库服务器故障后，可切换到从数据库继续工作。（备胎）
> 
> （2）读写分离：主数据库只负责业务数据的写入操作，而多个从数据库只负责业务数据的查询工作，在**读多写少**场景下，可以提高数据库工作效率。

**2）主从复制的工作原理如下：**

> （1）Master主库将数据变更记录，写到二进制日志(binary log)中
> 
> （2）Slave从库向mysql master发送**dump协议**，将master主库的binary log events拷贝到它的**中继日志**(relay log)
> 
> （3）Slave从库读取并回放中继日志中的事件，将改变的数据同步到自己的数据库。

Master主数据库**--》**写入二进制日志（binlog）**--》**Slave从数据库发送dump协议**--》**binlog events拷贝到中继日志（relay log）**--》**Slave读取中继日志，同步至自己的数据库。

![112d1597e9c8408f9b2f7ae225f3c940.png][]

## 1.4 maxwell部署 ##

**注：Maxwell-****1.30.0****及以上版本不再支持JDK****1.8****。**

**注：此处使用教学版安装包，教学版对原版进行了改造，增加了自定义Maxwell输出数据中ts时间戳的参数，生产环境请使用原版。**

（1）**将安装包解压至/****opt/module**

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

（2）修改文件夹名称

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

## 1.5 配置MYSQL ##

### 1.5.1 **启用MySQL****Binlog** ###

> MySQL服务器的Binlog默认是未开启的，如需进行同步，需要先进行开启。

**1****）修改MySQL配置文件****/etc/my.cnf**

[axing@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服务**

[axing@hadoop102 ~]$ sudo systemctl restart mysqld

### 1.5.2 **创建Maxwell所需数据库和用户** ###

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

**1）创建数据库**

msyql> CREATE DATABASE maxwell;

**2）调整****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'@'%';
    //任何外部可以访问maxwell
    mysql> GRANT SELECT, REPLICATION CLIENT, REPLICATION SLAVE ON *.* TO 'maxwell'@'%';

### **1.5.3 配置Maxwell** ###

**1****）修改Maxwell配置文件名称**

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

**2）修改Maxwell配置文件**

[axing@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&serverTimezone=Asia/Shanghai

## 1.6 maxwell的使用 ##

### 1.6.1 maxwell启停 ###

**1）启动Maxwell**

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

**2****）停止Maxwell**

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

**3）M****axwell****启停脚本**

#!/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

# 二：采集信息通道 #

> **前提准备**：打开zookeeper，kafka，maxwell
> 
> 建议打开顺序：
> 
> 1.zookeeper
> 
> 2.kafka
> 
> 3.maxwell
> 
> 不然，kafka进程可能能会死掉，我的虚拟机是这样的。

## 2.1 采集通道 maxwell配置 ##

**1****）修改Maxwell配置文件config****.properties**

[axing@hadoop102 maxwell]$ vim /opt/module/maxwell/config.properties

**2）配置参数如下**

log_level=info
    
    producer=kafka
    kafka.bootstrap.servers=hadoop102:9092,hadoop103:9092
    
    #kafka topic配置
    kafka_topic=topic_db
    
    # mysql login info
    host=hadoop102
    user=maxwell
    password=maxwell
    jdbc_options=useSSL=false&serverTimezone=Asia/Shanghai

> 注意：这里配置的kafka\_topic是topic\_db，下文kafka消费数据时，接收的topic也应该是topic\_db，前后一致。

**3）重新启动Maxwell**

[axing@hadoop102 bin]$ mxw.sh restart

**4****）通道测试**

（1）启动Zookeeper以及Kafka集群（如果已经打开就跳过这步）

（2）启动一个Kafka Console Consumer，消费topic\_db数据

[axing@hadoop103 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic topic_db

（3）生成模拟数据

[axing@hadoop102 bin]$ cd /opt/module/db_log/
    [axing@hadoop102 db_log]$ java -jar gmall2020-mock-db-2021-11-14.jar

> 这里的jar是一个脚本，运行jar包后会自动模拟生成数据。

（4）观察Kafka消费者是否能消费到数据

![7efbd83b1a4142d48e65a30a4a3f4594.png][]

[45e21fa1bf954fcaa2ff0304aeeaa252.png]: https://img-blog.csdnimg.cn/45e21fa1bf954fcaa2ff0304aeeaa252.png
[9e8b2b23b8a54bb1a4201d30a1f7612f.png]: https://img-blog.csdnimg.cn/9e8b2b23b8a54bb1a4201d30a1f7612f.png
[112d1597e9c8408f9b2f7ae225f3c940.png]: https://img-blog.csdnimg.cn/112d1597e9c8408f9b2f7ae225f3c940.png
[7efbd83b1a4142d48e65a30a4a3f4594.png]: https://img-blog.csdnimg.cn/7efbd83b1a4142d48e65a30a4a3f4594.png