AI 助理
备案控制台
开发者社区 数据库 文章 正文

logstash-input-jdbc实现mysql 与elasticsearch实时同步深入详解

2017-11-05 2636
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
本文涉及的产品
云数据库 RDS MySQL，集群系列 2核4GB
推荐场景：
搭建个人博客
RDS MySQL Serverless 基础系列，0.5-2RCU 50GB
推荐场景：
学生管理系统数据库设计 搭建个人博客
检索分析服务 Elasticsearch 版，2核4GB开发者规格 1个月
推荐场景：
数据可视化分析航班信息
简介： 引言：elasticsearch 的出现使得我们的存储、检索数据更快捷、方便。但很多情况下，我们的需求是：现在的数据存储在mysql、oracle等关系型传统数据库中，如何尽量不改变原有数据库表结构，将这些数据的insert,update,delete操作结果实时同步到elasticsearch(简称ES)呢？ 本文基于以上需求点展开实战讨论。

引言：

elasticsearch 的出现使得我们的存储、检索数据更快捷、方便。但很多情况下，我们的需求是：现在的数据存储在mysql、oracle等关系型传统数据库中，如何尽量不改变原有数据库表结构，将这些数据的insert,update,delete操作结果实时同步到elasticsearch(简称ES)呢？ 
本文基于以上需求点展开实战讨论。

1.对delete操作的实时同步泼冷水

到目前为止，所有google,stackoverflow,elastic.co,github上面搜索的插件和实时同步的信息，告诉我们：目前同步delete还没有好的解决方案。 
折中的解决方案如下： 
方案探讨：https://discuss.elastic.co/t/delete-elasticsearch-document-with-logstash-jdbc-input/47490/9 
http://stackoverflow.com/questions/34477095/elasticsearch-replication-of-other-system-data/34477639#34477639

方案一,

在原有的mysql数据库表中，新增一个字段status, 默认值为ok，如果要删除数据，实则用update操作，status改为deleted. 
这样，就能同步到es中。es中以status状态值区分该行数据是否存在。deleted代表已删除，ok代表正常。

方案二，

使用go elasticsearch 插件实现同步，如：。但是我实操发现，该插件不稳定，bug较多。我也给源码作者提出了bug。 
Bug详见：https://github.com/siddontang/go-mysql-elasticsearch/issues/46

关于删除操作的最终讨论解决方案（截止2016年6月24日）：

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

首先，软件删除而非物理删除数据，新增一个 flag 列，标识记录是否已经被删除，这样，相同的记录也会存在于Elasticsearch。可以执行简单的term查询操作,检索出已经删除的数据信息。 
其次，若需要执行cleanup清理数据操作（物理删除），只需要在数据库和ES中同时删除掉标记位deleted的记录即可。如：mysql执行：delete from cc where cc.flag=’deleted’; ES同样执行对应删除操作。

2.如何使用 插件实现insert,update 的同步更新操作？

我的上一篇博文：http://blog.csdn.net/laoyang360/article/details/51694519 做了些许探讨。 
除了上篇文章提到的三个插件，这里推荐试用过比较好用的logstash的一款插件，名称为： logstash-input-jdbc

3.如何安装logstash-input-jdbc插件？

【注意啦，注意啦20170920】：logstash5.X开始，已经至少集成了logstash-input-jdbc插件。所以，你如果使用的是logstash5.X，可以不必再安装，可以直接跳过这一步。 
参考：http://blog.csdn.net/yeyuma/article/details/50240595#quote 
网友博文已经介绍很详细，不再赘述。 
基本到这一步：

cd /opt/logstash/

sudo bin/plugin install logstash-input-jdbc

到此，基本就能成功。若不能请留言。

4,如何实现实时同步？

4.1 前提：mysql存在的数据库及表

数据库名为:test 
test下表名为:cc 
表中数据为：

mysql> use test;
Reading table information for completion of table and column names
You can turn off this feature to get a quicker startup with -A

Database changed
mysql> select * from cc;
+----+--------------------+---------+---------------------+
| id | name | status | modified_at |
+----+--------------------+---------+---------------------+
| 1 | laoyang360 | ok | 0000-00-00 00:00:00 |
| 2 | test002 | ok | 2016-06-23 06:16:42 |
| 3 | dllaoyang | ok | 0000-00-00 00:00:00 |
| 4 | huawei | ok | 0000-00-00 00:00:00 |
| 5 | jdbc_test_update08 | ok | 0000-00-00 00:00:00 |
| 7 | test7 | ok | 0000-00-00 00:00:00 |
| 8 | test008 | ok | 0000-00-00 00:00:00 |
| 9 | test9 | ok | 0000-00-00 00:00:00 |
| 10 | test10 | deleted | 0000-00-00 00:00:00 |
+----+--------------------+---------+---------------------+
9 rows in set (0.01 sec)
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20

4.2 需要两个文件：1）jdbc.conf; 2)jdbc.sql.

[root@5b9dbaaa148a logstash_jdbc_test]# cat jdbc.conf
input {
  stdin {
  }
  jdbc {
  # mysql jdbc connection string to our backup databse  后面的test对应mysql中的test数据库
  jdbc_connection_string => "jdbc:mysql://192.168.1.1:3306/test"
  # the user we wish to excute our statement as
  jdbc_user => "root"
  jdbc_password => "******"
  # the path to our downloaded jdbc driver
  jdbc_driver_library => "/elasticsearch-jdbc-2.3.2.0/lib/mysql-connector-java-5.1.38.jar"
  # the name of the driver class for mysql
  jdbc_driver_class => "com.mysql.jdbc.Driver"
  jdbc_paging_enabled => "true"
  jdbc_page_size => "50000"
#以下对应着要执行的sql的绝对路径。
  statement_filepath => "/usr/local/logstash/bin/logstash_jdbc_test/jdbc.sql"
#定时字段 各字段含义（由左至右）分、时、天、月、年，全部为*默认含义为每分钟都更新（测试结果，不同的话请留言指出）
  schedule => "* * * * *"
#设定ES索引类型
  type => "cc_type"
  }
}

filter {
  json {
  source => "message"
  remove_field => ["message"]
  }
}

output {
  elasticsearch {
#ESIP地址与端口
  hosts => "192.168.1.1:9200"
#ES索引名称（自己定义的）
  index => "cc_index"
#自增ID编号
  document_id => "%{id}"
  }
  stdout {
#以JSON格式输出
  codec => json_lines
  }
}

#要执行的sql语句。
选择哪些信息同步到ES中。
[root@5b9dbaaa148a logstash_jdbc_test]# cat jdbc.sql
select
  *
from
where cc.modified_at > :sql_last_value
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54

[注意啦！注意啦！注意啦！]

cc.modified_at, 这个modified_at是我自己定义的更改时间字段，默认值default是now()当前时间。 
而 :sql_last_value如果input里面use_column_value => true， 即如果设置为true的话，可以是我们设定的字段的上一次的值。 
默认 use_column_value => false， 这样 :sql_last_value为上一次更新的最后时刻值。 
也就是说，对于新增的值，才会更新。这样就实现了增量更新的目的。

有童鞋问，如何全量更新呢？ 答案：就是去掉where子句即可。

步骤1：

在logstash的bin路径下新建文件夹logstash_jdbc_test，并将上两个文件 1）jdbc.conf，2)jdbc.sql.模板拷贝到里面。

步骤2：

按照自己的mysql地址、es地址、建立的索引名称、类型名称修改conf，以及要同步内容修改sql。

步骤3：

执行logstash, 如下： 
[root@5b9dbaaa148a plugins]# ./logstash -f ./logstash_jdbc_test/jdbc.conf

步骤4：

验证同步是否成功。 
可以通过： 如下图所示： 
这里写图片描述

5,注意事项

如果你要测试go-mysql-elasticsearch可能会遇到下面三个Bug及解决方案如下：

【Bug1】

How to Setting The Binary Log Format 
http://dev.mysql.com/doc/refman/5.7/en/binary-log-setting.html

【Bug2】

what is inner http status address 
https://github.com/siddontang/go-mysql-elasticsearch/issues/11

【Bug3】

[2016/06/23 10:19:38] canal.go:146 [Error] canal start sync binlog err: ERROR 1236 (HY000): Misconfigured master - server id was not set 
http://dba.stackexchange.com/questions/76089/error-1236-from-master-after-restored-replication

6,小结

实操发现： logstash-input-jdbc 能较好的实现mysql的insert、update的操作的增量、全量数据同步更新到ES。 
但delete操作的实时同步没有很好的解决方案，如果你有，且都测试ok的话，请留言告诉我，不吝赐教！

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

文章标签：
检索分析服务 Elasticsearch版
云数据库 RDS MySQL 版
测试技术
SQL
关系型数据库
索引
数据库
关键词：
检索分析服务 Elasticsearch版mysql
实现云数据库 RDS MySQL 版elasticsearch实时同步
云数据库 RDS MySQL 版elasticsearch实时同步
相关实践学习
使用阿里云Elasticsearch体验信息检索加速
通过创建登录阿里云Elasticsearch集群，使用DataWorks将MySQL数据同步至Elasticsearch，体验多条件检索效果，简单展示数据同步和信息检索加速的过程和操作。
ElasticSearch 入门精讲
ElasticSearch是一个开源的、基于Lucene的、分布式、高扩展、高实时的搜索与数据分析引擎。根据DB-Engines的排名显示，Elasticsearch是最受欢迎的企业搜索引擎，其次是Apache Solr（也是基于Lucene）。 ElasticSearch的实现原理主要分为以下几个步骤： 用户将数据提交到Elastic Search 数据库中 通过分词控制器去将对应的语句分词，将其权重和分词结果一并存入数据 当用户搜索数据时候，再根据权重将结果排名、打分 将返回结果呈现给用户 Elasticsearch可以用于搜索各种文档。它提供可扩展的搜索，具有接近实时的搜索，并支持多租户。
尘光掠影
目录
相关文章
阿里云瑶池数据库SelectDB
|
1月前
|
SQL DataWorks 关系型数据库
阿里云 DataWorks 正式支持 SelectDB & Apache Doris 数据源，实现 MySQL 整库实时同步
阿里云数据库 SelectDB 版是阿里云与飞轮科技联合基于 Apache Doris 内核打造的现代化数据仓库，支持大规模实时数据上的极速查询分析。通过实时、统一、弹性、开放的核心能力，能够为企业提供高性价比、简单易用、安全稳定、低成本的实时大数据分析支持。SelectDB 具备世界领先的实时分析能力，能够实现秒级的数据实时导入与同步，在宽表、复杂多表关联、高并发点查等不同场景下，提供超越一众国际知名的同类产品的优秀性能，多次登顶 ClickBench 全球数据库分析性能排行榜。
阿里云瑶池数据库SelectDB
86 6
糜终
|
2月前
|
关系型数据库 MySQL API
MySQL 历史数据迁移到 Elasticsearch
MySQL 历史数据迁移到 Elasticsearch
糜终
89 4
qhhf_地道酬勤
|
3月前
|
NoSQL 关系型数据库 Redis
mall在linux环境下的部署（基于Docker容器），Docker安装mysql、redis、nginx、rabbitmq、elasticsearch、logstash、kibana、mongo
mall在linux环境下的部署（基于Docker容器），docker安装mysql、redis、nginx、rabbitmq、elasticsearch、logstash、kibana、mongodb、minio详细教程，拉取镜像、运行容器
qhhf_地道酬勤
571 3
mall在linux环境下的部署（基于Docker容器），Docker安装mysql、redis、nginx、rabbitmq、elasticsearch、logstash、kibana、mongo
最好zzz
|
2月前
|
消息中间件 监控 关系型数据库
MySQL数据实时同步到Elasticsearch：技术深度解析与实践分享
在当今的数据驱动时代，实时数据同步成为许多应用系统的核心需求之一。MySQL作为关系型数据库的代表，以其强大的事务处理能力和数据完整性保障，广泛应用于各种业务场景中。然而，随着数据量的增长和查询复杂度的提升，单一依赖MySQL进行高效的数据检索和分析变得日益困难。这时，Elasticsearch（简称ES）以其卓越的搜索性能、灵活的数据模式以及强大的可扩展性，成为处理复杂查询需求的理想选择。本文将深入探讨MySQL数据实时同步到Elasticsearch的技术实现与最佳实践。
最好zzz
138 0
三分钟热度的鱼
|
4月前
|
SQL 关系型数据库 MySQL
实时计算 Flink版产品使用问题之MySQL到MySOL的批量实时同步该如何操作
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架，广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器，提供了一套全面的解决方案，以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点，使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
三分钟热度的鱼
50 0
三分钟热度的鱼
|
6月前
|
Java 关系型数据库 MySQL
实时计算 Flink版产品使用问题之如何实现MySQL数据库之间的实时同步
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架，广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器，提供了一套全面的解决方案，以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点，使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
三分钟热度的鱼
95 7
三分钟热度的鱼
|
5月前
|
关系型数据库 MySQL 数据处理
实时计算 Flink版产品使用问题之任务无法实时同步MySQL到StarRocks中修改的数据，是什么原因
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架，广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器，提供了一套全面的解决方案，以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点，使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
三分钟热度的鱼
118 0
G佳伟
|
5月前
|
存储 关系型数据库 MySQL
【Elasticsearch】在es中实现mysql中的FIND_IN_SET查询条件
【Elasticsearch】在es中实现mysql中的FIND_IN_SET查询条件
G佳伟
134 0
冲冲冲c
|
6月前
|
DataWorks 安全 关系型数据库
DataWorks产品使用合集之如何进行mysql的实时同步
DataWorks作为一站式的数据开发与治理平台，提供了从数据采集、清洗、开发、调度、服务化、质量监控到安全管理的全套解决方案，帮助企业构建高效、规范、安全的大数据处理体系。以下是对DataWorks产品使用合集的概述，涵盖数据处理的各个环节。z
冲冲冲c
174 0
三分钟热度的鱼
|
7月前
|
Kubernetes 关系型数据库 MySQL
实时计算 Flink版产品使用合集之在Kubernetes(k8s)中同步MySQL变更到Elasticsearch该怎么操作
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架，广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStreamAPI、以及与上下游数据源和存储系统的丰富连接器，提供了一套全面的解决方案，以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点，使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
三分钟热度的鱼
91 1

热门文章

最新文章

  • 1
    RDS入门——Excel文件转存到RDS数据库实践
  • 2
    PostgreSQL SQL HINT的使用
  • 3
    mysql 同表 父子关系查询 【亲测可用】
  • 4
    MySQL基本数据库管理命令
  • 5
    mysql互为主从的环境，更新一条语句同时提交，为什么会出现数据不一致？
  • 6
    MySQL入门实战3：SQL问题诊断与调优
  • 7
    Mysql数据库初始化后更改数据存储目录
  • 8
    破解并重置mysql密码
  • 9
    MySQL创建数据表(CREATE TABLE语句)
  • 10
    mysql安装过程
  • 1
    分布式搜索引擎ElasticSearch
    276
  • 2
    Elasticsearch框架学习的难点和重点有哪些
    71
  • 3
    springboot集成ElasticSearch的具体操作（系统全文检索）
    92
  • 4
    Java操作elasticsearch
    46
  • 5
    window安装配置ElasticSearch
    45
  • 6
    5个 Elasticsearch 核心组件
    102
  • 7
    深入理解Elasticsearch倒排索引原理与优化策略
    644
  • 8
    Elasticsearch FSCrawler 一个bug及解决方案
    101
  • 9
    Elasticsearch 配置文件 path.data 中可以配置多个数据目录的路径吗？
    174
  • 10
    Elasticsearch 8.X 向量检索和普通检索能否实现组合检索？如何实现？
    169

    • 相关课程

    更多
  • MySQL企业常见架构与调优经验分享
  • 云数据库MySQL版快速上手教程
  • 阿里云云原生数据仓库AnalyticDB MySQL版 使用教程
  • MySQL实战进阶
  • 数据库及SQL/MySQL基础
  • 云数据库MySQL快速入门

    • 相关电子书

    更多
  • 搭建电商项目架构连接MySQL
  • 搭建4层电商项目架构，实战连接MySQL
  • PolarDB MySQL引擎重磅功能及产品能力盛大发布

    • 相关实验场景

    更多
  • MySQL基础-学生管理系统数据库设计
  • 如何快速连接云数据库RDS MySQL
  • MySQL数据库快速部署实践
  • MySQL引擎及架构优化
  • 基于EBS部署高性能的MySQL服务
  • 使用CloudLens采集RDS日志并进行审计分析

    • 推荐镜像

    更多
  • mysql
  • mongodb
  • postgresql
    • 下一篇
    【阿里云OSS】You have no right to access this object because of bucket acl.