0


mysql 5.7同步数据到es 7.6.2(集群)

小伙伴们,你们好,我是老寇

mysql驱动、canal安装包:https://pan.baidu.com/s/1swrV9ffJnmz4S0mfkuBbIw
提取码:1111

一、前提条件

  • 安装mycat 1.6,点点我
  • 安装es 7.6.2集群,点点我
  • 安装logstash 7.6.2,点点我

二、可选方案

在实际项目中,业务数据主流存储在mysql,但是mysql处理海量数据的搜索能力较差,推荐mysql搭配es,为业务提供强大的搜索能力成为业界主流方案,难点在于如何将mysql导入到es。

mysql数据同步到es,有以下几种方案:

  • 更新mysql数据库的同时通过es api实现数据写入es(同步)
  • 通过logstash同步数据到es(异步)
  • 通过mysql binlog,将数据同步到es(异步)

三、es api同步数据

直接调用es api即可,代码详情,请点点我

    @PutMapping("/sync")
    @ApiOperation("同步修改ES")
    @CrossOrigin
    public void updateData(@RequestBody final ElasticsearchModel model) {
        String id = model.getId();
        String indexName = model.getIndexName();
        String paramJson = model.getData();
        elasticsearchUtil.updateData(indexName,id,paramJson);
    }

四、logstash同步数据到es

参考博客:同步MYSQL数据到Elasticsearch

logstash通过使用jdbc-input实现定时同步mysql数据,因此需要提前下载mysql的jdbc驱动

1.配置logstash.mysql.conf

input {
  # 配置JDBC数据源
  jdbc {
    # mysql jdbc驱动路径
    jdbc_driver_library => "D:/xxxxx/mysql-connector-java-5.1.35.jar"
    # mysql jdbc驱动
    jdbc_driver_class => "com.mysql.jdbc.Driver"
    # mysql 连接地址
    jdbc_connection_string => "jdbc:mysql://localhost:8066/TESTDB"
    # mysql 账号
    jdbc_user => "root"
    # mysql 密码
    jdbc_password => "123456"
    # 定时任务配置,一分钟执行一次
    schedule => "* * * * *"
    # sql查询语句
    # 增量更新数据
    statement => "select * from boot_resource where id > :sql_last_value"
    # 允许sql_last_value的值来源于结果的某个字段值
    use_column_value => true
    # sql_last_value的值来自查询结果的最后一个id值
    tracking_column => "id"
    # 开启分页
    jdbc_paging_enabled => true
    # 分页大小
    jdbc_page_size => 50
  }
}

output {
  # 配置数据导入es
  elasticsearch {
    # 索引名称
    index => "laokou-resource"
    # es地址
    hosts => ["192.168.1.1:9200","192.168.1.2:9200","192.168.1.3:9200"]
    # document_id设置mysql表主键
    document_id => "%{id}"
  }
  stdout {
    codec => rubydebug
  }
}

** 2.jdbc schedule 配置说明**

jdbc schedule的配置规则

*   *   *   *   *
分  时  天  月  星期

各字段取值范围

  • 分 - 0 ~ 59
  • 时 - 0 ~ 23
  • 天 - 1 ~ 31
  • 月 - 1 ~ 12
  • 星期 - 0 ~ 6

常见特殊字符含义

  • 星号(*):代表所有值,如第一个星号,表示每分钟
  • 斜线(/):指定时间的间隔频率,如*/5,用在第一个分钟字段,表示每5分钟执行一次

举个栗子

# 每分钟执行一次
* * * * *

# 每5分钟执行一次
*/5 * * * *

# 每小时执行一次
* */1 * * *

# 每天八点执行一次
0 8 * * *

3.增量同步数据

同步数据的sql语句,直接扫描全表的数据,如果数据量较小,问题不大,如果数据量比较大,会直接卡死,logstash OOM挂掉,因此需要增量同步数据,并且每次仅同步新增数据

logstash提供了sql_last_value字段值,可以实现增量同步数据

实现思路:logstash每次执行sql时,会将sql查询结果的最后一条记录某个字段的值保存到sql_last_value字段中,下一次执行sql时,就以sql_last_value的值做条件,从这个值往后查询数据

statement => "select * from boot_resource where id > :sql_last_value"

4.数据分页

增量同步的数据太多,会导致logstash卡死,尤其是首次增量同步时(首次同步数据,其实是全表扫描),为避免一次查询太多数据,可以配置分页

    # 开启分页
    jdbc_paging_enabled => true
    # 分页大小
    jdbc_page_size => 50

5.启动logstash

logstash -f logstash.mysql.conf

五、mysql通过binlog同步数据到es

阿里的开源神器canal

1. canal简介

canal主要用于对mysql的增量日志进行解析(请注意,只支持增量解析,不支持全量解析),提供增量数据的订阅消费,对mysql增量数据进行实时同步,支持同步到mysql、elasticsearch、hbase等数据源

2.canal常用组件

  • canal-deployer(canal-server):监听mysql的binlog,把自己伪装成mysql slave,只负责接收数据,不做数据处理
  • canal-adapter:canal客户端,从canal-server中获取数据,对数据进行同步,可以同步到mysql、elasticsearch、hbase等数据源
  • canal-admin:提供整体配置管理、节点运维等功能,web界面方便用户快速和安全的操作

3.canal工作原理

  • canal模拟mysql master slave的交互协议,把自己伪装mysql slave,向mysql master发送dump协议(dump用于备份)
  • mysql master接收到dump请求,向canal推送binlog
  • canal通过解析binlog,将数据同步到其他的数据源

4.MySQL主从复制原理

  • mysql master在每个事务更新数据完成之前,将该操作记录串行写入binlog(二进制文件)文件中
  • mysql slave 开启一个I/O线程,该线程在master打开一个普通连接,主要工作的是binlog dump process,如果读取的进度已经跟上master,就会进入休眠状态并等待master产生新的事件,I/O线程的最终目的就是将这些事件写入到replay log(中继日志)
  • sql线程会读取中继日志,并按顺序执行该日志的sql事件,从而与mysql master的数据保持一致

5.下载canal 1.1.5并安装(win11教程,下载完解压文件)

百度网盘:https://pan.baidu.com/s/1swrV9ffJnmz4S0mfkuBbIw
提取码:1111

6.mysql配置

master 配置

log-bin=mysql-bin ##开启二进制日志
binlog-format=row ##二进制日志格式(row\mixed\statement)

查看binlog是否启用

show variables like '%log_bin%';

查看binlog模式

show variables like '%binlog_format%';

** 7.创建账号,用于订阅binlog**

CREATE USER canal IDENTIFIED BY '123456';  
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
FLUSH PRIVILEGES;

8.创建数据库

DROP TABLE IF EXISTS `boot_city`;
CREATE TABLE `boot_city` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT 'id',
  `parent_id` bigint(20) NOT NULL COMMENT '上级编号',
  `region` varchar(100) NOT NULL COMMENT '地区名称',
  `region_id` int(11) NOT NULL COMMENT '地区编号',
  PRIMARY KEY (`id`) USING BTREE
) ENGINE=InnoDB AUTO_INCREMENT=1324914279466823682 DEFAULT CHARSET=utf8mb4 ROW_FORMAT=DYNAMIC COMMENT='城市';

9.配置canal-server

1.配置conf/example/instance.properties

# 需要同步数据的MySQL地址
canal.instance.master.address=192.168.1.1:3306 
canal.instance.master.journal.name=
canal.instance.master.position=
canal.instance.master.timestamp=
canal.instance.master.gtid=
# 用于同步数据的数据库账号
canal.instance.dbUsername=canal
# 用于同步数据的数据库密码
canal.instance.dbPassword=123456
# 数据库连接编码
canal.instance.connectionCharset = UTF-8
# 需要订阅binlog的表过滤正则表达式
canal.instance.filter.regex=.*\\..*

2.启动服务只需要双击bin/startup.bat

10.配置canal-adapter

1.修改conf/application.yml

#修改 es\mysql配置 
  srcDataSources:
    defaultDS:
      url: jdbc:mysql://192.168.1.1:3306/kcloud?useUnicode=true
      username: root
      password: 123456
  canalAdapters:
  - instance: example # canal instance Name or mq topic name
    groups:
    - groupId: g1
      outerAdapters:
      - name: logger
      - name: es7
        hosts: 192.168.1.1:9300,192.168.1.2:9300,192.168.1.3:9300 # 127.0.0.1:9200 for rest mode
        properties:
          mode: transport # or rest
          # security.auth: test:123456 #  only used for rest mode
          cluster.name: laokou-elasticsearch

2.修改conf/es7/mytest_user.yml

dataSourceKey: defaultDS
destination: example
groupId: g1
esMapping:
  _index: laokou-city
  _id: id
  sql: "select a.id,a.parent_id,a.region,a.region_id from boot_city a"
  etlCondition: "where a.id >= {}"
  commitBatch: 3000

3.启动服务只需要双击bin/startup.bat

11.启动kibana,创建索引

PUT /laokou-city
{
  "mappings": {
    "properties": {
      "id":{
        "type": "long"
      },
      "parent_id":{
        "type": "long"
      },
      "region":{
        "type": "keyword"
      },
      "region_id":{
        "type": "integer"
      }
    }
  }
}

12.数据库插入数据

INSERT INTO `boot_city` VALUES ('1324912501966925826', '227', '东安县', '2093');
INSERT INTO `boot_city` VALUES ('1324912504676446210', '227', '双牌县', '2094');
INSERT INTO `boot_city` VALUES ('1324912506161229825', '227', '道县', '2095');
INSERT INTO `boot_city` VALUES ('1324912507595681794', '227', '江永县', '2096');
INSERT INTO `boot_city` VALUES ('1324912509306957825', '227', '宁远县', '2097');
INSERT INTO `boot_city` VALUES ('1324912511995506689', '227', '蓝山县', '2098');
INSERT INTO `boot_city` VALUES ('1324912514667278338', '227', '新田县', '2099');
INSERT INTO `boot_city` VALUES ('1324912516038815746', '227', '江华瑶族自治县', '2100');
INSERT INTO `boot_city` VALUES ('1324912517494239234', '228', '鹤城区', '2101');
INSERT INTO `boot_city` VALUES ('1324912520161816578', '228', '中方县', '2102');
INSERT INTO `boot_city` VALUES ('1324912521667571714', '228', '沅陵县', '2103');
INSERT INTO `boot_city` VALUES ('1324912523072663553', '228', '辰溪县', '2104');
INSERT INTO `boot_city` VALUES ('1324912524578418690', '228', '溆浦县', '2105');
INSERT INTO `boot_city` VALUES ('1324912526096756737', '228', '会同县', '2106');
INSERT INTO `boot_city` VALUES ('1324912527514431490', '228', '麻阳苗族自治县', '2107');
INSERT INTO `boot_city` VALUES ('1324912529036963841', '228', '新晃侗族自治县', '2108');
INSERT INTO `boot_city` VALUES ('1324912530588856321', '228', '芷江侗族自治县', '2109');
INSERT INTO `boot_city` VALUES ('1324912531956199426', '228', '靖州苗族侗族自治县', '2110');
INSERT INTO `boot_city` VALUES ('1324912533394845697', '228', '通道侗族自治县', '2111');

canal-server会监听binlog日志数据是否发生更改

13.效果截图

14.问题解决

注:如果能获取canal-sever数据,不能写入es

请替换canal-adapter的plugin目录下的client-adapter.es7x-1.1.5-jar-with-dependencies.jar

百度网盘:https://pan.baidu.com/s/1swrV9ffJnmz4S0mfkuBbIw
提取码:1111

大功告成

标签: elasticsearch mysql

本文转载自: https://blog.csdn.net/qq_39893313/article/details/123688809
版权归原作者 k↑ 所有, 如有侵权,请联系我们删除。

“mysql 5.7同步数据到es 7.6.2(集群)”的评论:

还没有评论