mysqlbinlog入elasticsearch

Posted 2020-10-31

环境准备：

10.99.35.214上安装elasticsearch-5.4.1.tar.gz、jdk-8u121-linux-x64.tar.gz、kibana-5.1.1-linux-x86_64.tar.gz、nginx-1.12.2.tar.gz

10.99.35.215、10.99.35.216上安装elasticsearch-5.4.1.tar.gz、jdk-8u121-linux-x64.tar.gz

10.99.35.209上安装mysql-5.7.17-linux-glibc2.5-x86_64.tar.gz、jdk-8u121-linux-x64.tar.gz、logstash-5.5.1.tar.gz、maxwell-1.10.7.tar.gz、kafka_2.11-0.11.0.1

209上安装mysql，通过maxwell抽取binlog日志，发送到kafka中，logstash抽取kafka中数据，并传给214、215、216组成elasticsearch集群，在214上通过kibana与nginx来展示

1, 开启binlog

vim /etc/my.cnf

server-id=1

log-bin=master（这一步开启binlog）

binlog_format=row

2, mysql授权

GRANT ALL on maxwell.* to'maxwell'@'%' identified by '[email protected]';

GRANT SELECT, REPLICATION CLIENT,REPLICATION SLAVE on *.* to 'maxwell'@'%';

flushprivileges;

3, 配置Maxwell

wget https://github.com/zendesk/maxwell/releases/download/v1.10.7/maxwell-1.10.7.tar.gz

tar xvfz maxwell-1.10.7.tar.gz

4, 开启maxwell命令行

nohup bin/maxwell --user='maxwell' --password='[email protected]' --host='10.99.35.209' --producer=kafka --kafka.bootstrap.servers=10.99.35.209:9092 &

解释：host参数是安装mysql的那台主机，最后的kafka.bootstrap.servers是安装kafka集群的节点主机名和端口号

5, kafka相关配置

说明（我的kafka是安装在主机名叫10.99.35.209，mysql也安装在10.99.35.209上，注意kafka里的配置文件端口号要和命令行里给的端口号一致）

wget http://mirror.bit.edu.cn/apache/kafka/0.10.2.1/kafka_2.11-0.10.2.1.tgz

mv kafka_2.11-0.10.2.1.tgz /usr/local/

cd /usr/local/

tar xvfz kafka_2.11-0.10.2.1.tgz

Kafka使用ZooKeeper，所以需要先启动一个ZooKeeper服务器，如果还没有。可以使用随Kafka一起打包的便捷脚本来获取一个快速但是比较粗糙的单节点ZooKeeper实例

nohup bin/zookeeper-server-start.sh config/zookeeper.properties &

Kafka在config目录下提供了一个基本的配置文件。为了保证可以远程访问Kafka，我们需要修改两处配置。打开config/server.properties文件，在很靠前的位置有listeners和 advertised.listeners两处配置的注释，去掉这两个注释，并且根据当前服务器的IP修改如下：

listeners=PLAINTEXT://:9092

advertised.listeners=PLAINTEXT://10.99.35.209:9092

当前服务器IP为10.99.35.209，你需要修改为外网或局域网可以访问到的服务器IP。

接下来启动Kafka服务：

nohup bin/kafka-server-start.sh config/server.properties &

创建一个主题叫maxwell以便于接受数据

bin/kafka-topics.sh --create --zookeeper 10.99.35.209:2181 --replication-factor 1 --partitions 1 --topic maxwell

创建好之后可以通过运行以下命令查看已创建的topic信息

bin/kafka-topics.sh --list  --zookeeper 10.99.35.209:2181

查看topic详细情况

bin/kafka-topics.sh --describe --zookeeper 10.99.35.209:2181  --topic maxwell

启动生产者窗口

bin/kafka-console-producer.sh --broker-list 10.99.35.209:9092 --topic maxwell

启动消费者窗口

bin/kafka-console-consumer.sh --zookeeper 10.99.35.209:2181 --topic maxwell --from-beginning

6, 安装配置logstash

修改logstash配置文件

log.level: info

path.logs: /usr/local/logstash-5.1.1/logs

实例：

启动Zookeeper

bin/zookeeper-server-start.sh -daemon config/zookeeper.properties

启动Kafka

bin/kafka-server-start.sh config/server.properties

启动Maxwell

bin/maxwell --user='maxwell' --password='[email protected]' --host='10.99.35.209' --producer=kafka --kafka.bootstrap.servers=10.99.35.209:9092

Logstash配置如下：

bin/logstash-plugin install logstash-filter-translate

input {

    kafka {

        bootstrap_servers => "10.99.35.209:9092"

        topics => ["maxwell"]

        codec => json {charset => ["ISO-8859-1"]}

        consumer_threads => 5

        decorate_events => true

    }

}

filter {

  mutate {   

       remove_field => ["database","table","ts","xid","commit","old","kafka"]

       rename => ["[data][id]","id"]

       rename => ["[data][first_name]","first_name"]

       rename => ["[data][last_name]","last_name"]

       rename => ["[data][age]","age"]

       rename => ["[data][about]","about"]

       rename => ["[data][interests]","interests"]

   }

   translate {

       field => "type"

       destination => "op_type"

       dictionary => [

         "insert", "index",

         "update", "update",

         "delete", "delete"

       ]

    }

}

output {

    elasticsearch {

        hosts => ["10.99.35.214:9200"]

        index => "megacorp"       

        document_id => "%{id}"

        document_type => "employee"

        action => "%{op_type}"

        workers => 1

        flush_size => 20000

        idle_flush_time => 10

        template_overwrite => true

    }

    stdout {}

}

启动Logstash和Elasticsearch，在MySQL数据库中增删改数据，Elasticsearch就会实时看到索引的变化。

#end