MySQL 與 Elasticsearch 數(shù)據(jù)不對(duì)稱問(wèn)題解決辦法
jdbc-input-plugin 只能實(shí)現(xiàn)數(shù)據(jù)庫(kù)的追加，對(duì)于 elasticsearch 增量寫入，但經(jīng)常jdbc源一端的數(shù)據(jù)庫(kù)可能會(huì)做數(shù)據(jù)庫(kù)刪除或者更新操作。這樣一來(lái)數(shù)據(jù)庫(kù)與搜索引擎的數(shù)據(jù)庫(kù)就出現(xiàn)了不對(duì)稱的情況。
當(dāng)然你如果有開(kāi)發(fā)團(tuán)隊(duì)可以寫程序在刪除或者更新的時(shí)候同步對(duì)搜索引擎操作。如果你沒(méi)有這個(gè)能力，可以嘗試下面的方法。
這里有一個(gè)數(shù)據(jù)表 article , mtime 字段定義了 ON UPDATE CURRENT_TIMESTAMP 所以每次更新mtime的時(shí)間都會(huì)變化
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 的查詢規(guī)則
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 => "* * * * *" #定時(shí)cron的表達(dá)式,這里是每分鐘執(zhí)行一次
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"
}
創(chuàng)建回收站表，這個(gè)事用于解決數(shù)據(jù)庫(kù)刪除，或者禁用 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 表創(chuàng)建觸發(fā)器
CREATE DEFINER=`dba`@`%` TRIGGER `article_BEFORE_UPDATE` BEFORE UPDATE ON `article` FOR EACH ROW
BEGIN
-- 此處的邏輯是解決文章?tīng)顟B(tài)變?yōu)?N 的時(shí)候，需要將搜索引擎中對(duì)應(yīng)的數(shù)據(jù)刪除。
IF NEW.status = 'N' THEN
insert into elasticsearch_trash(id) values(OLD.id);
END IF;
-- 此處邏輯是修改狀態(tài)到 Y 的時(shí)候，方式elasticsearch_trash仍然存在該文章ID，導(dǎo)致誤刪除。所以需要?jiǎng)h除回收站中得回收記錄。
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
接下來(lái)我們需要寫一個(gè)簡(jiǎn)單地 Shell 每分鐘運(yùn)行一次，從 elasticsearch_trash 數(shù)據(jù)表中取出數(shù)據(jù)，然后使用 curl 命令調(diào)用 elasticsearch restful 接口，刪除被收回的數(shù)據(jù)。
你還可以開(kāi)發(fā)相關(guān)的程序，這里提供一個(gè) Spring boot 定時(shí)任務(wù)例子。
實(shí)體
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;
}
}
倉(cāng)庫(kù)
package cn.netkiller.api.repository.elasticsearch;
import org.springframework.data.repository.CrudRepository;
import com.example.api.domain.elasticsearch.ElasticsearchTrash;
public interface ElasticsearchTrashRepository extends CrudRepositoryElasticsearchTrash, Integer>{
}
定時(shí)任務(wù)
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秒運(yùn)行一次調(diào)度任務(wù)
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 啟動(dòng)主程序。
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 數(shù)據(jù)不對(duì)稱問(wèn)題解決辦法的講解，如有疑問(wèn)請(qǐng)留言或者到本站社區(qū)交流討論，感謝閱讀，希望能幫助到大家，謝謝大家對(duì)本站的支持！
您可能感興趣的文章:- windows下MySQL5.6版本安裝及配置過(guò)程附有截圖和詳細(xì)說(shuō)明
- Mysql字符串截取函數(shù)SUBSTRING的用法說(shuō)明
- MySQL日期數(shù)據(jù)類型、時(shí)間類型使用總結(jié)
- MySQL創(chuàng)建用戶與授權(quán)方法
- MySQL 的CASE WHEN 語(yǔ)句使用說(shuō)明
- mysql update語(yǔ)句的用法詳解
- MySQL提示:The server quit without updating PID file問(wèn)題的解決辦法
