文档章节

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

netkiller-
 netkiller-
发布于 2017/08/22 15:23
字数 923
阅读 6.5K
收藏 110

行业解决方案、产品招募中!想赚钱就来传!>>>

本文节选自《Netkiller Database 手札》

作者:netkiller 网站:http://www.netkiller.cn

23.10.7. 解决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);
	}
}

 

netkiller-

netkiller-

粉丝 716
博文 274
码字总数 379578
作品 10
深圳
部门经理
私信 提问
加载中
此博客有 5 条评论,请先登录后再查看。
访问安全控制解决方案

本文是《轻量级 Java Web 框架架构设计》的系列博文。 今天想和大家简单的分享一下,在 Smart 中是如何做到访问安全控制的。也就是说,当没有登录或 Session 过期时所做的操作,会自动退回到...

黄勇
2013/11/03
3.5K
6
SQLServer实现split分割字符串到列

网上已有人实现sqlserver的split函数可将字符串分割成行,但是我们习惯了split返回数组或者列表,因此这里对其做一些改动,最终实现也许不尽如意,但是也能解决一些问题。 先贴上某大牛写的s...

cwalet
2014/05/21
9.6K
0
我的架构演化笔记 功能1: 基本的用户注册

“咚咚”,一阵急促的敲门声, 我从睡梦中惊醒,我靠,这才几点,谁这么早, 开门一看,原来我的小表弟放暑假了,来南京玩,顺便说跟我后面学习一个网站是怎么做出来的。 于是有了下面的一段...

强子哥哥
2014/05/31
976
3
Nutch学习笔记4-Nutch 1.7 的 索引篇 ElasticSearch

上一篇讲解了爬取和分析的流程,很重要的收获就是: 解析过程中,会根据页面的ContentType获得一系列的注册解析器, 依次调用每个解析器,当其中一个解析成功后就返回,否则继续执行下一个解...

强子哥哥
2014/06/26
712
0
数据库代码辅助工具--MaoCaiJun.Database

MaoCaiJun.DataBase 是一个用于 Microsoft Visual Studio 的数据库代码生成组件。它是基于 xml 文件的代码创建工具,支持sql2000,sql2005,sql2008,access, SQLite MaoCaiJun.Database 数据库...

mccj
2013/02/06
2.3K
1

没有更多内容

加载失败,请刷新页面

加载更多

Azure Application Gateway(一)对后端 Web App 进行负载均衡

一,引言   今天,我们学习一个新的知识点-----Azure Application Gateway,通过Azure 应用程序网关为我么后端的服务提供负载均衡的功能。我们再文章头中大概先了解一下什么是应用程序网关...

osc_lc4icfkt
27分钟前
4
0
WoLai(我来) 注册码 ——国内版 notion 【笔记】

注册码: SQGYG23 ❤ W4T9PKP JLTHNJP KMTXK7P JDHKJEM KRJXX5P 6M7PPAP DEGLMG3 N3BZMRI 87BR22I GSIWGWP GNGBNTI QA8URIM UDUV9VM IHKJA7P MD9ZA3M 3XR67ZI TBUP9JX TI4TYMM 注册完了可以把......

osc_c05lkk3u
29分钟前
15
0
2020hdu多校第二场比赛及补题

这一场我们队只A了一题 1010 Lead of Wisdom 直接爆搜,但是T了好几发,剪了下枝 如果一个物品的a,b,c,d都比不上另外一个同种物品的a,b,c,d,那这个物品就可以直接淘汰掉了 #include<iostrea...

osc_usgpahnw
30分钟前
21
0
为什么Java有瞬态字段? - Why does Java have transient fields?

问题: 为什么Java有瞬态字段? 解决方案: 参考一: https://stackoom.com/question/3opS/为什么Java有瞬态字段 参考二: https://oldbug.net/q/3opS/Why-does-Java-have-transient-fields...

富含淀粉
30分钟前
16
0
轻松搭建CAS 5.x系列(6)-在CAS Server上增加OAuth2.0协议

概述说明 CAS Server默认搭建出来,客户端程序只能按照CAS自身的协议接入。CAS的强大在于,有官方的插件,可以支持其他的协议。本章节就让CAS Server怎么增加OAuth2.0的登录协议。 安装步骤 ...

osc_inj0cicw
31分钟前
7
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部