LogStash实现MySQL数据增量同步到ElasticSearch

柔情只为你懂 2022-12-30 15:51 272阅读 0赞

**需求的由来**

> 在做一个项目的时候，需要一个搜索的功能，刚开始想到的是用数据库的模糊查询，但是考虑到效率的问题，于是就用了ElasticSearch（ES），但是MySQL中的数据怎么样到ES中呢，我们可能会想到，MySQL可以实现主从复制，通过binary log文件实现的，搜集了一波资料发现，LogStash可以实现这个数据同步的功能，有增量和全量，如果数据只同步一次的话，可以使用全量同步，如果数据会有更新的话，可以使用增量同步，真是nice。

**安装LogStash**  
要用人家的功能，第一步当然是安装此利器了，直接去官网下载即可[LogStash下载][LogStash]  
![在这里插入图片描述][20201226184145761.png]  
下载完之后解压，由于是要跟MySQL打交道的，当然必须要有MySQL的相关配置了，第一步在bin的同级目录下创建一个mysql目录，然后将MySQL驱动放里面即可，[下载MySQL驱动][MySQL]  
![在这里插入图片描述][20201226184531564.png]  
下载完毕之后，解压，将里面的jar包拷贝到刚刚创建的mysql目录下即可。  
![在这里插入图片描述][20201226184629631.png]  
然后是在bin目录下创建一个`logstash.conf`的文件，内容如下：

input {
      # 多张表的同步只需要设置多个jdbc的模块就行了
      jdbc {
          # mysql 数据库链接
          jdbc_connection_string => "jdbc:mysql://localhost:3306/newsblog?userSSL=true&useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai"
          # 驱动
          jdbc_driver_library => "D:/elasticsearch/logstash-7.8.0/mysql/mysql-connector-java-8.0.22.jar"
          # 驱动类名
          jdbc_driver_class => "com.mysql.cj.jdbc.Driver"      
          # 用户名和密码
          jdbc_user => "root"
          jdbc_password => "root"
          #设置监听间隔  各字段含义（由左至右）分、时、天、月、年，全部为*默认含义为每分钟都更新
          schedule => "*/2 * * * *"
          #直接执行sql语句
          statement => "select * from blog where blogId > :sql_last_value"
          # 用其他字段追踪
          use_column_value => true
          tracking_column => "blogid"
          # 记录最新的同步的offset信息，会自动创建该文件
          last_run_metadata_path => "D:/elasticsearch/syncpoint_table.txt"
          # 是否清空文件
          clean_run => false
        }
    
    }
    
    
    output {
      elasticsearch {
            #es的ip和端口
            hosts => ["http://localhost:9200"]
            #ES索引名称（自己定义的）
            index => "blog"
            #文档类型
            document_type => "_doc"
            #设置数据的id为数据库中的字段，这里都是小写
            document_id => "%{blogid}"
        }
        stdout {
            codec => json_lines
        }
    }

**启动**  
在启动ES的前提下，启动LogStash，在bin目录下执行命令  
`logstash -f logstash.conf`  
然后就会根据配置来进行同步数据了，会生成一个记录offset文件。  
![在这里插入图片描述][20201226185508477.png]  
然后通过Kibana可以进行查询到数据。  
![在这里插入图片描述][20201226185616691.png]

[LogStash]: https://www.elastic.co/cn/downloads/logstash
[20201226184145761.png]: /images/20221120/c7905545c9f546cf849c145c1d45d1dd.png
[MySQL]: https://dev.mysql.com/downloads/connector/j/
[20201226184531564.png]: /images/20221120/43d21f3f3f5d41f18995813d3eacd08c.png
[20201226184629631.png]: https://img-blog.csdnimg.cn/20201226184629631.png
[20201226185508477.png]: https://img-blog.csdnimg.cn/20201226185508477.png
[20201226185616691.png]: https://img-blog.csdnimg.cn/20201226185616691.png