elasticSearch数据导入工具logstash-input-jdbc 同步原理及相关问题解读

本文涉及的产品
检索分析服务 Elasticsearch 版,2核4GB开发者规格 1个月
云数据库 RDS MySQL Serverless,0.5-2RCU 50GB
简介: 前言:基于logstash-input-jdbc较其他插件的稳定性、易用性、版本和ES同步更新的特点,以下研究主要针对 logstash-input-jdbc 展开。

前言:

基于logstash-input-jdbc较其他插件的稳定性、易用性、版本和ES同步更新的特点,以下研究主要针对 logstash-input-jdbc 展开。 
针对logstash-input-jdbc常见的几个疑难问题,部分问题也在git和stackoverflow进行了激烈讨论,以下统一给出验证和解答。

1、logstash-input-jdbc 的同步原理是什么?

(1)、对于全量同步依据

配置文件jdbc.sql的sql语句的进行同步。

(2)、对于增量实时同步依据

1)设定的定时策略。

如最小更新间隔每分钟更新一次设定:schedule => “* * * * *”,目前最小更新间隔为1分钟,验证发现,不支持60s以内的秒级更新。

2)设定的sql语句。

如jdbc.sql, 决定同步哪些内容及同步更新的条件。

{"id":10,"name":"10test","@version":"1","@timestamp":"2016-06-29T03:18:00.177Z","type":"132c_type"}
  • 1

2:logstash-input-jdbc 只支持基于时间同步吗?

验证表名:同步更新除了支持根据时间同步外,还支持根据某自增列(如:自增ID)字段的变化进行同步。

上次举例只是举了同步时间变化的例子,设定条件:

[root@5b9dbaaa148a logstash_jdbc_test]# cat jdbc.sql_bak

select
        *
from
        cc
where   cc.modified_at > :sql_last_value
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7

实际进一步研究发现,在配置文件中有use_column_value字段决定,是否需要记录某个column 的值,如果 record_last_run 为真,可以自定义我们需要 track 的 column 名称,此时该参数就要为 true. 否则默认 track 的是 timestamp 的值.

举例:以下即是设定以id的变化作为同步条件的。

[root@5b9dbaaa148a logstash_jdbc_test]# cat jdbc_xm.sql
select
        *
from
        cc
where   cc.id >= :sql_last_value
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6

我们可以指定文件,来记录上次执行到的 tracking_column 字段的值 比如上次数据库有 12 条记录,查询完后该文件中就会有数字 12 这样的记录,下次执行 SQL 查询可以从 13 条处开始.

我们只需要在 SQL 语句中 WHERE MY_ID > :last_sql_value 即可. 其中 :last_sql_value 取得就是该文件中的值(12).

last_run_metadata_path => “/etc/logstash/run_metadata.d/my_info”

如:

[root@5b9 run_metadata.d]# cat /etc/logstash/run_metadata.d/my_info

--- 12
  • 1
  • 2
  • 3

已全局代码搜索,没有触发器trigger相关处理操作。

3:mysql和ES分别存储在两台服务器,且时间不一致,能否实现同步?

(1). 设定对于以时间作为判定条件的增量同步,可以以设定的时间为基准点进行同步。

验证发现:

显示的时间戳timestamp为ES上的UTC时间值(不论ES机器是什么时区,都会修改为UTC时间存入ES),显示的modified_at时间值为同步过来的mysql时间值转化为UTC的结果值。

更新的前提是必须满足: cc.modified_at >= :sql_last_value。即如果mysql的时间修改为小于sql_last_value的时刻值,是无法进行同步的。

   如:
  • 1
  • 2
[elasticsearch@5b9dbaaa148a run_metadata.d]$ cat my_info

--- 2016-06-29 02:19:00.182000000 Z
  • 1
  • 2
  • 3

(2). 对于选定某列作为判定条件(如自增ID),两者(mysql和ES)时间不一致,实际是也可以同步更新的。

验证发现:

测试设定的时间是mysql比ES早一天或者晚一天的时刻值,都可以实现同步更新操作。

4:如何支持实时同步mysql的delete操作到ES中?

logstash-input-jdbc插件不支持物理删除的同步更新。详见:

http://stackoverflow.com/questions/35813923/sync-postgresql-data-with-elasticsearch/35823497#35823497

https://github.com/logstash-plugins/logstash-input-jdbc/issues/145

解决方案:

同步删除操作改为同步update更新操作实现。

第一步:进行软件删除,而不是物理删除操作。

先不物理删除记录,而是软件删除,即新增一个 flag 列,标识记录是否已经被删除(默认为false,设置为true或者deleted代表已经被删除,业界通用方法),这样,通过已有的同步机制,相同的标记记录该行数据会同步更新到Elasticsearch。

第二步:ES中检索flag标记为true或者deleted的字段信息。

在ES可以执行简单的term查询操作,检索出已经删除的数据信息。

第三步:定时物理删除。

设置定时事件,间隔一段时间物理删除掉mysql和ES中的flag字段标记为true或deleted的记录,即完成物理删除操。

原文网址:http://blog.csdn.net/laoyang360/article/details/51793301


相关实践学习
使用阿里云Elasticsearch体验信息检索加速
通过创建登录阿里云Elasticsearch集群,使用DataWorks将MySQL数据同步至Elasticsearch,体验多条件检索效果,简单展示数据同步和信息检索加速的过程和操作。
ElasticSearch 入门精讲
ElasticSearch是一个开源的、基于Lucene的、分布式、高扩展、高实时的搜索与数据分析引擎。根据DB-Engines的排名显示,Elasticsearch是最受欢迎的企业搜索引擎,其次是Apache Solr(也是基于Lucene)。 ElasticSearch的实现原理主要分为以下几个步骤: 用户将数据提交到Elastic Search 数据库中 通过分词控制器去将对应的语句分词,将其权重和分词结果一并存入数据 当用户搜索数据时候,再根据权重将结果排名、打分 将返回结果呈现给用户 Elasticsearch可以用于搜索各种文档。它提供可扩展的搜索,具有接近实时的搜索,并支持多租户。
相关文章
|
28天前
|
关系型数据库 MySQL
elasticsearch对比mysql以及使用工具同步mysql数据全量增量
elasticsearch对比mysql以及使用工具同步mysql数据全量增量
21 0
|
1月前
|
存储 搜索推荐 数据挖掘
ElasticSearch架构介绍及原理解析
ElasticSearch架构介绍及原理解析
102 0
|
1月前
|
SQL JSON Java
【Elasticsearch专栏 10】深入探索:Elasticsearch如何进行数据导入和导出
在Elasticsearch中,数据导入常通过Bulk API、Logstash或Java客户端进行,支持JSON、CSV等格式。导出则可通过SQL查询、Scroll API或第三方工具如elasticdump实现,将数据以JSON、CSV等格式导出。这些方法确保了数据的高效、安全导入与导出。
|
3月前
|
自然语言处理 API 索引
Elasticsearch Analyzer原理分析并实现中文分词
Elasticsearch Analyzer原理分析并实现中文分词
74 0
|
1月前
|
监控 Java 测试技术
【Elasticsearch专栏 13】深入探索:Elasticsearch使用Curator工具删除Elasticsearch中的历史数据
使用Curator工具可以有效管理Elasticsearch中的旧数据,通过编写YAML配置文件定义删除操作。配置中指定了基于索引名称前缀和年龄的过滤器,确保仅删除符合条件的旧索引。执行删除操作时,Curator会应用过滤器识别目标索引,并向Elasticsearch发送删除请求。通过设置选项,如忽略空列表和超时时间,可以确保操作的灵活性和稳定性。使用Curator不仅释放了存储空间,还提高了查询性能,是维护Elasticsearch健康的重要工具
|
7月前
|
JSON NoSQL 关系型数据库
Logstash同步MySql数据到Elasticsearch
Logstash同步MySql数据到Elasticsearch
118 0
|
1月前
|
SQL Java 关系型数据库
性能工具之JMeter JDBC Request 基础
JDBC 本质其实是官方(sun 公司)定义的一套操作所有关系型数据库的规则,即接口。各个数据库厂商去实现这套接口,提供数据库驱动 jar 包。我们可以使用这套接口(JDBC)编程,真正执行的代码是驱动 jar 包中的实现类。
36 0
性能工具之JMeter JDBC Request 基础
|
5月前
|
JSON 数据格式 索引
实际使用Elasticdump工具对Elasticsearch集群进行数据备份和数据还原
就可以通过Elasticsearch的导入导出工具Elasticdump来实现,可以将Elasticsearch不同集群的数据进行索引备份和还原。
100 0
|
5月前
|
存储 数据可视化 搜索推荐
分布式系列教程(26) -分布式日志搜集工具Elasticsearch简介
分布式系列教程(26) -分布式日志搜集工具Elasticsearch简介
57 0
|
3月前
|
自然语言处理 数据可视化 算法
史上最详细Docker安装Elasticsearch、ik分词器、可视化工具,每一步都带有步骤图!!!
史上最详细Docker安装Elasticsearch、ik分词器、可视化工具,每一步都带有步骤图!!!
359 0
史上最详细Docker安装Elasticsearch、ik分词器、可视化工具,每一步都带有步骤图!!!