目前大部分mysql和elasticsearch同步机制使用的插件实现的,常用的插件包为:logstash-input-jdbc,go-mysql-elasticsearch, elasticsearch-jdbc、canal。
插件优缺点对比
1. logstash-input-jdbc
logstash官方插件,集成在logstash中,下载logstash即可,通过配置文件实现mysql与elasticsearch数据同步
优点
·能实现mysql数据全量和增量的数据同步,且能实现定时同步。
·版本更新迭代快,相对稳定。
·作为ES固有插件logstash一部分,易用。
缺点
·不能实现同步删除操作,MySQL数据删除后Elasticsearch中数据仍存在。
·同步最短时间差为一分钟,一分钟数据同步一次,无法做到实时同步。
2、go-mysql-elasticsearch
go-mysql-elasticsearch 是国内作者开发的一款插件
优点
·能实现mysql数据增加,删除,修改操作的实时数据同步
缺点
·无法实现数据全量同步Elasticsearch
·仍处理开发、相对不稳定阶段
3、elasticsearch-jdbc
目前最新的版本是2.3.4,支持的ElasticSearch的版本为2.3.4, 未实践
优点
·能实现mysql数据全量和增量的数据同步.
缺点
·目前最新的版本是2.3.4,支持的ElasticSearch的版本为2.3.4
·不能实现同步删除操作,MySQL数据删除后Elasticsearch中数据仍存在.
mysql安装
安装依赖
yum search libaio # 检索相关信息
yum install libaio # 安装依赖包
mysql是否安装
yum list installed | grep mysql
mysql卸载
yum -y remove mysql-libs.x86_64
mysql yum下载
wget -i -c http://dev.mysql.com/get/mysql57-community-release-el7-10.noarch.rpm
安装 MySQL
yum -y install mysql57-community-release-el7-10.noarch.rpm
yum -y install mysql-community-server
查看mysql安装位置
whereis mysql
启动mysql
systemctl start mysqld.service
systemctl status mysqld.service
关闭mysql
systemctl stop mysqld
查看密码
grep 'temporary password' /var/log/mysqld.log
mysql修改密码远程连接
SET PASSWORD = PASSWORD('/20as3SElksds0ew98');
GRANT ALL PRIVILEGES ON *.* TO 'root'@'%' IDENTIFIED BY '/20as3SElksds0ew98' WITH GRANT OPTION;
logstash-input-jdbc实现mysql数据库与elasticsearch同步
logstash5.x之后,集成了logstash-input-jdbc插件。安装logstash后通过命令安装logstash-input-jdbc插件
./logstash-plugin install logstash-input-jdbc
配置
在logstash/config文件夹下新建jdbc.conf,配置内容如下:
# 输入部分 input { stdin {} jdbc { # mysql数据库驱动 jdbc_driver_library => "../config/mysql-connector-java-5.1.30.jar" jdbc_driver_class => "com.mysql.jdbc.Driver" # mysql数据库链接,数据库名 jdbc_connection_string => "jdbc:mysql://localhost:3306/cmr" # mysql数据库用户名,密码 jdbc_user => "root" jdbc_password => "12345678" # 设置监听间隔 各字段含义(分、时、天、月、年),全部为*默认含义为每分钟更新一次 schedule => "* * * * *" # 分页 jdbc_paging_enabled => "true" # 分页大小 jdbc_page_size => "50000" # sql语句执行文件,也可直接使用 statement => 'select * from t_school_archives_fold create_time >= :sql_last_value order by create_time limit 200000' statement_filepath => "/config/jdbc.sql" # elasticsearch索引类型名 type => "t_employee" } } # 过滤部分(不是必须项) filter { json { source => "message" remove_field => ["message"] } } # 输出部分 output { elasticsearch { # elasticsearch索引名 index => "octopus" # 使用input中的type作为elasticsearch索引下的类型名 document_type => "%{type}" # <- use the type from each input # elasticsearch的ip和端口号 hosts => "localhost:9200" # 同步mysql中数据id作为elasticsearch中文档id document_id => "%{id}" } stdout { codec => json_lines } } # 注: 使用时请去掉此文件中的注释,不然会报错
在config 目录下新建jdbc.sql文件
select * from t_employee
运行
cd logstash-6.4.2 # 检查配置文件语法是否正确 bin/logstash -f config/jdbc.conf --config.test_and_exit # 启动 bin/logstash -f config/jdbc.conf --config.reload.automatic --config.reload.automatic: 会自动重新加载配置文件内容 在kibana中创建索引后查看同步数据 PUT octopus GET octopus/_search
Canal实现mysql数据库与elasticsearch同步
mysql
修改/etc/my.cnf
log-bin=mysql-bin binlog-format=ROW server-id=1
创建授权
create user canal identified by 'Canal@2020!'; #创建canal账户 grant select,replication slave,replication client on *.* to 'canal'@'%'; #授权canal账户查询和复制权限 flush privileges; #刷新授权
查看binlog是否正确启动
show variables like 'binlog_format%';
创建需要同步的数据库
create database canal_testdb character set utf8; CREATE TABLE canal_table ( #创建canal_table表,字段为 id age name address id int(11) NOT NULL, age int(11) NOT NULL, name varchar(200) NOT NULL, address varchar(1000) DEFAULT NULL, PRIMARY KEY (id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; INSERT INTO canal_testdb.canal_table(id, age, name, address) VALUES (1, 88, '小明', '测试');
Elasticsearch参考第一节
部署Canal-deployer服务端
下载并解压
# 没有的话新建 cd /usr/local/canal/canal-deployer/ wget https://github.com/alibaba/canal/releases/download/canal-1.1.5-alpha-1/canal.deployer-1.1.5-SNAPSHOT.tar.gz tar xf canal.deployer-1.1.5-SNAPSHOT.tar.gz
修改配置文件 instance.properties
vim /usr/local/canal-deployer/conf/example/instance.properties ################################################# ## mysql serverId , v1.0.26+ will autoGen canal.instance.mysql.slaveId=3 # enable gtid use true/false canal.instance.gtidon=false # position info canal.instance.master.address=127.0.0.1:3306 canal.instance.master.journal.name= canal.instance.master.position= canal.instance.master.timestamp= canal.instance.master.gtid= # rds oss binlog canal.instance.rds.accesskey= canal.instance.rds.secretkey= canal.instance.rds.instanceId= # table meta tsdb info canal.instance.tsdb.enable=true #canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_testdb #canal.instance.tsdb.dbUsername=canal #canal.instance.tsdb.dbPassword=canal #canal.instance.standby.address = #canal.instance.standby.journal.name = #canal.instance.standby.position = #canal.instance.standby.timestamp = #canal.instance.standby.gtid= # username/password canal.instance.dbUsername=canal canal.instance.dbPassword=Canal@2020! canal.instance.connectionCharset = UTF-8 # enable druid Decrypt database password canal.instance.enableDruid=false #canal.instance.pwdPublicKey=MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBALK4BUxdDltRRE5/zXpVEVPUgunvscYFtEip3pmLlhrWpacX7y7GCMo2/JM6LeHmiiNdH1FWgGCpUfircSwlWKUCAwEAAQ== # table regex canal.instance.filter.regex=.*\\..* # table black regex canal.instance.filter.black.regex= # table field filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2) #canal.instance.filter.field=test1.t_product:id/subject/keywords,test2.t_company:id/name/contact/ch # table field black filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2) #canal.instance.filter.black.field=test1.t_product:subject/product_image,test2.t_company:id/name/contact/ch # mq config canal.mq.topic=example # dynamic topic route by schema or table regex #canal.mq.dynamicTopic=mytest1.user,mytest2\\..*,.*\\..* canal.mq.partition=0 # hash partition config #canal.mq.partitionsNum=3 #canal.mq.partitionHash=test.table:id^name,.*\\..* #################################################
启动canal-deployer
因为canal-depaloyer由java开发,所以需要jdk环境,jdk版本需要大于1.5
yum install java-1.8.0-openjdk.x86_64 java-1.8.0-openjdk-devel.x86_64 -y
/usr/local/canal/canal-deployer/bin/startup.sh
查看日志及端口
tail -f /usr/local/canal/logs/example/example.log
canal-deployer默认监听三个端口,11110、11111、11112
11110:为admin管理端口
11111:为canal deployer 服务器占用的端口
11112:为指标下拉端口
部署Canal-adapter客户端
下载并解压
cd /usr/local/canal/canal-adapter/ wget https://github.com/alibaba/canal/releases/download/canal-1.1.5-alpha-1/canal.adapter-1.1.5-SNAPSHOT.tar.gz tar xf canal.adapter-1.1.5-SNAPSHOT.tar.gz
添加mysql8.0.18连接器
cd /usr/local/canal/canal-adapter/lib/ wget https://repo1.maven.org/maven2/mysql/mysql-connector-java/8.0.18/mysql-connector-java-8.0.18.jar chmod 777 /usr/local/canal-adapter/lib/mysql-connector-java-8.0.18.jar #权限修改与其它lib库一致 chmod +st /usr/local/canal-adapter/lib/mysql-connector-java-8.0.18.jar
修改application.yml
server: port: 8081 spring: jackson: date-format: yyyy-MM-dd HH:mm:ss time-zone: GMT+8 default-property-inclusion: non_null canal.conf: mode: tcp # kafka rocketMQ canalServerHost: 127.0.0.1:11111 # zookeeperHosts: slave1:2181 # mqServers: 127.0.0.1:9092 #or rocketmq # flatMessage: true batchSize: 500 syncBatchSize: 1000 retries: 0 timeout: srcDataSources: defaultDS: url: jdbc:mysql://127.0.0.1:3306/canal_testdb?useUnicode=true username: canal password: Canal@2020! canalAdapters: - instance: example # canal instance Name or mq topic name groups: - groupId: g1 outerAdapters: - name: logger - name: es7 hosts: 192.168.0.200:9300,192.168.0.200:8200 properties: mode: rest # or rest # security.auth: test:123456 # only used for rest mode cluster.name: my-es
修改适配器映射文件
vim /usr/local/canal/canal-adapter/conf/es7/mytest_user.yml dataSourceKey: defaultDS #指定在application.yml文件中srcDataSources源数据源自定义的名称 destination: example #cannal的instance或者MQ的topic,我们是把数据同步至es,所以不用修改,也用不到此处 groupId: g1 #对应MQ模式下的groupId, 只会同步对应groupId的数据 esMapping: #es中的Mapping设置 _index: canal_tsdb #指定索引名称 _id: _id #指定文档id,_id 此值则由es自动分配文档ID sql: "select a.id as _id,a.age,a.name,a.address from canal_table a" #sql映射 etlCondition: "where a.c_time>={}" #etl的条件参数 commitBatch: 3000 #提交批大小
Elasticsearch创建索引
POST canal_tsdb/_doc { "mappings":{ "_doc":{ "properties":{ "age":{ "type":"long" }, "name":{ "type":"text" }, "address":{ "type":"text" } } } }
启动Canal-adapter并写入数据
/usr/local/canal/canal-adapter/bin/startup.sh
tail -f /usr/local/canal/canal-adapter/logs/adapter/adapter.log
在MySQL再次插入一条数据并查看日志
INSERT INTO canal_tsdb.canal_table(id, age, name, address) VALUES (2, 88, '小明', '测试');
查看Canal-deployer服务端日志
tail -f /usr/local/canal/canal-deployer/logs/example/meta.log
在es里面可以看到数据
部署Canal Admin
canal-admin的限定依赖:
1.MySQL,用于存储配置和节点等相关数据
2.canal版本,要求>=1.1.4 (需要依赖canal-server提供面向admin的动态运维管理接口)
Canal Admin下载并解压
mkdir /usr/local/canal/canal-admin wget https://github.com/alibaba/canal/releases/download/canal-1.1.5-alpha-1/canal.admin-1.1.5-SNAPSHOT.tar.gz tar xf canal.admin-1.1.5-SNAPSHOT.tar.gz
application.yml
server: port: 8089 #Canal Admin监听端口 spring: jackson: date-format: yyyy-MM-dd HH:mm:ss #时间格式 time-zone: GMT+8 #时区 spring.datasource: #数据库信息 address: 192.168.0.200:8809 #指定Canal Admin所使用的数据库地址及端口 database: canal_manager #指定数据库名称 username: cadmin #指定数据库账户 password: Cadmin@2020! #指定数据库密码 driver-class-name: com.mysql.jdbc.Driver #指定数据库驱动 url: jdbc:mysql://${spring.datasource.address}/${spring.datasource.database}?useUnicode=true&characterEncoding=UTF-8&useSSL=false hikari: maximum-pool-size: 30 minimum-idle: 1 canal: #Canal UI界面默认账号密码 adminUser: admin adminPasswd: admin
创建数据库及授权用户
create database canal_manager character set utf8; create database canal_manager character set utf8; create user cadmin identified by 'Cadmin@2020!'; grant all on canal_manager.* to 'cadmin'@'%'; flush privileges;
/usr/local/canal/canal-admin/conf/canal_manager.sql;数据库数据导入
启动
/usr/local/canal/canal-admin/bin/startup.sh
作者:wmburst
来源:51Testing软件测试网原创