一.前言

之前网上看了很多文章，说mysql同步数据到Elasticsearch只能做增量同步，即只能新增（add）或者修改（update），不支持删除（delete）操作。所以一般的做法是使用逻辑删除，查询时带上逻辑删除字段，然后使用定时器定时去物理删除状态为删除的数据。最近刚好应用，一开始也是采用逻辑删除的做法，但是我的sql语句关联了4张表，每张表的逻辑删除状态有deleted（删除状态）、status（禁用启用状态）两个，一共就是4*2=8个逻辑删除字段，查询ES时需带上8个字段，状态字段太多了。这种做法实现后，花了一些时间看文档，发现可以实现mysql同步数据到ES实现删除，特此记录一下。

先说一下两种删除：

物理删除：即采用delete from table where id = 1；采用delete将数据直接从数据库删除；

逻辑删除：update table set deleted = 1（删除）where id =1；将数据标记为删除状态；

本文章实现的删除，主要是MySQL采用逻辑删除，同步到ES，ES采用物理删除。如果MySQL数据是物理删除，则logstash感知不到物理删除的数据，并不能更改到ES的数据状态。如果需要感知到MySQL的物理删除，需要使用其他数据同步方案，例如canal。

二.版本

mysql：5.6

logstash：7.6.2（版本与ES版本保持一致，6.x及以上版本支持pipeline）

Elasticsearch：7.6.2

三.安装logstash

环境：linux

mysql以及Elasticsearch的安装不在此教程；

3.1logstash安装教程

1.安装JDK1.8（版本对应需1.8），及配置JDK环境变量。

2.解压：tar -zxvf logstash-7.6.2.tar.gz

3.我这边使用的input插件是：logstash-input-jdbc；output插件是logstash-output-elasticsearch；这两个插件属于常见插件，logstash已经集成了，无需额外安装；

查看logstash已安装插件的命令：在安装目录的bin下面执行：./logstash-plugin list ；

4.使用logstash-input-jdbc输入插件，需要连接mysql驱动的jar包，将jar包放入指定目录，然后在配置文件的jdbc_driver_library 配置上对应的路径即可；

官网下载很慢，先上网盘链接，需要的自取

链接：https://pan.baidu.com/s/1z2jmU6AFQ2mV521dl7CgTg 提取码：yd8r

四.配置

logstash主要是三大块，input：数据源，filter：过滤器对数据进行过滤，output：输出；这三块都有很多插件方便使用，可以看看了解；我这里的实践是以下：

input：mysql，mysql的一行记录读取为一个事件（event）

filter：filter给数据加上@metadata标识数据的状态

output：ES

主要实现删除，是通过filter给数据加上状态，然后output里面根据状态，做出不同的响应，配置文件举个例子应该就很清楚；

测试配置文件语法的命令：在bin目录下执行：./logstash -f ../conf/main.conf -t

贴上我的sql示例，应该便于理解：

select id AS questionId, question, deleted, disabled, update_time AS updateTime from question where update_time >= :sql_last_value;

deleted 删除标志：0-正常、1-删除
disabled 禁用标志：0-正常、1-禁用
我这里是根据update_time更新时间做增量更新
注意点：
数据库里面下划线的字段比如update_time 一定要AS updateTime;
sql的列名不要使用type，type为logstash的关键字