logstash-input-jdbc實現mysql 與elasticsearch實時同步(ES與關係型數據庫同步)
引言：
elasticsearch 的出現使得咱們的存儲、檢索數據更快捷、方便。但不少狀況下，咱們的需求是：如今的數據存儲在mysql、oracle等關係型傳統數據庫中，如何儘可能不改變原有數據庫表結構，將這些數據的insert,update,delete操做結果實時同步到elasticsearch(簡稱ES)呢？
本文基於以上需求點展開實戰討論。php
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#34477639html
方案一,
在原有的mysql數據庫表中，新增一個字段status, 默認值爲ok，若是要刪除數據，實則用update操做，status改成deleted.
這樣，就能同步到es中。es中以status狀態值區分該行數據是否存在。deleted表明已刪除，ok表明正常。java
方案二，
使用go elasticsearch 插件實現同步，如：。可是我實操發現，該插件不穩定，bug較多。我也給源碼做者提出了bug。
Bug詳見：https://github.com/siddontang/go-mysql-elasticsearch/issues/46mysql
關於刪除操做的最終討論解決方案（截止2016年6月24日）：
首先，軟件刪除而非物理刪除數據，新增一個 flag 列，標識記錄是否已經被刪除，這樣，相同的記錄也會存在於Elasticsearch。能夠執行簡單的term查詢操做,檢索出已經刪除的數據信息。
其次，若須要執行cleanup清理數據操做（物理刪除），只須要在數據庫和ES中同時刪除掉標記位deleted的記錄便可。如：mysql執行：delete from cc where cc.flag=’deleted’; ES一樣執行對應刪除操做。github
2.如何使用 插件實現insert,update 的同步更新操做？
個人上一篇博文：http://blog.csdn.net/laoyang360/article/details/51694519 作了些許探討。
除了上篇文章提到的三個插件，這裏推薦試用過比較好用的logstash的一款插件，名稱爲： logstash-input-jdbcsql
3.如何安裝logstash-input-jdbc插件？
【注意啦，注意啦20170920】：logstash5.X開始，已經至少集成了logstash-input-jdbc插件。因此，你若是使用的是logstash5.X，能夠沒必要再安裝，能夠直接跳過這一步。
參考：http://blog.csdn.net/yeyuma/article/details/50240595#quote
網友博文已經介紹很詳細，再也不贅述。
基本到這一步：數據庫
cd /opt/logstash/json
sudo bin/plugin install logstash-input-jdbcruby
到此，基本就能成功。若不能請留言。
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：
驗證同步是否成功。
能夠經過： 以下圖所示： 
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
原文地址：https://blog.csdn.net/zkf541076398/article/details/79973090
- 1. 22.mysql 與elasticsearch實時同步常用插件及優缺點對比(ES與關係型數據庫同步)
- 2. elasticsearch+logstash_jdbc 實現mysql數據實時同步至es
- 3. elasticsearch與MySQL數據同步
- 4. MongoDB與Elasticsearch實現實時同步
- 5. go-mysql-elasticsearch實現mysql 與elasticsearch實時同步深刻詳解
- 6. go-mysql-elasticsearch實現mysql 與elasticsearch實時同步深入詳解
- 7. 數據同步-數據庫與數據庫之間的同步|mysql主從複製實現數據庫同步
- 8. elasticsearch與oracle數據庫數據同步
- 9. MySQL數據實時同步Elasticsearch 方案
- 10. mysql數據實時同步到Elasticsearch
- 更多相關文章...
- • 關係型數據庫是什麼？ - NoSQL教程
- • PHP 實例 - AJAX 與 XML - PHP教程
- • Spring Cloud 微服務實戰(三) - 服務註冊與發現
- • TiDB 在摩拜單車在線數據業務的應用和實踐
-
每一个你不满意的现在，都有一个你没有努力的曾经。
- 1. 22.mysql 與elasticsearch實時同步常用插件及優缺點對比(ES與關係型數據庫同步)
- 2. elasticsearch+logstash_jdbc 實現mysql數據實時同步至es
- 3. elasticsearch與MySQL數據同步
- 4. MongoDB與Elasticsearch實現實時同步
- 5. go-mysql-elasticsearch實現mysql 與elasticsearch實時同步深刻詳解
- 6. go-mysql-elasticsearch實現mysql 與elasticsearch實時同步深入詳解
- 7. 數據同步-數據庫與數據庫之間的同步|mysql主從複製實現數據庫同步
- 8. elasticsearch與oracle數據庫數據同步
- 9. MySQL數據實時同步Elasticsearch 方案
- 10. mysql數據實時同步到Elasticsearch