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

背景

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

建议首先熟悉一下canal同步方式:https://github.com/alibaba/canal/wiki

前提条件

  • 本地搭建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_increment
        primary key,
    name text null
);

搭建ELK

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

docker-compose.yaml

services:
  elasticsearch:
    image: elasticsearch:7.16.1
    container_name: es
    environment:
      discovery.type: single-node
      ES_JAVA_OPTS: "-Xms512m -Xmx512m"
    ports:
      - "9200:9200"
      - "9300:9300"
    healthcheck:
      test: ["CMD-SHELL", "curl --silent --fail localhost:9200/_cluster/health || exit 1"]
      interval: 10s
      timeout: 10s
      retries: 3
    networks:
      - elastic
  logstash:
    image: logstash:7.16.1
    container_name: log
    environment:
      discovery.seed_hosts: logstash
      LS_JAVA_OPTS: "-Xms512m -Xmx512m"
    volumes:
      - ./logstash/pipeline/logstash-nginx.config:/usr/share/logstash/pipeline/logstash-nginx.config
      - ./logstash/nginx.log:/home/nginx.log
    ports:
      - "5000:5000/tcp"
      - "5000:5000/udp"
      - "5044:5044"
      - "9600:9600"
    depends_on:
      - elasticsearch
    networks:
      - elastic
    command: logstash -f /usr/share/logstash/pipeline/logstash-nginx.config
  kibana:
    image: kibana:7.16.1
    container_name: kib
    ports:
      - "5601:5601"
    depends_on:
      - elasticsearch
    networks:
      - 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管理接口, 运行目录结构为:

- bin
    restart.sh
    startup.bat
    startup.sh
    stop.sh
- lib
   ...
- plugin 
    client-adapter.logger-1.1.1-jar-with-dependencies.jar
    client-adapter.hbase-1.1.1-jar-with-dependencies.jar
    ...
- conf
    application.yml
    - hbase
        mytest_person2.yml
- logs

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

适配器配置介绍

总配置文件 application.yml

adapter定义配置部分

canal.conf:
  canalServerHost: 127.0.0.1:11111          # 对应单机模式下的canal server的ip:port
  zookeeperHosts: slave1:2181               # 对应集群模式下的zk地址, 如果配置了canalServerHost, 则以canalServerHost为准
  mqServers: slave1:6667 #or rocketmq       # kafka或rocketMQ地址, 与canalServerHost不能并存
  flatMessage: true                         # 扁平message开关, 是否以json字符串形式投递数据, 仅在kafka/rocketMQ模式下有效
  batchSize: 50                             # 每次获取数据的批大小, 单位为K
  syncBatchSize: 1000                       # 每次同步的批数量
  retries: 0                                # 重试次数, -1为无限重试
  timeout:                                  # 同步超时时间, 单位毫秒
  mode: tcp # kafka rocketMQ                # canal client的模式: tcp kafka rocketMQ
  srcDataSources:                           # 源数据库
    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:ss
    time-zone: GMT+8
    default-property-inclusion: non_null

canal.conf:
  mode: tcp #tcp kafka rocketMQ rabbitMQ
  flatMessage: true
  zookeeperHosts:
  syncBatchSize: 1000
  retries: -1
  timeout:
  accessKey:
  secretKey:
  consumerProperties:
    # canal tcp consumer
    canal.tcp.server.host: 127.0.0.1:11111
    canal.tcp.zookeeper.hosts:
    canal.tcp.batch.size: 500
    canal.tcp.username:
    canal.tcp.password:
    # kafka consumer
    kafka.bootstrap.servers: 127.0.0.1:9092
    kafka.enable.auto.commit: false
    kafka.auto.commit.interval.ms: 1000
    kafka.auto.offset.reset: latest
    kafka.request.timeout.ms: 40000
    kafka.session.timeout.ms: 30000
    kafka.isolation.level: read_committed
    kafka.max.poll.records: 1000
    # rocketMQ consumer
    rocketmq.namespace:
    rocketmq.namesrv.addr: 127.0.0.1:9876
    rocketmq.batch.size: 1000
    rocketmq.enable.message.trace: false
    rocketmq.customized.trace.topic:
    rocketmq.access.channel:
    rocketmq.subscribe.filter:
    # rabbitMQ consumer
    rabbitmq.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: root
      password: 123456
  canalAdapters:
  - instance: example # canal instance Name or mq topic name
    groups:
    - groupId: g1
      outerAdapters:
      - 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: es7
        hosts:  172.22.80.1: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: 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数据库连接

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mfbz.cn/a/304014.html

如若内容造成侵权/违法违规/事实不符,请联系我们进行投诉反馈qq邮箱809451989@qq.com,一经查实,立即删除!

相关文章

shp与数据库(插入数据)

前言 正文 geopandas与shp文件创建表和录入数据 解释一下上面的代码 查询cd2表的geometry字段 查看一下表 前一篇博客的冲突 问题的解决 POLYGON与MUTLIPOLYGON的说明 解决问题的代码 修改表的创建 shapefile和sqlalchemy插入数据 数据库查询 最后 前言 前一篇讲解…

与AI合作 -- 写一个modern c++单例工厂

目录 前言 提问 bard给出的答案 AI答案的问题 要求bard改进 人类智能 AI VS 人类 前言 通过本文读者可以学到modern C单例模式工厂模式的混合体,同时也能看到:如今AI发展到了怎样的智能程度?怎样让AI帮助我们快速完成实现头脑中的想法&…

【hcie-cloud】【17】华为云Stack灾备服务介绍【灾备方案概述、备份解决方案介绍】【上】

文章目录 前言灾备方案概述灾备的定义灾备的重要性故障和灾难对业务连续性带来的挑战灾备系统的衡量指标RTO与RPO分析 灾备等级标准数据中心容灾解决方案全景图云灾备服务总结架构华为云Stack灾备服务总览 备份解决方案介绍云备份服务介绍备份服务架构介绍云备份服务组件功能介…

界面原型设计工具有哪些?看看这9个

界面原型设计是现代设计师必备的技能之一。在设计数字产品或应用程序时,界面原型是将概念转化为具体可交互界面的重要步骤。对于新手小白来说,选择一款易于上手且功能强大的界面原型设计工具至关重要。本文将介绍 9 个常用的界面原型设计工具&#xff0c…

计算机体系结构动态调度(计分板及Tomasulo)学习记录

1.动态调度核心思想:允许就绪指令越过前方停顿指令,提前进入运行(乱序执行) 就绪指令指不存在资源冲突、操作数已就绪的指令,例如,计分板算法使用计分板来实现,Tomasulo使用保留站来实现&#…

苹果电脑交互式原型设计软件Axure RP 9 mac特色介绍

Axure RP 9 for Mac是一款交互式原型设计软件,使用axure rp9以最佳的方式展示您的作品,优化现代浏览器并为现代工作流程设计。同时确保您的解决方案正确完整地构建。Axure RP 9 for Mac为您整理笔记,将其分配给UI元素,并合并屏幕注…

swing快速入门(三十九)进度对话框

🎁注释很详细,直接上代码 上一篇 🧧新增内容 🧨1.模拟耗时操作 🧨2.使用计时器更新进度对话框 🎀源码: package swing31_40;import javax.swing.*; import java.awt.event.ActionEvent; import …

【debug】为什么ansible中使用command出错

碎碎念 在使用ansible执行command的时候,遇到执行会出错的command 比如执行source打算读取环境变量的时候 错误提示为: 没有那个文件或目录:source 一开始以为是错误提示有问题,一直在testrc的路径上检查,但是同样一行命令使用…

C++八股学习心得.8

1.const知道吗?解释其作用 const 修饰类的成员变量,表示成员常量,不能被修改。 const修饰函数承诺在本函数内部不会修改类内的数据成员,不会调用其它非 const 成员函数。 如果 const 构成函数重载,const 对象只能调…

聊聊 Java 集合框架中的 ArrayList

其实 Java 集合框架也叫做容器,主要由两大接口派生而来,一个是 collection,主要存放对象的集合。另外一个是Map, 存储着键值对(两个对象)的映射表。 下面就来说说 List接口,List存储的元素是有序、可重复的。其下有三个…

配置git服务器

第一步: jdk环境配置 (1)搜索【高级系统设置】,选择【高级】选项卡,点【环境变量】 (2)在【系统变量】里面,点击【新建】 (3)添加JAVA_HOME环境变量JAVA_HO…

展台搭建的基本要求

一、选址及总平面布置 展厅人流量大,对选址和总平面布置的要求如下:展厅选址应位于市内或郊区交通便利的区域。大型展览馆应有足够的群众活动广场和停车面积 ,并应有室外陈列场地。室外场地要考虑环境的绿化和美化。各功能分区之间联系方便又互不干扰。建…

实现LCM在docker之间的通信

目录 1.docker容器互联 新建网络 连接容器 2.设置环境变量 3.在两个docker之间实现通信 1.docker容器互联 新建网络 $ docker network create -d bridge test-net 连接容器 运行一个容器并连接到新建的 test-net 网络: $ docker run -itd --name lcm_1 --network tes…

极智项目 | 实战Paddle戴口罩检测

欢迎关注我的公众号 [极智视界],获取我的更多项目分享 大家好,我是极智视界,本文来介绍 实战 Paddle 戴口罩检测项目。 本文介绍的 实战 Paddle 戴口罩检测项目,提供完整的可以一键执行的项目工程源码,获取方式有两个…

AI红娘开启约会新时代;网易云音乐Agent实践探索;微软生成式AI课程要点笔记;ComfyUI新手教程;图解RAG进阶技术 | ShowMeAI日报

👀日报&周刊合集 | 🎡生产力工具与行业应用大全 | 🧡 点赞关注评论拜托啦! 👀 Perplexity 官宣 7360 万美元B轮融资,打造世界上最快最准确的答案平台 https://blog.perplexity.ai/blog/perplexity-rais…

yyds,4 个 Pandas 必备神器!

Pandas是我们日常处理表格数据最常用的包,但是对于数据分析来说,Pandas的DataFrame还不够直观。 今天我们将介绍4个和Pandas相关的Python包,可以将Pandas的DataFrame转换交互式表格,让我们可以直接在上面进行数据分析的操作。 P…

天洑软件与常州大学成立电子信息专业学位硕士研究生联合培养基地

2023年12月27日,天洑软件与常州大学微电子与控制工程学院成立电子信息专业学位硕士研究生联合培养基地,授牌及企业导师聘任仪式于常州大学西太湖校区隆重举办。天洑软件前瞻技术研究院院长张儒受邀与会。 开幕式上,微电子与控制工程学院党委书…

NUXT3学习笔记

1.邂逅SPA、SSR 1.1 单页面应用程序 单页应用程序 (SPA) 全称是:Single-page application,SPA应用是在客户端呈现的(术语称:CSR(Client Side Render)) SPA的优点 只需加载一次 SPA应用程序只需…

Docker实战08|Docker管道及环境变量识别

上一篇文章中,讲解了如何通过Go语言实现对Docker Cgroup的资源限制 具体文章可见《Docker就应该这么学-07》 有需要的小伙伴可以回顾一下。 接下来本文会详细介绍一下Docker 管道及环境变量识别 管道及环境变量识别 获取代码 git clone https://gitee.com/mjr…

SEO全自动发布外链工具源码系统:自动增加权重 附带完整的搭建安装教程

SEO全自动发布外链工具是一款基于PHP和MySQL开发的外链发布工具。它通过自动化流程,帮助站长快速、有效地发布外链,提高网站的权重和排名。该工具支持多种外链发布平台,如论坛、博客、分类信息等,可自定义发布内容和格式&#xff…
最新文章