MySQL同步数据到Elasticsearch

背景

随着平台的业务日益增多，基于数据库的全文搜索查询速度较慢，已经无法满足需求。所以，决定基于Elasticsearch 做一个全文搜索平台，支持业务相关的搜索需求。那么第一个问题就是：如何从MySQL同步数据到Elasticsearch？

解决方案一：基于Logstash同步数据

该方案上次有详细说明过，这里就简单描述一下。

Logstash同步数据流程图：

优点： 1、组件少，只需要Logstash就可以实现； 2、配置简单，配置Logstash文件就可以。

缺点： 在数据量很大的情况下，Logstash可能会成为性能瓶颈

流程步骤

docker 启动Logstash

// docker启动logstash
docker run --name logstash -d -p 5044:5044 -v D:\work\iio\dockerFile\logstash\data:\usr\share\logstash logstash:7.11.2
复制代码

修改配置文件

进入logstash容器中修改配置文件

1）修改/config/logstash.yml 中的es地址
2）修改/pipeline/logstash.conf 中的相关配置（input、output、filler）
复制代码

logstash.conf 配置文件（仅供参考）：

jdbc {
        jdbc_connection_string => "jdbc:mysql://127.0.0.1:3306/dedao"
        jdbc_user => "root"
        jdbc_password => "****"
        jdbc_driver_library => "/usr/share/logstash/driver/mysql-connector-java-8.0.23.jar"
        jdbc_driver_class => "com.mysql.jdbc.Driver"
        jdbc_paging_enabled => "true"
        jdbc_page_size => "50000"
        lowercase_column_names => false
        # statement_filepath => "filename.sql"
        statement => "SELECT id, name FROM a"
        schedule => "* * * * *"
        type => "product"
    }
 
output {
 
        if[type]=="product"{
        elasticsearch {
            hosts => ["127.0.0.1:9200"]
            #manage_template => false
            #template_name => "myik"
            #template => "/usr/share/logstash/template/test_template.json"
            #template_overwrite => true
            document_id => "%{salesNo}"
            index => "logstash-dedao"
        }
 
        }
}
复制代码

解决方案二：基于canal同步数据

canal同步数据流程图：

优点：

    1、canal是同步MySQL的binlog日志，不需要全量更新数据；
 
    2、Kafka是一个高吞吐量的分布式发布订阅消息系统，性能高速度快。
复制代码

缺点：

    1、组件较多，有canal-server、Kafka 和canal-adapter 三个组件；
 
    2、配置相对复杂。
复制代码

流程步骤

修改MySQL配置

1、修改/etc/mysql/my.cnf 配置文件，开启binlog日志

[mysqld]
# 打开binlog
log-bin=mysql-bin
# # 选择ROW(行)模式
binlog-format=ROW
# # 配置MySQL replaction需要定义，不要和canal的slaveId重复
server_id=1
pid-file  = /var/run/mysqld/mysqld.pid
socket    = /var/run/mysqld/mysqld.sock
datadir    = /var/lib/mysql
#log-error  = /var/log/mysql/error.log
# By default we only accept connections from localhost
#bind-address  = 127.0.0.1
# Disabling symbolic-links is recommended to prevent assorted security risks
symbolic-links=0
character-set-server=utf8mb4
lower_case_table_names=1
sql-mode=STRICT_TRANS_TABLES,NO_ZERO_IN_DATE,NO_ZERO_DATE,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION
复制代码

2、在Navicat或DataGrip等终端执行命令，配置canal账号

grant SELECT, REPLICATION SLAVE, REPLICATION CLIENT on *.* to 'canal'@'%' identified by "canal";
 
flush privileges;
复制代码

docker 启动canal-server

1、启动canal-server

// 启动canal-server
docker run -p 11111:11111 --name canal -d canal/canal-server:v1.1.5
 
// 拷贝配置文件到本都路径
docker cp canal:/home/admin/canal-server/con