一、同步原理
基于mysql的binlog日志訂閱：binlog日志是mysql用來記錄數(shù)據(jù)實(shí)時(shí)的變化
mysql數(shù)據(jù)同步到es中分為兩種，分別是全量同步和增量同步
全量同步表示第一次建立好es索引之后，將mysql中所有數(shù)據(jù)一次性導(dǎo)入到es中
增量同步表示mysql中產(chǎn)生新的數(shù)據(jù)，這些新的數(shù)據(jù)包括三種情況，就是新插入mysql中的數(shù)據(jù)，更新老的數(shù)據(jù)，刪除的數(shù)據(jù)，這些數(shù)據(jù)的變動(dòng)與新增都要同步到es中
二、logstash-input-jdbc
logstash官方插件，集成在logstash中，下載logstash即可，通過配置文件實(shí)現(xiàn)mysql與elasticsearch數(shù)據(jù)同步
優(yōu)點(diǎn)
- 能實(shí)現(xiàn)mysql數(shù)據(jù)全量和增量的數(shù)據(jù)同步，且能實(shí)現(xiàn)定時(shí)同步
- 版本更新迭代快，相對(duì)穩(wěn)定
- 作為es固有插件logstash一部分，易用
缺點(diǎn)
- 不能實(shí)現(xiàn)同步刪除操作，mysql數(shù)據(jù)刪除后elasticsearch中數(shù)據(jù)仍存在
- 同步最短時(shí)間差為一分鐘，一分鐘數(shù)據(jù)同步一次，無法做到實(shí)時(shí)同步
三、go-mysql-elasticsearch
go-mysql-elasticsearch 是國內(nèi)作者開發(fā)的一款插件
優(yōu)點(diǎn)
- 能實(shí)現(xiàn)mysql數(shù)據(jù)全量和增量的數(shù)據(jù)同步
缺點(diǎn)
- 無法實(shí)現(xiàn)數(shù)據(jù)全量同步elasticsearch
- 仍處理開發(fā)、相對(duì)不穩(wěn)定階段
四、elasticsearch-jdbc
elasticsearch-jdbc 目前最新的版本是2.3.4，支持的elasticsearch的版本為2.3.4, 未實(shí)踐
優(yōu)點(diǎn)
- 能實(shí)現(xiàn)mysql數(shù)據(jù)全量和增量的數(shù)據(jù)同步
缺點(diǎn)
- 目前最新的版本是2.3.4，支持的elasticsearch的版本為2.3.4
- 不能實(shí)現(xiàn)同步刪除操作，mysql數(shù)據(jù)刪除后elasticsearch中數(shù)據(jù)仍存在
五、logstash-input-jdbc實(shí)現(xiàn)同步
第一步安裝：
logstash5.x之后，集成了logstash-input-jdbc插件。安裝logstash后通過命令安裝logstash-input-jdbc插件
|
1
2
|
cd /logstash-6.4.2/bin./logstash-plugin install logstash-input-jdbc |
第二步配置：
在logstash-6.4.2/config文件夾下新建jdbc.conf，配置如下
在logstash-6.4.2/config 目錄下新建jdbc.sql文件
|
1
|
select * from t_employee |
第三步運(yùn)行
|
1
2
3
4
5
|
cd logstash-6.4.2# 檢查配置文件語法是否正確bin/logstash -f config/jdbc.conf --config.test_and_exit# 啟動(dòng)bin/logstash -f config/jdbc.conf --config.reload.automatic |
--config.reload.automatic：會(huì)自動(dòng)重新加載配置文件內(nèi)容
在kibana中創(chuàng)建索引后查看同步數(shù)據(jù)
|
1
2
|
put octopusget octopus/_search |
六、go-mysql-elasticsearch實(shí)現(xiàn)同步
第一步：mysql binlog日志
go-mysql-elasticsearch通過mysql中binlog日志實(shí)現(xiàn)數(shù)據(jù)增加,刪除,修改同步elasticsearch
mysql的binlog日志主要用于數(shù)據(jù)庫的主從復(fù)制與數(shù)據(jù)恢復(fù)。binlog中記錄了數(shù)據(jù)的增刪改查操作，主從復(fù)制過程中，主庫向從庫同步binlog日志，從庫對(duì)binlog日志中的事件進(jìn)行重放，從而實(shí)現(xiàn)主從同步。
mysql binlog日志有三種模式，分別為：
|
1
2
3
|
row: 記錄每一行數(shù)據(jù)被修改的情況，但是日志量太大statement: 記錄每一條修改數(shù)據(jù)的sql語句，減少了日志量，但是sql語句使用函數(shù)或觸發(fā)器時(shí)容易出現(xiàn)主從不一致mixed: 結(jié)合了row和statement的優(yōu)點(diǎn)，根據(jù)具體執(zhí)行數(shù)據(jù)操作的sql語句選擇使用row或者statement記錄日志 |
要通過mysql binlog將數(shù)據(jù)同步到es集群，只能使用row模式，因?yàn)橹挥衦ow模式才能知道m(xù)ysql中的數(shù)據(jù)的修改內(nèi)容。
以u(píng)pdate操作為例，row模式的binlog日志內(nèi)容示例如下：
|
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
|
set timestamp=1527917394/*!*/; begin /*!*/; # at 3751 #180602 13:29:54 server id 1 end_log_pos 3819 crc32 0x8dabdf01 table_map: `webservice`.`building` mapped to number 74 # at 3819 #180602 13:29:54 server id 1 end_log_pos 3949 crc32 0x59a8ed85 update_rows: table id 74 flags: stmt_end_f binlog ' uisswxmbaaaaraaaaosoaaaaaeoaaaaaaaeacndlynnlcnzpy2uacgj1awxkaw5naayidwepereg wacaaqaaaahfq40= uisswx8baaaaggaaag0paaaaaeoaaaaaaaeaagag///a1gcaaaaaaaalynvpbgrpbmctmtaadwb3 ukrnbjnlylv5d1k3ajvbd64www+ufsdwbwaaaaaaaatidwlszgluzy0xmaepahdsre1um0tivxl3 wtdqnvsprhzbd64whe2owq== '/*!*/; ### update `webservice`.`building` ### where ### @1=2006 /* longint meta=0 nullable=0 is_null=0 */ ### @2='building-10' /* varstring(192) meta=192 nullable=0 is_null=0 */ ### @3=0 /* tinyint meta=0 nullable=0 is_null=0 */ ### @4='wrdmn3kbuywy7j5' /* varstring(384) meta=384 nullable=0 is_null=0 */ ### @5=1527754262 /* timestamp(0) meta=0 nullable=0 is_null=0 */ ### @6=1527754262 /* timestamp(0) meta=0 nullable=0 is_null=0 */ ### set ### @1=2006 /* longint meta=0 nullable=0 is_null=0 */ ### @2='building-10' /* varstring(192) meta=192 nullable=0 is_null=0 */ ### @3=1 /* tinyint meta=0 nullable=0 is_null=0 */ ### @4='wrdmn3kbuywy7j5' /* varstring(384) meta=384 nullable=0 is_null=0 */ ### @5=1527754262 /* timestamp(0) meta=0 nullable=0 is_null=0 */ ### @6=1527754262 /* timestamp(0) meta=0 nullable=0 is_null=0 */ # at 3949 #180602 13:29:54 server id 1 end_log_pos 3980 crc32 0x58226b8f xid = 182 commit/*!*/; |
statement模式下binlog日志內(nèi)容示例為：
|
1
2
3
4
5
6
|
set timestamp=1527919329/*!*/; update building set status=1 where id=2000 /*!*/; # at 688 #180602 14:02:09 server id 1 end_log_pos 719 crc32 0x4c550a7d xid = 200 commit/*!*/; |
從row模式和statement模式下update操作的日志內(nèi)容可以看出，row模式完整地記錄了要修改的某行數(shù)據(jù)更新前的所有字段的值以及更改后所有字段的值，而statement模式只單單記錄了update操作的sql語句。我們要將mysql的數(shù)據(jù)實(shí)時(shí)同步到es， 只能選擇row模式的binlog, 獲取并解析binlog日志的數(shù)據(jù)內(nèi)容，執(zhí)行es document api，將數(shù)據(jù)同步到es集群中。
查看，修改binlog模式
|
1
2
3
4
5
6
7
8
9
10
11
|
# 查看binlog模式mysql> show variables like "%binlog_format%"; # 修改binlog模式mysql> set global binlog_format='row'; # 查看binlog是否開啟mysql> show variables like 'log_bin'; # 開啟bīnlog修改my.cnf文件log-bin = mysql-bin |
第二步安裝
|
1
2
3
4
5
6
7
8
9
10
11
12
|
# 安裝gosudo apt-get install go # 安裝godepgo get github.com/tools/godep # 獲取go-mysql-elasticsearch插件go get github.com/siddontang/go-mysql-elasticsearch # 安裝go-mysql-elasticsearch插件cd go/src/github.com/siddontang/go-mysql-elasticsearchmake |
第三步配置
go/src/github.com/siddontang/go-mysql-elasticsearch/etc/river.toml
|
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
55
56
57
58
59
|
# mysql address, user and password# user must have replication privilege in mysql.my_addr = "127.0.0.1:3306" # 需要同步的mysql基本設(shè)置my_user = "root"my_pass = "root" # elasticsearch addresses_addr = "127.0.0.1:9200" # 本地elasticsearch配置 # path to store data, like master.info, and dump mysql data data_dir = "./var" # 數(shù)據(jù)存儲(chǔ)的url# 以下配置保存默認(rèn)不變# inner http status addressstat_addr = "127.0.0.1:12800" # pseudo server id like a slave server_id = 1001 # mysql or mariadbflavor = "mysql"# mysqldump execution pathmysqldump = "mysqldump" # mysql data source[[source]]schema = "test" //elasticsearch 與 mysql 同步時(shí)對(duì)應(yīng)的數(shù)據(jù)庫名稱 # only below tables will be synced into elasticsearch.# 要同步test這個(gè)database里面的幾張表。對(duì)于一些項(xiàng)目如果使用了分表機(jī)制，我們可以用通配符來匹配，譬如t_[0-9]{4}，就可# 以匹配 table t_0000 到 t_9999。tables = ["t", "t_[0-9]{4}", "tfield", "tfilter"] # below is for special rule mapping# 對(duì)一個(gè) table，我們需要指定將它的數(shù)據(jù)同步到 es 的哪一個(gè) index 的 type 里面。如果不指定，我們默認(rèn)會(huì)用起 schema # name 作為 es 的 index 和 type[[rule]]schema = "test" //數(shù)據(jù)庫名稱table = "t" //表名稱index = "test" //對(duì)應(yīng)的索引名稱type = "t" //對(duì)應(yīng)的類型名稱 # 將所有滿足格式 t_[0-9]{4} 的 table 同步到 es 的 index 為 test，type 為 t 的下面。當(dāng)然，這些表需要保證# schema 是一致的[[rule]]schema = "test"table = "t_[0-9]{4}"index = "test"type = "t" # 對(duì)于 table tfilter，我們只會(huì)同步 id 和 name 這兩列，其他的都不會(huì)同步filter = ["id", "name"]# table tfield 的 column id ，我們映射成了 es_id，而 tags 則映射成了 es_tags# list 這個(gè)字段，他顯示的告知需要將對(duì)應(yīng)的 column 數(shù)據(jù)轉(zhuǎn)成 es 的 array type。這個(gè)現(xiàn)在通常用于 mysql 的 varchar # 等類型，我們可能會(huì)存放類似 “a,b,c” 這樣的數(shù)據(jù)，然后希望同步給 es 的時(shí)候變成 [a, b, c] 這樣的列表形式。 [rule.field]# map column `id` to es field `es_id`id="es_id"# map column `tags` to es field `es_tags` with array type tags="es_tags,list"# map column `keywords` to es with array typekeywords=",list" |
第四步運(yùn)行
|
1
2
|
cd go/src/github.com/siddontang/go-mysql-elasticsearchbin/go-mysql-elasticsearch -config=./etc/river.toml |
七、elasticsearch-jdbc實(shí)現(xiàn)同步
下載工具
解壓：unzip elasticsearch-jdbc-2.3.2.0-dist.zip
設(shè)置環(huán)境變量
|
1
2
|
[root@autofelix /]# vi /etc/profileexport jdbc_importer_home=/elasticsearch-jdbc-2.3.2.0 |
使環(huán)境變量生效
|
1
|
[root@autofelix /]# source /etc/profile |
配置參考
第一步：在根目錄下建立根目錄下新建文件夾odbc_es 如下
|
1
2
3
|
[root@autofelix /]# ll /odbc_es/drwxr-xr-x 2 root root 4096 jun 16 03:11 logs-rwxrwxrwx 1 root root 542 jun 16 04:03 mysql_import_es.sh |
第二步：新建腳本mysql_import_es.sh，內(nèi)容如下
|
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
|
[root@autofelix odbc_es]# cat mysql_import_es.sh'#!/bin/shbin=$jdbc_importer_home/binlib=$jdbc_importer_home/libecho '{"type" : "jdbc","jdbc": {"elasticsearch.autodiscover":true,"elasticsearch.cluster":"my-application", #簇名，詳見：/usr/local/elasticsearch/config/elasticsearch.yml"url":"jdbc:mysql://10.8.5.101:3306/test", #mysql數(shù)據(jù)庫地址"user":"root", #mysql用戶名"password":"123456", #mysql密碼"sql":"select * from cc","elasticsearch" : { "host" : "10.8.5.101", "port" : 9300},"index" : "myindex", #新的index"type" : "mytype" #新的type}}'| java \ -cp "${lib}/*" \ -dlog4j.configurationfile=${bin}/log4j2.xml \ org.xbib.tools.runner \ org.xbib.tools.jdbcimporter |
第三步：為 mysql_import_es.sh 添加可執(zhí)行權(quán)限。
|
1
|
[root@autofelix odbc_es]# chmod a+x mysql_import_es.sh |
第四步：執(zhí)行腳本mysql_import_es.sh
|
1
|
[root@autofelix odbc_es]# ./mysql_import_es.sh |
到此這篇關(guān)于詳解mysql如何實(shí)現(xiàn)數(shù)據(jù)同步到elasticsearch的文章就介紹到這了,更多相關(guān)mysq數(shù)據(jù)同步到elasticsearch內(nèi)容請(qǐng)搜索服務(wù)器之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持服務(wù)器之家！
原文鏈接：https://blog.csdn.net/weixin_41635750/article/details/121445723
