21.go-mysql-elasticsearch实现mysql 与elasticsearch实时同步(ES与关系型数据库同步)
赵智
引言：
go-mysql-elasticsearch 是国内作者开发的一款插件。测试表明：该插件优点：能实现同步增、删、改、查操作。不足之处（待完善的地方）：
1、仍处理开发、相对不稳定阶段；
2、没有日志，不便于排查问题及查看同步结果。
本文深入详解了插件的安装、使用、增删改查同步测试。
1. go-mysql-elasticsearch 插件安装
步骤1：安装go
yum install go
步骤2：安装godep
go get github.com/tools/godep
步骤3：获取go-mysql-elastisearch插件
go get github.com/siddontang/go-mysql-elasticsearch
步骤4：安装go-mysql-elastisearch插件
cd $GOPATH/src/github.com/siddontang/go-mysql-elasticsearch
make
2.go-mysql-elasticsearch 插件使用
2.1修改配置文件
[root@5b9dbaaa148a etc]# cat river.toml
# MySQL address, user and password
# user must have replication privilege in MySQL.
my_addr = "192.168.1.1:3306"
my_user = "root"
my_pass = "password@!"
# Elasticsearch address
es_addr = "192.168.1.1:9200"
# Path to store data, like master.info, and dump MySQL data
data_dir = "./var"
# Inner Http status address
stat_addr = "192.168.1.1:12800"
# pseudo server id like a slave
server_id = 1
# mysql or mariadb
flavor = "mysql"
# mysqldump execution path
mysqldump = "mysqldump"
# MySQL data source
[[source]]
schema = "test"
# Only below tables will be synced into Elasticsearch.
# "test_river_[0-9]{4}" is a wildcard table format, you can use it if you have many sub tables, like table_0000 - table_1023
# I don't think it is necessary to sync all tables in a database.
tables = ["cc"]
# Below is for special rule mapping
#[[rule]]
#schema = "test"
#table = "cc"
#index = "go_river"
#type = "go_rivert"
# title is MySQL test_river field name, es_title is the customized name in Elasticsearch
# [rule.field]
# This will map column title to elastic search my_title
# title="es_title"
# This will map column tags to elastic search my_tags and use array type
# tags="my_tags,list"
# This will map column keywords to elastic search keywords and use array type
#keywords=",list"
# wildcard table rule, the wildcard table must be in source tables
[[rule]]
schema = "test"
table = "cc"
index = "gocc"
type = "gocc_t"
# title is MySQL test_river field name, es_title is the customized name in Elasticsearch
[[rule.fields]]
mysql = "mysql101"
elastic = "es_mysql101"- 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
- 60
- 61
2.2执行同步操作
cd $GOPATH/src/github.com/siddontang/go-mysql-elasticsearch
./bin/go-mysql-elasticsearch -config=./etc/river.toml
3. go-mysql-elasticsearch 插件同步测试结果
3.1插入Insert操作实时同步验证（验证ok）
3.1.1Mysql端插入操作
mysql> insert into cc(id,name) values(12, ‘test12’);
Query OK, 1 row affected (0.06 sec)
3.1.2Mysql执行insert后查询结果
mysql> select * from cc where id =12;
+—-+——–+——–+———————+
| id | name | status | modified_at |
+—-+——–+——–+———————+
| 12 | test12 | ok | 2016-06-24 02:27:29 |
+—-+——–+——–+———————+
1 row in set (0.02 sec)
3.1.3ES端能查询到新增的value字段。
[root@5b9dbaaa148a bin]# curl -XGET http://192.168.1.1:9200/gocc/_search?pretty -d '
> {"query":
> {"term":
> {"id":12}}}'
{
"took" : 402,
"timed_out" : false,
"_shards" : {
"total" : 8,
"successful" : 8,
"failed" : 0
},
"hits" : {
"total" : 1,
"max_score" : 1.0,
"hits" : [ {
"_index" : "gocc",
"_type" : "gocc_t",
"_id" : "12",
"_score" : 1.0,
"_source" : {
"id" : 12,
"modified_at" : "2016-06-24T02:27:29+01:00",
"name" : "test12",
"status" : "ok"
}
} ]
}
}- 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
3.2修改Update操作实时同步验证（验证ok）
3.2.1mysql执行更新操作
mysql> update cc set name = 'test12_001' where id = 12;
Query OK, 1 row affected (0.05 sec)
Rows matched: 1 Changed: 1 Warnings: 0- 1
- 2
- 3
3.2.2mysql执行修改后查询
Mysql查询修改后结果：
mysql> select * from cc where id = 12;
+----+------------+--------+---------------------+
| id | name | status | modified_at |
+----+------------+--------+---------------------+
| 12 | test12_001 | ok | 2016-06-24 02:27:29 |
+----+------------+--------+---------------------+
1 row in set (0.00 sec)- 1
- 2
- 3
- 4
- 5
- 6
- 7
3.2.3 ES查询修改结果
[root@5b9dbaaa148a bin]# curl -XGET http://192.168.1.1:9200/gocc/_search?pretty -d '
{"query":
{"term":
{"id":12}}}'
{
"took" : 59,
"timed_out" : false,
"_shards" : {
"total" : 8,
"successful" : 8,
"failed" : 0
},
"hits" : {
"total" : 1,
"max_score" : 1.0,
"hits" : [ {
"_index" : "gocc",
"_type" : "gocc_t",
"_id" : "12",
"_score" : 1.0,
"_source" : {
"id" : 12,
"modified_at" : "2016-06-24T02:27:29+01:00",
"name" : "test12_001",
"status" : "ok"
}
} ]
}
}- 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
3.3删除操作实时同步验证
3.3.1Mysql执行删除操作
mysql> delete from cc where id = 12;
Query OK, 1 row affected (0.04 sec)- 1
- 2
3.3.2删除后查询表
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 | dlllaoyang360 | ok | 0000-00-00 00:00:00 |
| 11 | test11 | ok | 2016-06-24 02:09:15 |
| 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 | test009 | ok | 0000-00-00 00:00:00 |
| 10 | test10 | ok | 2016-06-24 02:08:14 |
+----+--------------------+--------+---------------------+
9 rows in set (0.02 sec)- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
3.3.3ES查询删除后结果
[root@5b9dbaaa148a bin]# curl -XGET http://192.168.1.1:9200/gocc/_search?pretty -d '
{"query":
{"term":
{"id":12}}}'
{
"took" : 40,
"timed_out" : false,
"_shards" : {
"total" : 8,
"successful" : 8,
"failed" : 0
},
"hits" : {
"total" : 0,
"max_score" : null,
"hits" : [ ]
}
}- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
4小结
验证发现：
（1）go-mysql-elasticsearch插件可以实现同步insert、update、delete操作。
（2）可视化做的不好，没有打印日志。
（3）go-mysql-elasticsearch尚不大稳定，出现过无法同步成功的情况，但没有报错。不便于排查。
类似资料：