mysql同步elasticsearch调研
mysql同步elasticsearch调研
db同步elasticsearch调研背景
目前项目采用的是更新数据后再更新elasticsearch,各种历史原因导致很多数据并不是同步的，业务互相紧耦合, 所以需要调研适合团队发展的 db同步es机制,从业务层面剔除这部分功能维护。 下面是本人在搭建、配置、调试过程中一些总结和踩完坑后整理的配置。
搭建测试环境,低版本es:
下载elasticsearch2.3.3安装包。 下载地址：https://download.elastic.co/elasticsearch/release/org/elasticsearch/distribution/tar/elasticsearch/2.3.3/elasticsearch-2.3.3.tar.gz 安装elasticsearch-head插件,下载地址: https://github.com/mobz/elasticsearch-head/archive/master.zip
插件安装
head插件:
brew install node
npm install grunt --save-dev
npm install grunt-contrib-clean grunt-contrib-concat grunt-contrib-watch
grunt-contrib-connect grunt-contrib-copy grunt-contrib-jasmine
到head插件目录，运行grunt server，启动head 9100第一种方案 logstash-input-jdbc
1.安装logstash
此方案为定时扫描表进行同步方式,优点用户多,使用方便 缺点实时性不好，最少一分钟同步
安装前要求安装java环境
wget https://artifacts.elastic.co/downloads/logstash/logstash-5.5.0.zip
unzip logstash-5.5.0.zip
测试是否安装成功，屏幕数据将全部输出
./logstash -e 'input { stdin { } } output { stdout {} }'安装logstash插件,logstash-input-jdbc
./logstash-plugin install logstash-input-jdbc
Validating logstash-input-jdbc
Installing logstash-input-jdbc
Installed successfully确保安装ruby gem
yum install gem更换源
gem sources --add https://gems.ruby-china.org/ --remove https://gems.ruby-china.com/
gem sources -l 查看源是否更新
bin/logstash-plugin list 查看插件列表配置
mkdir config-mysql
创建mysql.confmysql.conf
input {
stdin {
}
jdbc {
jdbc_connection_string => "jdbc:mysql://10.1.58.94:3306/test"
jdbc_user => "root"
jdbc_password => ""
jdbc_driver_library => "/root/logstash-5.5.0/bin/config-mysql/mysql-connector-java-5.1.38.jar"
jdbc_driver_class => "com.mysql.jdbc.Driver"
jdbc_paging_enabled => "true"
jdbc_page_size => "50000"
statement_filepath => "/root/logstash-5.5.0/bin/config-mysql/sync.sql"
schedule => "* * * * *"
type => "article"
jdbc_default_timezone =>"Asia/Shanghai"
}
}
output {
elasticsearch {
hosts => "10.1.58.94:9200"
index => "synctest"
document_id => "%{id}"
}
stdout {
codec => json_lines
}
}实际同步sql文件 sync.sql
select
id,name,is_deleted
from seas_article
where update_time >= :sql_last_valueelasticsearch外机访问需要在elasticsearch.yml配置:
network.bind_host: 0.0.0.0运行(查看控制台扫描、添加纪录) 如果首次全亮同步,去掉配置文件中where条件
./logstash -f config-mysql/mysql.conf 
此处注意es要手动创建mapping,否则时间类可能插入报错
第二种方案 go-mysql-elasticsearch
基于binlog的同步方案,优点比较实时.缺点需要改造数据库配置
项目地址: https://github.com/siddontang/go-mysql-elasticsearch
首先查看mysql的binlog纪录形式
show variables like 'binlog_format%';配置my.cnf
[mysqld]
log-bin=mysql-bin
binlog_format=row #binglog日志必须配置为row
server_id=1001 #此server_id必须与项目配置中相同
binlog-row-image=full
...
[mysqldump] #配置mysqldump命令首次全量同步
quick
max_allowed_packet=16M
user=root
password=xxxx1.安装golang,设置gopath
2.安装项目
go get github.com/siddontang/go-mysql-elasticsearch
cd $GOPATH/src/github.com/siddontang/go-mysql-elasticsearch
make3.创建river.toml配置最简化版
my_addr = "10.1.58.94:3306" #数据库地址
my_user = "root" #数据库名
my_pass = "x x x" #密码，必须设置,否则无法运行
es_addr = "10.1.58.94:9200" #es地址 es必须配置network允许外网访问
data_dir = "./var"
stat_addr = "127.0.0.1:12800"
server_id = 1001 #server_id，必须和mysql配置一致
flavor = "mysql"
mysqldump = "mysqldump"
[[source]]
schema = "test" #数据库
tables = ["seas_article"] #要同步的表
[[rule]]
schema = "test" #具体数据库
table = "seas_article" #具体单表
index = "synctest" #es index
type = "article" #es type实际情况可能多个表组成一个index type,具体配置如下
[[source]]
schema = "test"
tables = ["seas_article","user_info"]
[[rule]]
schema = "test"
table = "seas_article"
index = "synctest"
type = "article"
[[rule]]
schema = "test"
table = "user_info"
index = "synctest"
type = "article"
id = ["id"] #根据哪个主键关联
[rule.field]
name="user_name" #字段别名
项目启动:
./bin/go-mysql-elasticsearch -config=./river.toml查看控制台输出。
其他注意事项:
当修改表结构后,要删除./var/下面的信息
控制台日志
以上工具实验结果个人总结:
1. logstash-input-jdbc工具
优点用户比较多,操作简单，不需要修改数据库配置。因为直接定时sql扫描,及时性最快也需要一分钟可同步, 无法响应物理性数据删除 ,数据量和性能方面需要压测。
2. go-mysql-elasticsearch工具
优点采用binlog同步方式,测试过程响应及时。但是需要修改数据库配置, 稳定性、数据量同样需要压测