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

我不是女神ヾ 2022-07-14 10:24 122阅读 0赞

# 引言： #

elasticsearch 的出现使得我们的存储、检索数据更快捷、方便。但很多情况下，我们的需求是：现在的数据存储在[MySQL][]、[Oracle][]等关系型传统[数据库][MySQL]中，如何尽量不改变原有数据库表结构，将这些数据的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][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][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][https_github.com_siddontang_go-mysql-elasticsearch_issues_46]

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

[http://stackoverflow.com/questions/35813923/sync-postgresql-data-with-elasticsearch/35823497\#35823497][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][http_blog.csdn.net_laoyang360_article_details_51694519] 做了些许探讨。   
除了上篇文章提到的三个插件，这里推荐试用过比较好用的logstash的一款插件，名称为： logstash-input-jdbc

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

参考：[http://blog.csdn.net/yeyuma/article/details/50240595\#quote][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
      
     
      
       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
      
     
      
       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： ###

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

# 5,注意事项 #

如果你要[测试][Link 1]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][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][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][http_dba.stackexchange.com_questions_76089_error-1236-from-master-after-restored-replication]

# 6,小结 #

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

2016-6-23 pm22:42 思于家中床前

作者：铭毅天下   
转载请标明出处，原文地址：[http://blog.csdn.net/laoyang360/article/details/51747266][http_blog.csdn.net_laoyang360_article_details_51747266]   
如果感觉本文对您有帮助，请点击‘顶’支持一下，您的支持是我坚持写作最大的动力，谢谢！

[MySQL]: http://lib.csdn.net/base/mysql
[Oracle]: http://lib.csdn.net/base/oracle
[https_discuss.elastic.co_t_delete-elasticsearch-document-with-logstash-jdbc-input_47490_9]: 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]: http://stackoverflow.com/questions/34477095/elasticsearch-replication-of-other-system-data/34477639#34477639
[Go]: http://lib.csdn.net/base/go
[https_github.com_siddontang_go-mysql-elasticsearch_issues_46]: https://github.com/siddontang/go-mysql-elasticsearch/issues/46
[http_stackoverflow.com_questions_35813923_sync-postgresql-data-with-elasticsearch_35823497_35823497]: http://stackoverflow.com/questions/35813923/sync-postgresql-data-with-elasticsearch/35823497#35823497
[http_blog.csdn.net_laoyang360_article_details_51694519]: http://blog.csdn.net/laoyang360/article/details/51694519
[http_blog.csdn.net_yeyuma_article_details_50240595_quote]: http://blog.csdn.net/yeyuma/article/details/50240595#quote
[20160623223744605]: /images/20220714/26f9dc576edd403d8b7d70a5101697de.png
[Link 1]: http://lib.csdn.net/base/softwaretest
[http_dev.mysql.com_doc_refman_5.7_en_binary-log-setting.html]: http://dev.mysql.com/doc/refman/5.7/en/binary-log-setting.html
[https_github.com_siddontang_go-mysql-elasticsearch_issues_11]: https://github.com/siddontang/go-mysql-elasticsearch/issues/11
[http_dba.stackexchange.com_questions_76089_error-1236-from-master-after-restored-replication]: http://dba.stackexchange.com/questions/76089/error-1236-from-master-after-restored-replication
[http_blog.csdn.net_laoyang360_article_details_51747266]: http://blog.csdn.net/laoyang360/article/details/51747266