大数据项目任务：mysql与es集群数据实时同步更新

启动数据库net start mysql,

输入mysql -u root -p ,密码：root

出现mysql>配置完成

输入(alter user user() identified by "密码";)

mysql退出 mysql>quit;

net start mysql打开数据库

mysql退出 mysql>quit;

输入net stop mysql关闭数据库

任务一、对数据库mysql进行操作的时候，es也要跟着实时动态更新。

1.mysql数据库与elasticsearch全文检索的同步，通过binlog的设置对mysql数据库操作的日志进行记录，利用python的模块对日志进行操作，再利用kafka的生产者消费者模式进行订阅，最终实现elasticsearch的同步。

binlog技术实现MySQL和elasticsearch进行同步

binlog是数据库里的二进制日志文件，存储关于数据库增删改查的操作，只要有增删改查的操作就可以在binlog中体现出来，

用这个日志可以体现用户的增删改查。

解决两个问题

1.binlog如何设置

binlog与sql数据库有关系，可以在mysql配置文件my.ini，当配置文件修改时，重启数据库

打开命令行，输入mysql -u root -p ,密码：root；进入数据库

显示与binlog有关的变量: show global variables like "%binlog%";

修改binlog_format变量值为行级row：set global binlog_format="ROW";

再用show binlog events;查看数据库增删改的操作

2.python如何读取binlog数据的

python是通过mysql-replication实现binlog日志的读取的

然后打开pycharm去导入binlog监控数据库的操作

我到最后一步老报错那一行代码，for row in binlogstream.rows:

说什么编码错误:LookupError: unknown encoding: utf8mb3，在这里卡了很久，一直找不到解决方法

这里解决了这个问题，原因是我的数据库在创建的时候可能出了点问题，是使用命令行创建的，但之后试了一下使用SQLyog创建数据库选中基字符集类型之后，有一个uft8mb3的字符集，这个之前在网上查找的是数据库改成8.0版本之后没有utf8了，只有utf8mb3代替了，但这里还是会乱码，这之后改掉了创建表的时候列的数据类型，然后可以正常同步数据库的内容，成功输出

之后进行了下一步

下一步就是通过这个日志 把读出来的信息同步到elasticsearch中就更好了

因为数据量很大，所以使用Kafka生产者消费者模式对数据进行处理，消息来消息去

Kafka消息系统

采用页面缓存技术+磁盘顺序读写技术

Page cache基于页面缓存技术+磁盘顺序读写

Kafka实现需要zookeeper调度，对服务器集群进行管理的一个服务

启动zookeeper命令行:

进入d盘

cd d: 敲命令：zkServer

启动kafka服务命令行：先进入windows

kafka-server-start

再设置config中的server.properties

启动kafka

D:\kafka\kafka_2.13-3.4.0\bin\windows>kafka-server-start ..\..\config\server.properties

启动kafka完之后，启动kafka生产者消费者模式，主要利用的是kafka生产者消费者模式

打开命令行，进入d盘，

d：

ｃｄ　ｋａｆｋａ

ｃｄ　kafka_2.13-3.4.0

ｃｄ　ｂｉｎ

ｃｄ　ｗｉｎｄｏｗｓ

ｄｉｒ

ｋａｆｋａ是基于ｔｏｐｉｃ消息订阅模式的，利用topic由生产者发送到消费者，所以要先关联ｔｏｐｉｃ

启动生产者

kafka-console-producer --broker-list localhost:9092 --topic test

启动消费者（基于topic消息订阅的）

kafka-console-consumer --bootstrap-server localhost:9092 --topic test

在生产者输入this a book

如下图消费者会传过去数据

上述是命令行的启动和设置方式

现在用python模块来实现kafka的生产者消费者模式，安装的是kafka-python模块，实现了python操作kafka的服务

生产者端使用KafkaProducer

消费者端使用KafkaConsumer

注意：在消费者一端的参数要有生产者一端的变量名

现在要直接结合之前的reader，将mysql数据库的数据同步到kafka生产者消费者中，最后运行，再随便改几个数据库的数据，最后可以看到kafka消费者也同步了数据

这样就成功的将reader加入了kafka_producer中

接下来进行elasticsearch数据的同步

1. MySQL数据首先存储在elasticsearch中

（1）python读取mysql数据，用pysql的模块，建一个reader_data读取pysql数据

（2）把读取的数据存储到elasticsearch当中，可以使用Elasticsearch.index方法进行存储

先启动es

笔记做到这儿的时候，我不明白之前为什么不这样读取数据，还要用kafka生产者，消费者模式来处理，然后代码跑不通，构建不了es索引

调试很多次 调试不通，所以先去找找利用Java写SQL与es数据同步

2. 当MySQL中的数据发生变化时，同时elasticsearch中的数据也要发生变化

1.mysql数据库与elasticsearch全文检索的同步，通过binlog的设置对mysql数据库操作的日志进行记录，利用python的模块对日志进行操作，再利用kafka的生产者消费者模式进行订阅，最终实现elasticsearch的同步。

2.mysql数据库与elasticsearch全文检索的同步，通过binlog的设置对mysql数据库操作的日志进行记录，利用python的模块对日志进行操作，再利用kafka的生产者消费者模式进行订阅，最终实现elasticsearch的同步。本讲主要针对elasticsearch的增删改同步数据库。

create database readerBinlog default charset=utf8;

use readerBinlog;

create table mytable(id int(11),name varchar(20));

insert into mytable values(1,"孙大圣");

select * from mytable;

show binlog events;

update mytable set name ="黑熊怪" where id=5;

delete from mytable where id=5;