使用canal增量同步mysql数据库信息到ElasticSearch

本文涉及的产品
RDS MySQL Serverless 基础系列,0.5-2RCU 50GB
检索分析服务 Elasticsearch 版,2核4GB开发者规格 1个月
云数据库 RDS MySQL,集群系列 2核4GB
推荐场景:
搭建个人博客
简介: 本文介绍如何使用canal增量同步mysql数据库信息到ElasticSearch。(注意:是增量!!!) 1.简介 1.1 canal介绍 Canal是一个基于MySQL二进制日志的高性能数据同步系统。

本文介绍如何使用canal增量同步mysql数据库信息到ElasticSearch。(注意:是增量!!!)

1.简介

1.1 canal介绍

Canal是一个基于MySQL二进制日志的高性能数据同步系统。Canal广泛用于阿里巴巴集团(包括https://www.taobao.com),以提供可靠的低延迟增量数据管道,github地址:https://github.com/alibaba/canal

Canal Server能够解析MySQL binlog并订阅数据更改,而Canal Client可以实现将更改广播到任何地方,例如数据库和Apache Kafka。

它具有以下功能:

  1. 支持所有平台。
  2. 支持由Prometheus提供支持的细粒度系统监控。
  3. 支持通过不同方式解析和订阅MySQL binlog,例如通过GTID。
  4. 支持高性能,实时数据同步。(详见Performance)
  5. Canal Server和Canal Client都支持HA / Scalability,由Apache ZooKeeper提供支持
  6. Docker支持。

缺点:

不支持全量更新,只支持增量更新。

完整wiki地址:https://github.com/alibaba/canal/wiki

1.2 运作原理

原理很简单:

  1. Canal模拟MySQL的slave的交互协议,伪装成mysql slave,并将转发协议发送到MySQL Master服务器。
  2. MySQL Master接收到转储请求并开始将二进制日志推送到slave(即canal)。
  3. Canal将二进制日志对象解析为自己的数据类型(原始字节流)

如图所示:

1.3 同步es

在同步数据到es的时候需要使用适配器:canal adapter。目前最新版本1.1.3,下载地址:https://github.com/alibaba/canal/releases

目前es貌似支持6.x版本,不支持7.x版本!!!

2.准备工作

2.1 es和jdk

安装es可以参考:https://www.dalaoyang.cn/article/78

安装jdk可以参考:https://www.dalaoyang.cn/article/16

2.2 安装canal server

下载canal.deployer-1.1.3.tar.gz

wget https://github.com/alibaba/canal/releases/download/canal-1.1.3/canal.deployer-1.1.3.tar.gz

解压文件

tar -zxvf canal.deployer-1.1.3.tar.gz

进入解压后的文件夹

cd canal.deployer-1.1.3

修改conf/example/instance.properties文件,主要注意以下几处:

  • canal.instance.master.address:数据库地址,例如127.0.0.1:3306
  • canal.instance.dbUsername:数据库用户
  • canal.instance.dbPassword:数据库密码

完整内容如下:

#################################################
## mysql serverId , v1.0.26+ will autoGen
# canal.instance.mysql.slaveId=0

# 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=
#canal.instance.tsdb.dbUsername=
#canal.instance.tsdb.dbPassword=

#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=root
canal.instance.dbPassword=12345678
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=

# 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-1.1.3目录下,启动canal:

sh bin/startup.sh

查看日志:

vi logs/canal/canal.log

查看具体instance日志:

 vi logs/example/example.log

关闭命令

sh bin/stop.sh

2.3 安装canal-adapter

下载canal.adapter-1.1.3.tar.gz

wget https://github.com/alibaba/canal/releases/download/canal-1.1.3/canal.adapter-1.1.3.tar.gz

解压

tar -zxvf canal.adapter-1.1.3.tar.gz

进入解压后的文件夹

cd canal.adapter-1.1.3

修改conf/application.yml文件,主要注意如下内容,由于是yml文件,注意我这里说明的属性名称:

  • server.port:canal-adapter端口号
  • canal.conf.canalServerHost:canal-server地址和ip
  • canal.conf.srcDataSources.defaultDS.url:数据库地址
  • canal.conf.srcDataSources.defaultDS.username:数据库用户名
  • canal.conf.srcDataSources.defaultDS.password:数据库密码
  • canal.conf.canalAdapters.groups.outerAdapters.hosts:es主机地址,tcp端口

完整内容如下:

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
  canalServerHost: 127.0.0.1:11111
  batchSize: 500
  syncBatchSize: 1000
  retries: 0
  timeout:
  accessKey:
  secretKey:
  srcDataSources:
    defaultDS:
      url: jdbc:mysql://127.0.0.1:3306/test?useUnicode=true
      username: root
      password: 12345678
  canalAdapters:
  - instance: example
    groups:
    - groupId: g1
      outerAdapters:
      - name: es
        hosts: 127.0.0.1:9300
        properties:
         cluster.name: elasticsearch

另外需要配置conf/es/*.yml文件,adapter将会自动加载conf / es下的所有.yml结尾的配置文件。在介绍配置前,需要先介绍一下本案例使用的表结构,如下:

CREATE TABLE `test` (
  `id` int(11) NOT NULL,
  `name` varchar(200) NOT NULL,
  `address` varchar(1000) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

需要手动去es中创建索引,比如这里使用es-head创建,如下图:

test索引结构如下:

{
    "mappings":{
        "_doc":{
            "properties":{
                "name":{
                    "type":"text"
                },
                "address":{
                    "type":"text"
                }
            }
        }
    }
}

接下来创建test.yml(文件名随意),内容很好理解_index为索引名称,sql为对应语句,内容如下:

dataSourceKey: defaultDS
destination: example
groupId:
esMapping:
  _index: test
  _type: _doc
  _id: _id
  upsert: true
  sql: "select a.id as _id,a.name,a.address from test a"
  commitBatch: 3000

配置完成后,回到canal-adapter根目录,执行命令启动

bin/startup.sh

查看日志

vi logs/adapter/adapter.log

关闭canal-adapter命令

bin/stop.sh

3.测试

都启动成功后,先查看一下es-head,如图,现在是没有任何数据的。

接下来,我们在数据库中插入一条数据进行测试,语句如下:

INSERT INTO `test`.`test`(`id`, `name`, `address`) VALUES (7, '北京', '北京市朝阳区');

然后在看一下es-head,如下

接下来看一下日志,如下:

2019-06-22 17:54:15.385 [pool-2-thread-1] DEBUG c.a.otter.canal.client.adapter.es.service.ESSyncService - DML: {"data":[{"id":7,"name":"北京","address":"北京市朝阳区"}],"database":"test","destination":"example","es":1561197255000,"groupId":null,"isDdl":false,"old":null,"pkNames":["id"],"sql":"","table":"test","ts":1561197255384,"type":"INSERT"} 
Affected indexes: test 

小知识点:上面介绍的查看日志的方法可能不是很好用,推荐使用如下语法,比如查看日志最后200行:

tail -200f logs/adapter/adapter.log

4.总结

1.全量更新不能实现,但是增删改都是可以的。
2.一定要提前创建好索引。
3.es配置的是tcp端口,比如默认的9300

相关实践学习
使用阿里云Elasticsearch体验信息检索加速
通过创建登录阿里云Elasticsearch集群,使用DataWorks将MySQL数据同步至Elasticsearch,体验多条件检索效果,简单展示数据同步和信息检索加速的过程和操作。
ElasticSearch 入门精讲
ElasticSearch是一个开源的、基于Lucene的、分布式、高扩展、高实时的搜索与数据分析引擎。根据DB-Engines的排名显示,Elasticsearch是最受欢迎的企业搜索引擎,其次是Apache Solr(也是基于Lucene)。 ElasticSearch的实现原理主要分为以下几个步骤: 用户将数据提交到Elastic Search 数据库中 通过分词控制器去将对应的语句分词,将其权重和分词结果一并存入数据 当用户搜索数据时候,再根据权重将结果排名、打分 将返回结果呈现给用户 Elasticsearch可以用于搜索各种文档。它提供可扩展的搜索,具有接近实时的搜索,并支持多租户。
目录
相关文章
|
1月前
|
canal 消息中间件 关系型数据库
Canal作为一款高效、可靠的数据同步工具,凭借其基于MySQL binlog的增量同步机制,在数据同步领域展现了强大的应用价值
【9月更文挑战第1天】Canal作为一款高效、可靠的数据同步工具,凭借其基于MySQL binlog的增量同步机制,在数据同步领域展现了强大的应用价值
197 4
|
3天前
|
消息中间件 canal 关系型数据库
Maxwell:binlog 解析器,轻松同步 MySQL 数据
Maxwell:binlog 解析器,轻松同步 MySQL 数据
30 11
|
1月前
|
NoSQL 关系型数据库 Redis
mall在linux环境下的部署(基于Docker容器),Docker安装mysql、redis、nginx、rabbitmq、elasticsearch、logstash、kibana、mongo
mall在linux环境下的部署(基于Docker容器),docker安装mysql、redis、nginx、rabbitmq、elasticsearch、logstash、kibana、mongodb、minio详细教程,拉取镜像、运行容器
mall在linux环境下的部署(基于Docker容器),Docker安装mysql、redis、nginx、rabbitmq、elasticsearch、logstash、kibana、mongo
|
20天前
|
存储 自然语言处理 关系型数据库
ElasticSearch基础3——聚合、补全、集群。黑马旅游检索高亮+自定义分词器+自动补全+前后端消息同步
聚合、补全、RabbitMQ消息同步、集群、脑裂问题、集群分布式存储、黑马旅游实现过滤和搜索补全功能
ElasticSearch基础3——聚合、补全、集群。黑马旅游检索高亮+自定义分词器+自动补全+前后端消息同步
|
2月前
|
关系型数据库 MySQL Linux
mysql 主从同步 实现增量备份
【8月更文挑战第28天】mysql 主从同步 实现增量备份
37 3
|
2月前
|
消息中间件 关系型数据库 MySQL
实时计算 Flink版产品使用问题之使用CTAS同步MySQL到Hologres时出现的时区差异,该如何解决
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
2月前
|
SQL 缓存 关系型数据库
MySQL主从同步如何操作?
随着业务增长,单台MySQL服务器难以应对高并发访问和潜在的故障风险。主从同步(Master-Slave)通过读写分离提升数据库处理能力,具备多项优势:读写分离减轻主数据库压力、支持一主多从增强扩展性与高可用性、以及数据备份确保容灾恢复。MySQL利用binlog实现主从数据同步,记录所有写操作,不包含查询。binlog有三种格式:Statement(基于SQL语句)、Row(基于行更改)、Mixed(结合前两者优点)。主从复制涉及三个关键线程:主库的binlog dump thread和从库的I/O thread与SQL thread。
MySQL主从同步如何操作?
|
2月前
|
存储 关系型数据库 MySQL
MySQL主从同步如何保证数据一致性?
MySQL主从同步如何保证数据一致性?
76 0
MySQL主从同步如何保证数据一致性?
|
2月前
|
SQL 存储 关系型数据库
实时计算 Flink版产品使用问题之同步MySQL多张表的过程中,内存释放依赖于什么
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
2月前
|
SQL 存储 关系型数据库
MySQL主从同步延迟原因与解决方法
MySQL主从同步延迟原因与解决方法
132 0

热门文章

最新文章

下一篇
无影云桌面