0
点赞
收藏
分享

微信扫一扫

[Maxwell基础]--手把手搭建maxwell+kafka的环境

栖桐 2023-07-04 阅读 74
mysqlmaxwellServersql代码人生


Mysql bin log学习

1、mysql 版本

$ mysql -v

Server version: 5.6.41 MySQL Community Server (GPL)

 

2、mysql bin log 介绍

https://dev.mysql.com/doc/internals/en/binary-log.html

 

2.1是否启用binlog日志

show variables like 'log_bin';

 

2.2查看binlog

show binlog events;

 

2.3查看最新一个binlog日志文件名称

show master status;

 

2.4查找binlog日志

find / -name mysql-bin -type f

 

2.5查看详细的日志配置信息

SHOW  GLOBAL VARIABLES LIKE '%log%';

 

2.6mysql数据存储目录

show variables like '%dir%';

 

2.7查看binlog的目录

  

mysql> show global variables like "%log_bin%";
       +---------------------------------+--------------------------------+
       | Variable_name                   | Value                          |
       +---------------------------------+--------------------------------+
       | log_bin                         | ON                             |
       | log_bin_basename              | /var/lib/mysql/mysql-bin       |
       | log_bin_index                   | /var/lib/mysql/mysql-bin.index |
       | log_bin_trust_function_creators  | OFF                            |
       | log_bin_use_v1_row_events     | OFF                            |
       | sql_log_bin                     | ON                             |
       +---------------------------------+--------------------------------+

注：参数说明

https://dev.mysql.com/doc/refman/5.6/en/replication-options-binary-log.html#sysvar_log_bin_basename

1)log_bin_basename：保存二进制日志文件的基本名称和路径

2)log_bin_index；保存二进制日志索引文件的基本名称和路径，该log_bin_index系统变量在MySQL 5.6.4添加

 

 

2.8查看bin-log二进制文件

相关ddl和dml操作

   /

执行的sql语句：

     

mysql> create database test_binlog;
       Query OK, 1 row affected (0.00 sec)
 
       mysql> use test_binlog;
       Database changed
 
       mysql> create table test_bin_log (id int ,name varchar(30));
       Query OK, 0 rows affected (0.02 sec)
 
       mysql> insert into test_bin_log values(1,"mysql001");
       Query OK, 1 row affected (0.00 sec)
 
       mysql> insert into test_bin_log values(2,"mysql002");
       Query OK, 1 row affected (0.01 sec)     
 
  //

 

2.8.1方法1（推荐）

     show binlog events in 'mysql-bin.000003';

     mysql> show binlog events in 'mysql-bin.000003';

+------------------+

| Log_name         | Pos  | Event_type  | Server_id | End_log_pos | Info                                                                                                                                                                               |

----------------+

| mysql-bin.000003 |    4 | Format_desc |         1 |         120 | Server ver: 5.6.41-log, Binlog ver: 4                                                                                                                                              |

| mysql-bin.000003 |  120 | Query       |         1 |         235 | create database test_binlog                                                                                                                                                        |

| mysql-bin.000003 |  235 | Query       |         1 |         375 | use `test_binlog`; create table test_bin_log (id int ,name varchar(30))                                                                                                            |

| mysql-bin.000003 |  375 | Query       |         1 |         468 | BEGIN                                                                                                                                                                              |

| mysql-bin.000003 |  468 | Query       |         1 |         601 | use `test_binlog`; insert into test_bin_log values(1,"mysql001")                                                                                                                   |

| mysql-bin.000003 |  601 | Xid         |         1 |         632 | COMMIT /* xid=69452 */                                                                                                                                                             |

| mysql-bin.000003 |  632 | Query       |         1 |         725 | BEGIN                                                                                                                                                                              |

| mysql-bin.000003 |  725 | Query       |         1 |         858 | use `test_binlog`; insert into test_bin_log values(2,"mysql002")                                                                                                                   |

| mysql-bin.000003 |  858 | Xid         |         1 |         889 | COMMIT /* xid=69589 */                                                                                                                                                             |

| mysql-bin.000003 |  889 | Query       |         1 |         965 | BEGIN                                                                                                                                                                              |

| mysql-bin.000003 |  965 | Query       |         1 |        1188 | use `ambari`; UPDATE QRTZ_TRIGGERS SET TRIGGER_STATE = 'WAITING' WHERE SCHED_NAME = 'ExecutionScheduler' AND (TRIGGER_STATE = 'ACQUIRED' OR TRIGGER_STATE = 'BLOCKED')             |

| mysql-bin.000003 | 1188 | Query       |         1 |        1423 | use `ambari`; UPDATE QRTZ_TRIGGERS SET TRIGGER_STATE = 'PAUSED' WHERE SCHED_NAME = 'ExecutionScheduler' AND (TRIGGER_STATE = 'PAUSED_BLOCKED' OR TRIGGER_STATE = 'PAUSED_BLOCKED') |

| mysql-bin.000003 | 1423 | Query       |         1 |        1572 | use `ambari`; DELETE FROM QRTZ_FIRED_TRIGGERS WHERE SCHED_NAME = 'ExecutionScheduler'                                                                                              |

| mysql-bin.000003 | 1572 | Query       |         1 |        1647 | COMMIT                                                                                                                                                                             |

+------------------+------+-------------+-----------+-------------

2.8.2 方法2

 [root@bdp03nn01 mysql]# mysqlbinlog -v --base64-output=decode-rows --start-datetime="2018-08-22 08:30:00" --stop-datetime="2018-08-26 00:30:00" /var/lib/mysql/mysql-bin.000003  

 \\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\

/*!50530 SET @@SESSION.PSEUDO_SLAVE_MODE=1*/;

/*!40019 SET @@session.max_insert_delayed_threads=0*/;

/*!50003 SET @OLD_COMPLETION_TYPE=@@COMPLETION_TYPE,COMPLETION_TYPE=0*/;

DELIMITER /*!*/;

# at 4

#180823 16:28:03 server id 1  end_log_pos 120 CRC32 0x7f57797a     Start: binlog v 4, server v 5.6.41-log created 180823 16:28:03

# Warning: this binlog is either in use or was not closed properly.

BINLOG '

              E3B+Ww8BAAAAdAAAAHgAAAABAAQANS42LjQxLWxvZwAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA

              AAAAAAAAAAAAAAAAAAAAAAAAEzgNAAgAEgAEBAQEEgAAXAAEGggAAAAICAgCAAAACgoKGRkAAXp5

              V38=

              '/*!*/;

              # at 120

              #180823 16:53:40 server id 1  end_log_pos 235 CRC32 0xe6eb36b4    Query     thread_id=1088    exec_time=0  error_code=0

              SET TIMESTAMP=1535014420/*!*/;

              SET @@session.pseudo_thread_id=1088/*!*/;

              SET @@session.foreign_key_checks=1, @@session.sql_auto_is_null=0, @@session.unique_checks=1, @@session.autocommit=1/*!*/;

              SET @@session.sql_mode=1075838976/*!*/;

              SET @@session.auto_increment_increment=1, @@session.auto_increment_offset=1/*!*/;

              /*!\C utf8 *//*!*/;

              SET @@session.character_set_client=33,@@session.collation_connection=33,@@session.collation_server=33/*!*/;

              SET @@session.lc_time_names=0/*!*/;

              SET @@session.collation_database=DEFAULT/*!*/;

              create database test_binlog

              /*!*/;

              # at 235

              #180823 16:54:38 server id 1  end_log_pos 375 CRC32 0x8267becb    Query     thread_id=1088    exec_time=0  error_code=0

              use `test_binlog`/*!*/;

              SET TIMESTAMP=1535014478/*!*/;

              create table test_bin_log (id int ,name varchar(30))

              /*!*/;

              # at 375

              #180823 16:55:12 server id 1  end_log_pos 468 CRC32 0x10fcd3fe      Query     thread_id=1088    exec_time=0  error_code=0

              SET TIMESTAMP=1535014512/*!*/;

              BEGIN

              /*!*/;

              # at 468

              #180823 16:55:12 server id 1  end_log_pos 601 CRC32 0x161b0df5     Query     thread_id=1088    exec_time=0  error_code=0

              SET TIMESTAMP=1535014512/*!*/;

              insert into test_bin_log values(1,"mysql001")

              /*!*/;

              # at 601

              #180823 16:55:12 server id 1  end_log_pos 632 CRC32 0x776d3e6c    Xid = 69452

              COMMIT/*!*/;

              # at 632

              #180823 16:55:18 server id 1  end_log_pos 725 CRC32 0xc2d44757    Query     thread_id=1088    exec_time=0  error_code=0

              SET TIMESTAMP=1535014518/*!*/;

              BEGIN

              /*!*/;

              # at 725

              #180823 16:55:18 server id 1  end_log_pos 858 CRC32 0xde83d0d3    Query     thread_id=1088    exec_time=0  error_code=0

              SET TIMESTAMP=1535014518/*!*/;

              insert into test_bin_log values(2,"mysql002")

              /*!*/;

              # at 858

              #180823 16:55:18 server id 1  end_log_pos 889 CRC32 0xbb62f482     Xid = 69589

              COMMIT/*!*/;

              DELIMITER ;

              # End of log file

              ROLLBACK /* added by mysqlbinlog */;

              /*!50003 SET COMPLETION_TYPE=@OLD_COMPLETION_TYPE*/;

              /*!50530 SET @@SESSION.PSEUDO_SLAVE_MODE=0*/;

   \\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\

 

参数说明：

       --start-datetime：从二进制日志中读取指定等于时间戳或者晚于本地计算机的时间

       --stop-datetime： 从二进制日志中读取指定小于时间戳或者等于本地计算机的时间 取值和上述一样

       --start-position：从二进制日志中读取指定position 事件位置作为开始。

       --stop-position： 从二进制日志中读取指定position 事件位置作为事件截至

 

3、解析和采集

3.1工具

canal：

免费-阿里巴巴

（

https://github.com/alibaba/canal/wiki/Introduction，

https://slidesplayer.com/slide/11557608/，

https://docs.google.com/presentation/d/1MkszUPYRDkfVPz9IqOT1LLT5d9tuwde_WC8GZvjaDRg/edit#）

 

maxwells： 免费-亚马逊（http://maxwells-daemon.io/）

 

3.2举例

maxwells:  https://github.com/bobshaw1912/cdc-kinesis-demo，

canal:     https://github.com/alibaba/canal/wiki/QuickStart

 

3.3版本说明

3.3.1 canal

当前最高支持版本是<mysql 5.5

功能多，太重

https://github.com/alibaba/canal/wiki/AdminGuide

 

3.3.2 maxwells

功能少，较轻，适合我们的应用场景

4、应用场景

4.1 数据异构、缓冲和任务分发

http://www.seekwd.com/article/details/150

 

4.2 实现流式实时分析架构

https://aws.amazon.com/cn/blogs/china/mysql-binlog-architecture/

 

4.3 数据恢复

https://www.cnblogs.com/kevingrace/p/5907254.html

 

4.4 数据实时处理方法及系统

https://patents.google.com/patent/CN106446239A/zh

 

5、搭建

5.1 手把手搭建maxwell+kafka的环境

http://maxwells-daemon.io/

5.2 环境说明

5.2.1 需要的版本

apache kafka:  0.8.2以上，版本需要对应，否则可能出现不兼容

maxwells :     1.2.X以上

mysql    :     5.1，5.5，5.6，5.7

jdk       :     1.8以上

操作系统  :     centos7

5.2.2 集群上的环境

测试集群：hdp-2.6.4.0-91、jdk-1.8.0_141、kafka-0.10.1.2.6.4.0-91

生产环境：hdp-2.6.0.3-8、jdk-1.8.0_141、kafka-0.10.1.2.6.0.3-8

自玩环境：hdp-2.6.5.0-292、jdk-1.8.0_172-b11、mysql-5.6.41、kafka-1.0.0.2.6.5.0-292

5.2.3 maxwells版本

maxwells-1.17.0 ----> https://github.com/zendesk/maxwell/tree/v1.17.0

 

5.3 安装

5.3.1 测试kafka是否正常安装

5.3.1.1 创建kafka的topic

/usr/hdp/2.6.5.0-292/kafka/bin/kafka-topics.sh --create  --topic  maxwells --zookeeper localhost:2181 --partitions 6 --replication-factor 2

 

5.3.1.2 生产数据

/usr/hdp/2.6.5.0-292/kafka/bin/kafka-console-producer.sh  --broker-list bdp03nn01:6667,bdp03dn01:6667,bdp03nn02:6667 --topic maxwells

 

5.3.1.3 消费数据

/usr/hdp/2.6.5.0-292/kafka/bin/kafka-console-consumer.sh --zookeeper bdp03nn01:2181,bdp03dn01:2181,bdp03nn02:2181 --topic maxwells --from-beginning

 

5.3.1.4 描述topic

/usr/hdp/2.6.5.0-292/kafka/bin/kafka-topics.sh --describe  --zookeeper bdp03nn01:2181 --topic maxwells

 

5.3.2 安装maxwell

5.3.2.1 下载、并解压maxwells
$ cd /usr/local
$ wget https://github.com/zendesk/maxwell/releases/download/v1.17.0/maxwell-1.17.0.tar.gz
 
5.3.2.2 配置mysql
5.3.2.2.1编辑my.cnf
$ vi /etc/my.cnf
[mysqld]
server_id=1
log-bin=master
binlog_format=row

 

5.3.2.2.2 添加权限

 

mysql> GRANT ALL  on *.* to 'maxwell'@'172.16.5.117' identified by 'mysqlmaxwellpwd' ;
mysql> GRANT SELECT, REPLICATION CLIENT, REPLICATION SLAVE on *.* to 'maxwell'@'172.16.5.117';
mysql> flush privaleges;
 
5.3.2.2.3 刷新bin-log日志、重启mysql
mysql> flush logs;
#systemctl  restart mysqld.service

 

5.3.3 运行maxwells

5.3.3.1 未过滤库

/usr/local/maxwell/bin/maxwell --user='maxwell' --password='mysqlmaxwellpwd' --host='172.16.5.117' \
--producer=kafka --kafka.bootstrap.servers=bdp03nn01:6667,bdp03dn01:6667,bdp03nn02:6667 \
--kafka_topic=maxwells

5.3.3.2 过滤指定规则的库

/usr/local/maxwell/bin/maxwell --user='maxwell' --password='mysqlmaxwellpwd' --host='172.16.5.117' \

--producer=kafka --kafka.bootstrap.servers=bdp03nn01:6667,bdp03dn01:6667,bdp03nn02:6667 \

--kafka_topic=maxwells  --filter 'exclude: ambari.*, include: test_binlog.*'

 

参考过滤配置：http://maxwells-daemon.io/filtering/

 

5.4 消费数据

#/usr/hdp/2.6.5.0-292/kafka/bin/kafka-console-consumer.sh --zookeeper bdp03nn01:2181,bdp03dn01:2181,bdp03nn02:2181 --topic maxwells --from-beginning

 

命令行监控的数据：

{"database":"ambari","table":"alert_current","type":"update","ts":1535363391,"xid":98489,"commit":true,"data":{"alert_id":122,"definition_id":82,"history_id":451,"maintenance_state":"ON","original_timestamp":1534996534493,"latest_timestamp":1535363390926,"latest_text":"Connection failed: [Errno 111] Connection refused to bdp03nn01:16000","occurrences":4991,"firmness":"HARD"},"old":{"latest_timestamp":1535363330967,"occurrences":4990}}

 

5.5 maxwell 的json数据说明

http://maxwells-daemon.io/dataformat/

 

5.6 参考

https://github.com/zendesk/maxwell

http://maxwells-daemon.io/compat/

https://github.com/zendesk/maxwell/releases

https://github.com/zendesk/maxwell/blob/master/CHANGELOG.md

http://maxwells-daemon.io/config/

https://docs.hortonworks.com/HDPDocuments/HDP2/HDP-2.6.3/bk_security/content/secure-kafka-produce-events.html

https://docs.hortonworks.com/HDPDocuments/HDP2/HDP-2.6.5/bk_kafka-component-guide/content/ch_kafka-development.html

 

举报

相关推荐

手把手搭建MQTT测试环境

MQTT服务端客户端物联网yyds干货盘点

手把手带你搭建Linux环境（保姆级）

linuxfpga开发ubuntu

手把手搭建自己的网站

服务器

手把手教你 SpringBoot整合Kafka

Spring BootkafkaspringapacheLinux系统/运维

【Linux编程】手把手教你搭建Linux编程环境

linux服务器运维shell虚拟机

Maxwell监控MySQL的增量数据

Maxwellkafka数据mysqlMySQL数据库

手把手搭建QEMU ARM64开发环境

linux文件系统pythonqemuLinux内核软件研发

Canal或maxwell实时采集MySQL数据到Kafka

mysqlcanalmaxwell数据同步实时大数据

手把手教你玩转AI之环境搭建-集成开发环境

pythonpycharmjupytervscode

手把手教你搭建 RabbitMQ 集群

rabbitmq分布式javadocker数据
0 条评论