天天看點
傳回

logstash-input-jdbc實作mysql 與elasticsearch實時同步深入詳解引言：

引言：

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)

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

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

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：

驗證同步是否成功。

可以通過： 如下圖所示：

logstash-input-jdbc實作mysql 與elasticsearch實時同步深入詳解引言：

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
mysql 資料庫 關系型資料庫 java 資料庫連接配接 logstash jdbc實時同步 jdbc tomcat啟動mysql jdbc調用mysql mysql jdbc驅動 java資料連接配接
上一篇: 程式員，如何寫好文檔？前言
下一篇: go-mysql-elasticsearch實作mysql 與elasticsearch實時同步深入詳解引言：

繼續閱讀