MySQL 与 Elasticsearch 数据不对称问题解决办法

2023-09-25 23:10:06 60

MySQL与Elasticsearch数据不对称问题解决办法

jdbc-input-plugin只能实现数据库的追加，对于elasticsearch增量写入，但经常jdbc源一端的数据库可能会做数据库删除或者更新操作。这样一来数据库与搜索引擎的数据库就出现了不对称的情况。

当然你如果有开发团队可以写程序在删除或者更新的时候同步对搜索引擎操作。如果你没有这个能力，可以尝试下面的方法。

这里有一个数据表article,mtime字段定义了ONUPDATECURRENT_TIMESTAMP所以每次更新mtime的时间都会变化

mysql>descarticle;
+-------------+--------------+------+-----+--------------------------------+-------+
|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||ONUPDATECURRENT_TIMESTAMP||
+-------------+--------------+------+-----+--------------------------------+-------+
7rowsinset(0.00sec)

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*fromarticlewheremtime>: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'这种情况的。

CREATETABLE`elasticsearch_trash`(
`id`int(11)NOTNULL,
`ctime`timestampNULLDEFAULTCURRENT_TIMESTAMP,
PRIMARYKEY(`id`)
)ENGINE=InnoDBDEFAULTCHARSET=utf8

为article表创建触发器

CREATEDEFINER=`dba`@`%`TRIGGER`article_BEFORE_UPDATE`BEFOREUPDATEON`article`FOREACHROW
BEGIN
--此处的逻辑是解决文章状态变为N的时候，需要将搜索引擎中对应的数据删除。
IFNEW.status='N'THEN
insertintoelasticsearch_trash(id)values(OLD.id);
ENDIF;
--此处逻辑是修改状态到Y的时候，方式elasticsearch_trash仍然存在该文章ID，导致误删除。所以需要删除回收站中得回收记录。
IFNEW.status='Y'THEN
deletefromelasticsearch_trashwhereid=OLD.id;
ENDIF;
END

CREATEDEFINER=`dba`@`%`TRIGGER`article_BEFORE_DELETE`BEFOREDELETEON`article`FOREACHROW
BEGIN
--此处逻辑是文章被删除同事将改文章放入搜索引擎回收站。
insertintoelasticsearch_trash(id)values(OLD.id);
END

接下来我们需要写一个简单地Shell每分钟运行一次，从elasticsearch_trash数据表中取出数据，然后使用curl命令调用elasticsearchrestful接口，删除被收回的数据。

你还可以开发相关的程序，这里提供一个Springboot定时任务例子。

实体

packagecn.netkiller.api.domain.elasticsearch;

importjava.util.Date;

importjavax.persistence.Column;
importjavax.persistence.Entity;
importjavax.persistence.Id;
importjavax.persistence.Table;

@Entity
@Table
publicclassElasticsearchTrash{
@Id
privateintid;

@Column(columnDefinition="TIMESTAMPDEFAULTCURRENT_TIMESTAMP")
privateDatectime;

publicintgetId(){
returnid;
}

publicvoidsetId(intid){
this.id=id;
}

publicDategetCtime(){
returnctime;
}

publicvoidsetCtime(Datectime){
this.ctime=ctime;
}

}

仓库

packagecn.netkiller.api.repository.elasticsearch;

importorg.springframework.data.repository.CrudRepository;

importcom.example.api.domain.elasticsearch.ElasticsearchTrash;

publicinterfaceElasticsearchTrashRepositoryextendsCrudRepository{


}

定时任务

packagecn.netkiller.api.schedule;

importorg.elasticsearch.action.delete.DeleteResponse;
importorg.elasticsearch.client.transport.TransportClient;
importorg.elasticsearch.rest.RestStatus;
importorg.slf4j.Logger;
importorg.slf4j.LoggerFactory;
importorg.springframework.beans.factory.annotation.Autowired;
importorg.springframework.scheduling.annotation.Scheduled;
importorg.springframework.stereotype.Component;

importcom.example.api.domain.elasticsearch.ElasticsearchTrash;
importcom.example.api.repository.elasticsearch.ElasticsearchTrashRepository;

@Component
publicclassScheduledTasks{
privatestaticfinalLoggerlogger=LoggerFactory.getLogger(ScheduledTasks.class);

@Autowired
privateTransportClientclient;

@Autowired
privateElasticsearchTrashRepositoryalasticsearchTrashRepository;

publicScheduledTasks(){
}

@Scheduled(fixedRate=1000*60)//60秒运行一次调度任务
publicvoidcleanTrash(){
for(ElasticsearchTrashelasticsearchTrash:alasticsearchTrashRepository.findAll()){
DeleteResponseresponse=client.prepareDelete("information","article",elasticsearchTrash.getId()+"").get();
RestStatusstatus=response.status();
logger.info("delete{}{}",elasticsearchTrash.getId(),status.toString());
if(status==RestStatus.OK||status==RestStatus.NOT_FOUND){
alasticsearchTrashRepository.delete(elasticsearchTrash);
}
}
}
}

Springboot启动主程序。

packagecn.netkiller.api;

importorg.springframework.boot.SpringApplication;
importorg.springframework.boot.autoconfigure.SpringBootApplication;
importorg.springframework.scheduling.annotation.EnableScheduling;

@SpringBootApplication
@EnableScheduling
publicclassApplication{

publicstaticvoidmain(String[]args){
SpringApplication.run(Application.class,args);
}
}

以上就是MySQL与Elasticsearch数据不对称问题解决办法的讲解，如有疑问请留言或者到本站社区交流讨论，感谢阅读，希望能帮助到大家，谢谢大家对本站的支持！

MySQL 与 Elasticsearch 数据不对称问题解决办法

热门推荐

随机推荐