酷站(www.ku0.com)-致力于为互联网从业者提供专业的网络资源资讯

热门关键词:  企业  as  baidu  c4rp3nt3r  美女

Mysql实现数据同步到Elasticsearch的介绍

来源:互联网 作者:秩名 人气: 发布时间:2021-11-24
本篇文章主要介绍了Mysql实现数据同步到Elasticsearch的介绍,对大家的学习或者工作具有一定的参考学习价值,感兴趣的小伙伴们可以参考一下,也感谢大家对酷站(ku0.com)的支持。

一、同步原理

基于Mysql的binlog日志订阅:binlog日志是Mysql用来记录数据实时的变化,Mysql数据同步到ES中分为两种,分别是全量同步和增量同步,全量同步表示第一次建立好ES索引之后,将Mysql中所有数据一次性导入到ES中,增量同步表示Mysql中产生新的数据,这些新的数据包括三种情况,就是新插入Mysql中的数据,更新老的数据,删除的数据,这些数据的变动与新增都要同步到ES中

二、logstash-input-jdbc

logstash官方插件,集成在logstash中,下载logstash即可,通过配置文件实现mysql与elasticsearch数据同步

优点

能实现mysql数据全量和增量的数据同步,且能实现定时同步

版本更新迭代快,相对稳定

作为ES固有插件logstash一部分,易用

缺点

不能实现同步删除操作,MySQL数据删除后Elasticsearch中数据仍存在

同步最短时间差为一分钟,一分钟数据同步一次,无法做到实时同步

三、go-mysql-elasticsearch

go-mysql-elasticsearch 是国内作者开发的一款插件

优点

能实现mysql数据全量和增量的数据同步

缺点

无法实现数据全量同步Elasticsearch

仍处理开发、相对不稳定阶段

四、elasticsearch-jdbc

elasticsearch-jdbc 目前最新的版本是2.3.4,支持的ElasticSearch的版本为2.3.4, 未实践

优点

能实现mysql数据全量和增量的数据同步

缺点

目前最新的版本是2.3.4,支持的ElasticSearch的版本为2.3.4

不能实现同步删除操作,MySQL数据删除后Elasticsearch中数据仍存在

五、logstash-input-jdbc实现同步

第一步安装:

logstash5.x之后,集成了logstash-input-jdbc插件。安装logstash后通过命令安装logstash-input-jdbc插件

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文件

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

六、go-mysql-elasticsearch实现同步

第一步:mysql binlog日志

go-mysql-elasticsearch通过mysql中binlog日志实现数据增加,删除,修改同步elasticsearch

mysql的binlog日志主要用于数据库的主从复制与数据恢复。binlog中记录了数据的增删改查操作,主从复制过程中,主库向从库同步binlog日志,从库对binlog日志中的事件进行重放,从而实现主从同步。

mysql binlog日志有三种模式,分别为:

ROW:   记录每一行数据被修改的情况,但是日志量太大
STATEMENT:   记录每一条修改数据的SQL语句,减少了日志量,但是SQL语句使用函数或触发器时容易出现主从不一致
MIXED:   结合了ROW和STATEMENT的优点,根据具体执行数据操作的SQL语句选择使用ROW或者STATEMENT记录日志

要通过mysql binlog将数据同步到ES集群,只能使用ROW模式,因为只有ROW模式才能知道mysql中的数据的修改内容。

以UPDATE操作为例,ROW模式的binlog日志内容示例如下:

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日志内容示例为:

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操作的日志内容可以看出,ROW模式完整地记录了要修改的某行数据更新前的所有字段的值以及更改后所有字段的值,而STATEMENT模式只单单记录了UPDATE操作的SQL语句。我们要将mysql的数据实时同步到ES, 只能选择ROW模式的binlog, 获取并解析binlog日志的数据内容,执行ES document api,将数据同步到ES集群中。

查看,修改binlog模式

 # 查看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

第二步安装

# 安装go
sudo apt-get install go
 
# 安装godep
go 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-elasticsearch
make 

第三步配置

go/src/github.com/siddontang/go-mysql-elasticsearch/etc/river.toml

# MySQL address, user and password
# user must have replication privilege in MySQL.
my_addr = "127.0.0.1:3306"     # 需要同步的mysql基本设置
my_user = "root"
my_pass = "root"
 
# Elasticsearch address
es_addr = "127.0.0.1:9200"     # 本地elasticsearch配置
 
# Path to store data, like master.info, and dump MySQL data 
data_dir = "./var"             # 数据存储的url
# 以下配置保存默认不变
# Inner Http status address
stat_addr = "127.0.0.1:12800"
 
# pseudo server id like a slave 
server_id = 1001
 
# mysql or mariadb
flavor = "mysql"
# mysqldump execution path
mysqldump = "mysqldump"
 
# MySQL data source
[[source]]
schema = "test"             //elasticsearch 与 mysql 同步时对应的数据库名称
 
# Only below tables will be synced into Elasticsearch.
# 要同步test这个database里面的几张表。对于一些项目如果使用了分表机制,我们可以用通配符来匹配,譬如t_[0-9]{4},就可# 以匹配 table  t_0000 到 t_9999。
tables = ["t", "t_[0-9]{4}", "tfield", "tfilter"]  
 
# Below is for special rule mapping
# 对一个 table,我们需要指定将它的数据同步到 ES 的哪一个 index 的 type 里面。如果不指定,我们默认会用起 schema  # name 作为 ES 的 index 和 type
[[rule]]
schema = "test"    //数据库名称
table = "t"        //表名称
index = "test"        //对应的索引名称
type = "t"            //对应的类型名称
 
# 将所有满足格式 t_[0-9]{4} 的 table 同步到 ES 的 index 为 test,type 为 t 的下面。当然,这些表需要保证
# schema 是一致的
[[rule]]
schema = "test"
table = "t_[0-9]{4}"
index = "test"
type = "t"
 
# 对于 table tfilter,我们只会同步 id 和 name 这两列,其他的都不会同步
filter = ["id", "name"]
# table tfield 的 column id ,我们映射成了 es_id,而 tags 则映射成了 es_tags
# list 这个字段,他显示的告知需要将对应的 column 数据转成 ES 的 array type。这个现在通常用于 MySQL 的 varchar # 等类型,我们可能会存放类似 “a,b,c” 这样的数据,然后希望同步给 ES 的时候变成 [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 type
keywords=",list"

 第四步运行

cd go/src/github.com/siddontang/go-mysql-elasticsearch
bin/go-mysql-elasticsearch -config=./etc/river.toml

七、elasticsearch-jdbc实现同步 

设置环境变量

[root@autofelix /]# vi /etc/profile
export JDBC_IMPORTER_HOME=/elasticsearch-jdbc-2.3.2.0

使环境变量生效

[root@autofelix /]# source /etc/profile

配置参考:https://github.com/jprante/elasticsearch-jdbc

第一步:在根目录下建立根目录下新建文件夹odbc_es 如下

[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,内容如下

[root@autofelix odbc_es]# cat mysql_import_es.sh
'#!/bin/sh
bin=$JDBC_IMPORTER_HOME/bin
lib=$JDBC_IMPORTER_HOME/lib
echo '{
"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数据库地址
"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 添加可执行权限。

[root@autofelix odbc_es]# chmod a+x mysql_import_es.sh 

第四步:执行脚本mysql_import_es.sh

[root@autofelix odbc_es]# ./mysql_import_es.sh 

版权声明:本文内容来源于互联网或用户自行发布贡献,该文观点仅代表原作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌抄袭侵权/违法违规的内容, 请发送邮件至 959677720#qq.cn(#换@) 举报,一经查实,本站将立刻删除。
原文链接:https://blog.csdn.net/weixin_41635750/article/details/121445723

相关文章

  • MySQL中查询当前时间间隔前1天的数据

    MySQL中查询当前时间间隔前1天的数据

    1.背景 实际项目中我们都会遇到分布式定时任务执行的情况,有时定时任务执行的时候如果查询的数据量比较大时,我们会选择执行时间间隔几天前的数据筛选后进行执行......
    12-09
  • Mysql实现数据同步到Elasticsearch的介绍

    Mysql实现数据同步到Elasticsearch的介绍

    一、同步原理 基于Mysql的binlog日志订阅:binlog日志是Mysql用来记录数据实时的变化,Mysql数据同步到ES中分为两种,分别是全量同步和增量同步,全量同步表示第一次建立好......
    11-24
  • MySQL数据库10秒内插入百万条数据的方法

    MySQL数据库10秒内插入百万条数据的方法

    首先我们思考一个问题: 要插入如此庞大的数据到数据库,正常情况一定会频繁地进行访问,什么样的机器设备都吃不消。那么如何避免频繁访问数据库,能否做到一次......
    10-31
  • MySQL数据库事务transaction示例介绍

    MySQL数据库事务transaction示例介绍

    1、什么是事务? 一个事务是一个完整的业务逻辑单元,不可再分。 比如:银行账户转账,从A账户向B账户转账10000,需要执行两条update语句: update t_act set balance=balance-......
    10-27
  • MySQL七种JOIN类型介绍

    MySQL七种JOIN类型介绍

    在开始之前,我们创建两个表用于演示将要介绍的其中JOIN类型。 建表 CREATE TABLE `tbl_dept` ( `id` INT(11) NOT NULL AUTO_INCREMENT, `deptName` VARCHAR(30) DEFAULT NULL, `locAdd` VARCHAR(40) DEFAUL......
    10-22
  • MySQL索引下推的详细介绍

    MySQL索引下推的详细介绍

    前言: 索引下推(ICP)是针对MySQL使用索引从表中检索数据行的情况的优 在没有索引下推的情况下,MySQL通过存储引擎遍历索引来定位表中的数据行并将它们返回给MySQl服务......
    10-09
  • 解决MySql8.0查看事务隔离级别报错的问题

    解决MySql8.0查看事务隔离级别报错的问题

    ySql8.0 查看事务隔离级别报错 问题 数据库查看事务隔离级别 select @@global.tx_isolation,@@tx_isolation; 报错: mysql select @@global.tx_isolation,@@tx_isolation; ERROR 1193 (HY000): Unknown system ......
    10-09
  • mysql如何能有效防止删库跑路

    mysql如何能有效防止删库跑路

    大家肯定听说过,有些开发者由于个人失误,在delete或者update语句的时候没有添加where语句,导致整个表数据错乱。 mysql安全模式:mysql发现delete、update语句没有添加wher......
    10-05
  • Mysql binlog日志文件过大的解决方法

    Mysql binlog日志文件过大的解决方法

    磁盘突然报错使用率过大,排查原因,发现mysql的binlog文件占用过大 命令 ls -l -h mysql-binlog是MySQL数据库的二进制日志,用于记录用户对数据库操作的SQL语句((除了数据查......
    10-05
  • MySQL中limit对查询语句性能的影响

    MySQL中limit对查询语句性能的影响

    首先说明一下MySQL的版本: mysql select version();+-----------+| version() |+-----------+| 5.7.17 |+-----------+1 row in set (.00 sec) 表结构: mysql desc test;+--------+---------------------+------+-----+---......
    09-21