本地搭建部署Canal笔记-实现MySQL与ElasticSearch7数据同步

2024-01-08 13:12
文章标签 数据 实现 mysql 部署 笔记 搭建 elasticsearch7 canal 同步 本地 database

本文主要是介绍本地搭建部署Canal笔记-实现MySQL与ElasticSearch7数据同步，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

背景

本地搭建canal实现mysql数据到es的简单的数据同步，仅供学习参考

建议首先熟悉一下canal同步方式：https://github.com/alibaba/canal/wiki/%E7%AE%80%E4%BB%8B

前提条件

  • 本地搭建MySQL数据库
  • 本地搭建ElasticSearch
  • 本地搭建canal-server
  • 本地搭建canal-adapter

操作步骤

本地环境为window11，大部分组件采用docker进行部署,MySQL采用8.0.27，

在这里插入图片描述

搭建MySQL数据库

推荐使用docker-desktop，可视化操作，首先找到需要使用的版本直接pull到本地，通过界面进行容器启动，指定环境变量并定义挂载卷

在这里插入图片描述

也可本地使用docker命令部署MySQL，需要指定环境变量，root账户密码和挂载本地配置文件

#命令根据自己本地路径和配置进行更改
docker run  
--env=MYSQL_ROOT_PASSWORD=123456 
--volume=D:\docker\mysql\data:/var/lib/mysql/ 
--volume=D:\docker\mysql\conf:/etc/mysql/conf.d 
--volume=/var/lib/mysql 
-p 3306:3306 
-p 33060:33060 --runtime=runc -d 
--privileged=true 
mysql:latest

命令说明：

  • -p 3306:3306
    将宿主机的 3306 端口映射到 docker 容器的 3306 端口，格式为：主机(宿主)端口:容器端口
  • --name mysql
    运行服务的名字
  • -v
    挂载数据卷，格式为：宿主机目录或文件:容器内目录或文件
  • -e MYSQL_ROOT_PASSWORD=123456
    初始化 root 用户的密码为 123456
  • -d
    后台程序运行容器
  • --privileged=true 开启特殊权限
    Docker 挂载主机目录时（添加容器数据卷），如果 Docker 访问出现 cannot open directory:Permission denied，在挂载目录的命令后多加一个 --privileged=true 参数即可。
    因为出于安全原因，容器不允许访问任何设备，privileged 让 docker 应用容器获取宿主机 root 权限（特殊权限），允许我们的 Docker 容器访问连接到主机的所有设备。容器获得所有能力，可以访问主机的所有设备，例如，CD-ROM、闪存驱动器、连接到主机的硬盘驱动器等。

下一步在MySQL配置文件中my.cnf设置开启binlog,更改配合文件之后重启MySQL

# Copyright (c) 2017, Oracle and/or its affiliates. All rights reserved.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; version 2 of the License.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301 USA#
# The MySQL  Server configuration file.
#
# For explanations see
# http://dev.mysql.com/doc/mysql/en/server-system-variables.html[mysqld]
pid-file        = /var/run/mysqld/mysqld.pid
socket          = /var/run/mysqld/mysqld.sock
datadir         = /var/lib/mysql
secure-file-priv= NULL--skip-host-cache
--skip-name-resolve[mysqld]
log-bin=mysql-bin #添加这一行就ok
binlog-format=ROW #选择row模式
server_id=1 #配置mysql replaction需要定义，不能和canal的slaveId重复# Custom config should go here
!includedir /etc/mysql/conf.d/

更改完配置文件之后验证binlog是否打开，并创建canal用户，设置权限

-- 查询binlog配置
show variables like 'binlog_format';
show variables like 'log_bin';-- 创建canal账户
CREATE USER canal IDENTIFIED BY 'canal';
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';FLUSH PRIVILEGES;show grants for 'canal';

准备工作

  • 新建数据库
CREATE SCHEMA `canal_manager` DEFAULT CHARACTER SET utf8mb4 ;
  • 新增测试表
create table test_canal
(id   int auto_incrementprimary key,name text null
);

搭建ELK

本地搭建elk推荐使用docker-compose进行搭建，单独搭建的话需要进行各种配置比较麻烦,我本地搭建的版本为7.16.1，如果需要更改版本或者更改其他配置可以直接更改下面docker-compose.yaml文件

docker-compose.yaml

services:elasticsearch:image: elasticsearch:7.16.1container_name: esenvironment:discovery.type: single-nodeES_JAVA_OPTS: "-Xms512m -Xmx512m"ports:- "9200:9200"- "9300:9300"healthcheck:test: ["CMD-SHELL", "curl --silent --fail localhost:9200/_cluster/health || exit 1"]interval: 10stimeout: 10sretries: 3networks:- elasticlogstash:image: logstash:7.16.1container_name: logenvironment:discovery.seed_hosts: logstashLS_JAVA_OPTS: "-Xms512m -Xmx512m"volumes:- ./logstash/pipeline/logstash-nginx.config:/usr/share/logstash/pipeline/logstash-nginx.config- ./logstash/nginx.log:/home/nginx.logports:- "5000:5000/tcp"- "5000:5000/udp"- "5044:5044"- "9600:9600"depends_on:- elasticsearchnetworks:- elasticcommand: logstash -f /usr/share/logstash/pipeline/logstash-nginx.configkibana:image: kibana:7.16.1container_name: kibports:- "5601:5601"depends_on:- elasticsearchnetworks:- elastic
networks:elastic:driver: bridge

Deploy with docker compose

此处不再详细讲解docker-compse如何部署了

$ docker compose up -d
Creating network "elasticsearch-logstash-kibana_elastic" with driver "bridge"
Creating es ... done
Creating log ... done
Creating kib ... done

Expected result

Listing containers must show three containers running and the port mapping as below:

$ docker ps
CONTAINER ID        IMAGE                 COMMAND                  CREATED             STATUS                    PORTS                                                                                            NAMES
173f0634ed33        logstash:7.8.0        "/usr/local/bin/dock…"   43 seconds ago      Up 41 seconds             0.0.0.0:5000->5000/tcp, 0.0.0.0:5044->5044/tcp, 0.0.0.0:9600->9600/tcp, 0.0.0.0:5000->5000/udp   log
b448fd3e9b30        kibana:7.8.0          "/usr/local/bin/dumb…"   43 seconds ago      Up 42 seconds             0.0.0.0:5601->5601/tcp                                                                           kib
366d358fb03d        elasticsearch:7.8.0   "/tini -- /usr/local…"   43 seconds ago      Up 42 seconds (healthy)   0.0.0.0:9200->9200/tcp, 0.0.0.0:9300->9300/tcp

After the application starts, navigate to below links in your web browser:

  • Elasticsearch: http://localhost:9200
  • Logstash: http://localhost:9600
  • Kibana: http://localhost:5601/api/status

准备工作

  • 进入kibana，新增测试索引
PUT /test_canal
{"mappings":{"properties":{"id": {"type": "long"},"name": {"type": "text"}}
}
}

搭建canal-server

推荐使用官网下载release版本进行解压部署，官网地址：https://github.com/alibaba/canal/wiki/QuickStart，也可以使用docker进行部署，重点是部署完成后需要修改配置文件。

  • 更改配置
vi conf/example/instance.properties

## mysql serverId
canal.instance.mysql.slaveId = 1234
#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.standby.address = 
#canal.instance.standby.journal.name =
#canal.instance.standby.position = 
#canal.instance.standby.timestamp = 
#username/password，需要改成自己的数据库信息
canal.instance.dbUsername = canal  
canal.instance.dbPassword = canal
canal.instance.defaultDatabaseName =
canal.instance.connectionCharset = UTF-8
#table regex
canal.instance.filter.regex = .\*\\\\..\*

具体配置文件介绍可以查看：https://github.com/alibaba/canal/wiki/AdminGuide

注意： 使用docker部署时，如果没有设置为同一个网络，数据库地址无法使用127.0.0.1

  • 启动
sh bin/startup.sh

搭建canal-adapter

基本功能

canal 1.1.1版本之后, 增加客户端数据落地的适配及启动功能, 目前支持功能:

  • 客户端启动器
  • 同步管理REST接口
  • 日志适配器, 作为DEMO
  • 关系型数据库的数据同步(表对表同步), ETL功能
  • HBase的数据同步(表对表同步), ETL功能
  • (后续支持) ElasticSearch多表数据同步,ETL功能

适配器整体结构

client-adapter分为适配器和启动器两部分, 适配器为多个fat jar, 每个适配器会将自己所需的依赖打成一个包, 以SPI的方式让启动器动态加载, 目前所有支持的适配器都放置在plugin目录下

启动器为 SpringBoot 项目, 支持canal-client启动的同时提供相关REST管理接口, 运行目录结构为:

- binrestart.shstartup.batstartup.shstop.sh
- lib...
- plugin client-adapter.logger-1.1.1-jar-with-dependencies.jarclient-adapter.hbase-1.1.1-jar-with-dependencies.jar...
- confapplication.yml- hbasemytest_person2.yml
- logs

以上目录结构最终会打包成 canal-adapter-*.tar.gz 压缩包

适配器配置介绍

总配置文件 application.yml

adapter定义配置部分

canal.conf:canalServerHost: 127.0.0.1:11111          # 对应单机模式下的canal server的ip:portzookeeperHosts: slave1:2181               # 对应集群模式下的zk地址, 如果配置了canalServerHost, 则以canalServerHost为准mqServers: slave1:6667 #or rocketmq       # kafka或rocketMQ地址, 与canalServerHost不能并存flatMessage: true                         # 扁平message开关, 是否以json字符串形式投递数据, 仅在kafka/rocketMQ模式下有效batchSize: 50                             # 每次获取数据的批大小, 单位为KsyncBatchSize: 1000                       # 每次同步的批数量retries: 0                                # 重试次数, -1为无限重试timeout:                                  # 同步超时时间, 单位毫秒mode: tcp # kafka rocketMQ                # canal client的模式: tcp kafka rocketMQsrcDataSources:                           # 源数据库defaultDS:                              # 自定义名称url: jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true   # jdbc url username: root                                            # jdbc 账号password: 121212                                          # jdbc 密码canalAdapters:                            # 适配器列表- instance: example                       # canal 实例名或者 MQ topic 名groups:                                 # 分组列表- groupId: g1                           # 分组id, 如果是MQ模式将用到该值outerAdapters:                        # 分组内适配器列表- name: logger                        # 日志打印适配器
......

说明:

  1. 一份数据可以被多个group同时消费, 多个group之间会是一个并行执行, 一个group内部是一个串行执行多个outerAdapters, 比如例子中logger和hbase
  2. 目前client adapter数据订阅的方式支持两种，直连canal server 或者 订阅kafka/RocketMQ的消息

操作流程

从官网下载relaes版本之后进行解压，解压之后修改cnf目录下得application.yml文件,根据自己得需求选择下面不同的adapters进行配置，我本地安装的es是es7.16，选择es7的配置，具体配置的含义请参考上方适配器配置介绍

  • 配置application.yaml
server:port: 8081
spring:jackson:date-format: yyyy-MM-dd HH:mm:sstime-zone: GMT+8default-property-inclusion: non_nullcanal.conf:mode: tcp #tcp kafka rocketMQ rabbitMQflatMessage: truezookeeperHosts:syncBatchSize: 1000retries: -1timeout:accessKey:secretKey:consumerProperties:# canal tcp consumercanal.tcp.server.host: 127.0.0.1:11111canal.tcp.zookeeper.hosts:canal.tcp.batch.size: 500canal.tcp.username:canal.tcp.password:# kafka consumerkafka.bootstrap.servers: 127.0.0.1:9092kafka.enable.auto.commit: falsekafka.auto.commit.interval.ms: 1000kafka.auto.offset.reset: latestkafka.request.timeout.ms: 40000kafka.session.timeout.ms: 30000kafka.isolation.level: read_committedkafka.max.poll.records: 1000# rocketMQ consumerrocketmq.namespace:rocketmq.namesrv.addr: 127.0.0.1:9876rocketmq.batch.size: 1000rocketmq.enable.message.trace: falserocketmq.customized.trace.topic:rocketmq.access.channel:rocketmq.subscribe.filter:# rabbitMQ consumerrabbitmq.host:rabbitmq.virtual.host:rabbitmq.username:rabbitmq.password:rabbitmq.resource.ownerId:srcDataSources:defaultDS:url: jdbc:mysql://127.0.0.1:3306/test?useUnicode=true # 配置数据库地址username: rootpassword: 123456canalAdapters:- instance: example # canal instance Name or mq topic namegroups:- groupId: g1outerAdapters:- name: logger
#      - name: rdb
#        key: mysql1
#        properties:
#          jdbc.driverClassName: com.mysql.jdbc.Driver
#          jdbc.url: jdbc:mysql://127.0.0.1:3306/mytest2?useUnicode=true
#          jdbc.username: root
#          jdbc.password: 121212
#          druid.stat.enable: false
#          druid.stat.slowSqlMillis: 1000
#      - name: rdb
#        key: oracle1
#        properties:
#          jdbc.driverClassName: oracle.jdbc.OracleDriver
#          jdbc.url: jdbc:oracle:thin:@localhost:49161:XE
#          jdbc.username: mytest
#          jdbc.password: m121212
#      - name: rdb
#        key: postgres1
#        properties:
#          jdbc.driverClassName: org.postgresql.Driver
#          jdbc.url: jdbc:postgresql://localhost:5432/postgres
#          jdbc.username: postgres
#          jdbc.password: 121212
#          threads: 1
#          commitSize: 3000
#      - name: hbase
#        properties:
#          hbase.zookeeper.quorum: 127.0.0.1
#          hbase.zookeeper.property.clientPort: 2181
#          zookeeper.znode.parent: /hbase- name: es7hosts:  172.22.80.1:9300 # 127.0.0.1:9200 for rest modeproperties:mode: transport # or rest# security.auth: test:123456 #  only used for rest modecluster.name: docker-cluster
#      - name: kudu
#        key: kudu
#        properties:
#          kudu.master.address: 127.0.0.1 # ',' split multi address
#      - name: phoenix
#        key: phoenix
#        properties:
#          jdbc.driverClassName: org.apache.phoenix.jdbc.PhoenixDriver
#          jdbc.url: jdbc:phoenix:127.0.0.1:2181:/hbase/db
#          jdbc.username:
#          jdbc.password:

  • 新增测试yml文件，再es7目录下新建testcanal.yml配置，更加复杂的映射处理可以参考canal官网示例：

    https://github.com/alibaba/canal/wiki/Sync-ES

在这里插入图片描述

  • 启动canal-adapter启动器
bin/startup.sh

windows启动.bat文件

  • 验证

新增mysql mytest.test_canal表的数据, 将会自动同步到es的test_canal索引下面, 并会打出DML的log

番外：adapter管理REST接口

注意：部分操作需要指定cnf下的yml配置

查询所有订阅同步的canal instance或MQ topic
curl http://127.0.0.1:8081/destinations
数据同步开关
curl http://127.0.0.1:8081/syncSwitch/example/off -X PUT

针对 example 这个canal instance/MQ topic 进行开关操作. off代表关闭, instance/topic下的同步将阻塞或者断开连接不再接收数据, on代表开启

注: 如果在配置文件中配置了 zookeeperHosts 项, 则会使用分布式锁来控制HA中的数据同步开关, 如果是单机模式则使用本地锁来控制开关

数据同步开关状态
curl http://127.0.0.1:8081/syncSwitch/example

查看指定 canal instance/MQ topic 的数据同步开关状态

手动ETL
  • 不带参数
curl http://127.0.0.1:8081/etl/es7/testcanal.yml -X POST
  • 带参数
curl http://127.0.0.1:8081/etl/hbase/mytest_person2.yml -X POST -d "params=2018-10-21 00:00:00"

导入数据到指定类型的库, 如果params参数为空则全表导入, 参数对应的查询条件在配置中的etlCondition指定

查看相关库总数据
curl http://127.0.0.1:8081/count/hbase/mytest_person2.yml

常见问题

not part of the cluster Cluster [elasticsearch], ignoring…

查看canal-adapter配置中的application.yml配置是否和es中的cluster_name配置相同
在这里插入图片描述
在这里插入图片描述

连接问题

如果本地使用docker进行部署，未指定网桥，切记不能使用127.0.0.1进行es连接或者mysql数据库连接

这篇关于本地搭建部署Canal笔记-实现MySQL与ElasticSearch7数据同步的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/583565。

相关文章

Android实现悬浮按钮功能

Android实现悬浮按钮功能

《Android实现悬浮按钮功能》在很多场景中,我们希望在应用或系统任意界面上都能看到一个小的“悬浮按钮”（FloatingButton）,用来快速启动工具、展示未读信息或快捷操作,所以本文给大家介绍... 目录一、项目概述二、相关技术知识三、实现思路四、整合代码4.1 Java 代码（MainActivi
阅读更多...
使用Python实现一个优雅的异步定时器

使用Python实现一个优雅的异步定时器

《使用Python实现一个优雅的异步定时器》在Python中实现定时器功能是一个常见需求,尤其是在需要周期性执行任务的场景下,本文给大家介绍了基于asyncio和threading模块,可扩展的异步定... 目录需求背景代码1. 单例事件循环的实现2. 事件循环的运行与关闭3. 定时器核心逻辑4. 启动与停
阅读更多...
基于Python实现读取嵌套压缩包下文件的方法

基于Python实现读取嵌套压缩包下文件的方法

《基于Python实现读取嵌套压缩包下文件的方法》工作中遇到的问题,需要用Python实现嵌套压缩包下文件读取,本文给大家介绍了详细的解决方法,并有相关的代码示例供大家参考,需要的朋友可以参考下... 目录思路完整代码代码优化思路打开外层zip压缩包并遍历文件：使用with zipfile.ZipFil
阅读更多...
Python实现word文档内容智能提取以及合成

Python实现word文档内容智能提取以及合成

《Python实现word文档内容智能提取以及合成》这篇文章主要为大家详细介绍了如何使用Python实现从10个左右的docx文档中抽取内容,再调整语言风格后生成新的文档,感兴趣的小伙伴可以了解一下... 目录核心思路技术路径实现步骤阶段一：准备工作阶段二：内容提取 (python 脚本)阶段三：语言风格调
阅读更多...
C#实现将Excel表格转换为图片(JPG/ PNG)

C#实现将Excel表格转换为图片(JPG/ PNG)

《C#实现将Excel表格转换为图片(JPG/PNG)》Excel表格可能会因为不同设备或字体缺失等问题,导致格式错乱或数据显示异常,转换为图片后,能确保数据的排版等保持一致,下面我们看看如何使用C... 目录通过C# 转换Excel工作表到图片通过C# 转换指定单元格区域到图片知识扩展C# 将 Excel
阅读更多...
基于Java实现回调监听工具类

基于Java实现回调监听工具类

《基于Java实现回调监听工具类》这篇文章主要为大家详细介绍了如何基于Java实现一个回调监听工具类,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下... 目录监听接口类 Listenable实际用法打印结果首先，会用到 函数式接口 Consumer, 通过这个可以解耦回调方法，下面先写一个
阅读更多...
使用Java将DOCX文档解析为Markdown文档的代码实现

使用Java将DOCX文档解析为Markdown文档的代码实现

《使用Java将DOCX文档解析为Markdown文档的代码实现》在现代文档处理中,Markdown（MD）因其简洁的语法和良好的可读性,逐渐成为开发者、技术写作者和内容创作者的首选格式,然而,许多文... 目录引言1. 工具和库介绍2. 安装依赖库3. 使用Apache POI解析DOCX文档4. 将解析
阅读更多...
Qt中QGroupBox控件的实现

Qt中QGroupBox控件的实现

《Qt中QGroupBox控件的实现》QGroupBox是Qt框架中一个非常有用的控件,它主要用于组织和管理一组相关的控件,本文主要介绍了Qt中QGroupBox控件的实现,具有一定的参考价值,感兴趣... 目录引言一、基本属性二、常用方法2.1 构造函数 2.2 设置标题2.3 设置复选框模式2.4 是否
阅读更多...
C++使用printf语句实现进制转换的示例代码

C++使用printf语句实现进制转换的示例代码

《C++使用printf语句实现进制转换的示例代码》在C语言中,printf函数可以直接实现部分进制转换功能,通过格式说明符（formatspecifier）快速输出不同进制的数值,下面给大家分享C+... 目录一、printf 原生支持的进制转换1. 十进制、八进制、十六进制转换2. 显示进制前缀3. 指
阅读更多...
springboot整合阿里云百炼DeepSeek实现sse流式打印的操作方法

springboot整合阿里云百炼DeepSeek实现sse流式打印的操作方法

《springboot整合阿里云百炼DeepSeek实现sse流式打印的操作方法》：本文主要介绍springboot整合阿里云百炼DeepSeek实现sse流式打印,本文给大家介绍的非常详细,对大... 目录1.开通阿里云百炼,获取到key2.新建SpringBoot项目3.工具类4.启动类5.测试类6.测
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2