MySQL 与 Elasticsearch 数据不对称问题解决办法
mysql 与 elasticsearch 数据不对称问题解决办法
jdbc-input-plugin 只能实现数据库的追加,对于 elasticsearch 增量写入,但经常jdbc源一端的数据库可能会做数据库删除或者更新操作。这样一来数据库与搜索引擎的数据库就出现了不对称的情况。
当然你如果有开发团队可以写程序在删除或者更新的时候同步对搜索引擎操作。如果你没有这个能力,可以尝试下面的方法。
这里有一个数据表 article , mtime 字段定义了 on update current_timestamp 所以每次更新mtime的时间都会变化
mysql> desc article; +-------------+--------------+------+-----+--------------------------------+-------+ | field | type | null | key | default | extra | +-------------+--------------+------+-----+--------------------------------+-------+ | id | int(11) | no | | 0 | | | title | mediumtext | no | | null | | | description | mediumtext | yes | | null | | | author | varchar(100) | yes | | null | | | source | varchar(100) | yes | | null | | | content | longtext | yes | | null | | | status | enum('y','n')| no | | 'n' | | | ctime | timestamp | no | | current_timestamp | | | mtime | timestamp | yes | | on update current_timestamp | | +-------------+--------------+------+-----+--------------------------------+-------+ 7 rows in set (0.00 sec)
logstash 增加 mtime 的查询规则
jdbc { jdbc_driver_library => "/usr/share/java/mysql-connector-java.jar" jdbc_driver_class => "com.mysql.jdbc.driver" jdbc_connection_string => "jdbc:mysql://localhost:3306/cms" jdbc_user => "cms" jdbc_password => "password" schedule => "* * * * *" #定时cron的表达式,这里是每分钟执行一次 statement => "select * from article where mtime > :sql_last_value" use_column_value => true tracking_column => "mtime" tracking_column_type => "timestamp" record_last_run => true last_run_metadata_path => "/var/tmp/article-mtime.last" }
创建回收站表,这个事用于解决数据库删除,或者禁用 status = 'n' 这种情况的。
create table `elasticsearch_trash` ( `id` int(11) not null, `ctime` timestamp null default current_timestamp, primary key (`id`) ) engine=innodb default charset=utf8
为 article 表创建触发器
create definer=`dba`@`%` trigger `article_before_update` before update on `article` for each row begin -- 此处的逻辑是解决文章状态变为 n 的时候,需要将搜索引擎中对应的数据删除。 if new.status = 'n' then insert into elasticsearch_trash(id) values(old.id); end if; -- 此处逻辑是修改状态到 y 的时候,方式elasticsearch_trash仍然存在该文章id,导致误删除。所以需要删除回收站中得回收记录。 if new.status = 'y' then delete from elasticsearch_trash where id = old.id; end if; end create definer=`dba`@`%` trigger `article_before_delete` before delete on `article` for each row begin -- 此处逻辑是文章被删除同事将改文章放入搜索引擎回收站。 insert into elasticsearch_trash(id) values(old.id); end
接下来我们需要写一个简单地 shell 每分钟运行一次,从 elasticsearch_trash 数据表中取出数据,然后使用 curl 命令调用 elasticsearch restful 接口,删除被收回的数据。
你还可以开发相关的程序,这里提供一个 spring boot 定时任务例子。
实体
package cn.netkiller.api.domain.elasticsearch; import java.util.date; import javax.persistence.column; import javax.persistence.entity; import javax.persistence.id; import javax.persistence.table; @entity @table public class elasticsearchtrash { @id private int id; @column(columndefinition = "timestamp default current_timestamp") private date ctime; public int getid() { return id; } public void setid(int id) { this.id = id; } public date getctime() { return ctime; } public void setctime(date ctime) { this.ctime = ctime; } }
仓库
package cn.netkiller.api.repository.elasticsearch; import org.springframework.data.repository.crudrepository; import com.example.api.domain.elasticsearch.elasticsearchtrash; public interface elasticsearchtrashrepository extends crudrepository<elasticsearchtrash, integer>{ }
定时任务
package cn.netkiller.api.schedule; import org.elasticsearch.action.delete.deleteresponse; import org.elasticsearch.client.transport.transportclient; import org.elasticsearch.rest.reststatus; import org.slf4j.logger; import org.slf4j.loggerfactory; import org.springframework.beans.factory.annotation.autowired; import org.springframework.scheduling.annotation.scheduled; import org.springframework.stereotype.component; import com.example.api.domain.elasticsearch.elasticsearchtrash; import com.example.api.repository.elasticsearch.elasticsearchtrashrepository; @component public class scheduledtasks { private static final logger logger = loggerfactory.getlogger(scheduledtasks.class); @autowired private transportclient client; @autowired private elasticsearchtrashrepository alasticsearchtrashrepository; public scheduledtasks() { } @scheduled(fixedrate = 1000 * 60) // 60秒运行一次调度任务 public void cleantrash() { for (elasticsearchtrash elasticsearchtrash : alasticsearchtrashrepository.findall()) { deleteresponse response = client.preparedelete("information", "article", elasticsearchtrash.getid() + "").get(); reststatus status = response.status(); logger.info("delete {} {}", elasticsearchtrash.getid(), status.tostring()); if (status == reststatus.ok || status == reststatus.not_found) { alasticsearchtrashrepository.delete(elasticsearchtrash); } } } }
spring boot 启动主程序。
package cn.netkiller.api; import org.springframework.boot.springapplication; import org.springframework.boot.autoconfigure.springbootapplication; import org.springframework.scheduling.annotation.enablescheduling; @springbootapplication @enablescheduling public class application { public static void main(string[] args) { springapplication.run(application.class, args); } }
以上就是mysql 与 elasticsearch 数据不对称问题解决办法的讲解,如有疑问请留言或者到本站社区交流讨论,感谢阅读,希望能帮助到大家,谢谢大家对本站的支持!
上一篇: Unity3D基于陀螺仪实现VR相机功能
下一篇: C#WinFrom导出Excel过程解析
推荐阅读
-
使用cmd运行mysql数据库的时候,报错:"不是内部命令也不是可有运行的程序"问题的解决办法
-
MySQL 与 Elasticsearch 数据不对称问题解决办法
-
mysql修改密码、找不到mysql数据库、mysql1045等问题的解决办法
-
在PHP+Apache+MySQL环境下(windows系统),连接SQLServer数据出现“can not find driver”问题的解决办法
-
Django与MySQL数据库版本兼容问题
-
Redis与MySQL数据库一致性问题
-
数据插入到mysql数据库出现编码异常处理问题的解决办法
-
MyEclipse连接MySQL数据库出现的问题以及解决办法
-
MySql数据库的ERROR1044(42000)错误问题解决办法
-
mySQL中建表的中文字段与中文乱码问题解决办法