docker环境安装mysql、canal、elasticsearch，基于binlog利用canal实现mysql的数据同步到elasticsearch中（二）

2023-07-05 809

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

本文涉及的产品

RDS Agent（兼容OpenClaw），2核4GB

RDS DuckDB + QuickBI 企业套餐，8核32GB + QuickBI 专业版

RDS MySQL DuckDB 分析主实例，基础系列 4核8GB

简介： docker环境安装mysql、canal、elasticsearch，基于binlog利用canal实现mysql的数据同步到elasticsearch中

5.安装canal-server

# 1.拉取镜像
sudo docker pull canal/canal-server:v1.1.5
# 2.创建容器并启动
# --name canal115 为容器命名为canal115
# -p 11111:11111 表示将容器的11111端口映射到主机的11111端口
# --link mysql5736:mysql5736 表示将mysql5736容器链接到canal115容器，并且在canal115容器中可以使用mysql5736作为主机名访问mysql5736容器
# -id canal/canal-server:v1.1.5 表示使用镜像canal/canal-server:v1.1.5创建并启动一个新的后台运行的容器。
sudo docker run --name canal115 \
-p 11111:11111 \
--link mysql5736:mysql5736 \
-id canal/canal-server:v1.1.5
# 3.在运行的canal115容器中打开一个新的终端会话
sudo docker exec -it canal115 /bin/bash
# 4.修改配置文件
vi canal-server/conf/example/instance.properties

# 把0改成10，只要不和mysql的id相同就行
canal.instance.mysql.slaveId=10
# 修改成mysql对应的账号密码，mysql5736就是mysql镜像的链接别名
canal.instance.master.address=mysql5736:3306
canal.instance.dbUsername=root
canal.instance.dbPassword=123456

# 5.退出canal容器会话
exit
# 6.重启canal
sudo docker restart canal115
# 7.在运行的canal115容器中打开一个新的终端会话
sudo docker exec -it canal115 /bin/bash
# 8.进入到日志 example 目录下 
cd canal-server/logs/example/
# 9.查看日志，实时查看指定日志文件的最后100行，并将新添加到日志文件中的内容实时输出到屏幕。
tail -100f example.log

截图如下，说明已经链接上了mysql主机，此时mysql中的数据变化，都会在canal中有同步。

# 10.退出日志打印
ctrl + c
# 11.退出容器
exit

6.实战开发-数据库设计

6.1 开发背景

假设有一个博客模块，即有一个博客圈，用户可以通过搜索来获取与自己输入的关键字相关的数据列表，类似于CSDN的搜索。

由于只是 demo，因此这里会设计的很简单。

6.2 数据库表设计

使用客户端连接到容器mysql5736，执行下面语句进行建表操作。

必须在安装 canal-adapter 容器前先进行建表操作，否则对 canal-adapter 修改配置后，canal-adapter会重启失败。

# 如果存在 es_demo 数据库则删除
DROP DATABASE IF EXISTS `es_demo`;
# 创建新数据库
CREATE DATABASE `es_demo`;
# 使用数据库
USE `es_demo`;
# 创建一张用户表，注意自增主键id是从1000开始
CREATE TABLE `user`(
  `id` INT NOT NULL AUTO_INCREMENT COMMENT '主键id',
  `username` VARCHAR(24) NOT NULL COMMENT '用户名',
  `icon` VARCHAR(255) NOT NULL COMMENT '头像url',
  PRIMARY KEY(`id`)
)ENGINE=INNODB AUTO_INCREMENT=1000 DEFAULT CHARSET=utf8mb4 COMMENT='用户表';
# 创建一张博客表，注意自增主键id是从1000开始
CREATE TABLE `blog`(
  `id` INT NOT NULL AUTO_INCREMENT COMMENT '主键id',
  `user_id` INT NOT NULL COMMENT '用户id（雪花算法生成）',
  `title` VARCHAR(255) NOT NULL COMMENT '标题',
  `tags` VARCHAR(64) NOT NULL COMMENT '标签',
  `introduce` VARCHAR(512) NOT NULL COMMENT '介绍',
  `content` TEXT NOT NULL COMMENT '文章内容', 
  `create_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
  `update_time` DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '修改时间',
  PRIMARY KEY(`id`),
  KEY `idx_user_create`(`user_id`,`create_time` DESC)
)ENGINE=INNODB AUTO_INCREMENT=1000 DEFAULT CHARSET=utf8mb4 COMMENT='博客信息表';

7.安装canal-adapter

# 1.镜像拉取
sudo docker pull slpcat/canal-adapter:v1.1.5

# 2.你可以选择依旧在同一台虚拟机上创建容器并启动，那么启动方式如下
sudo docker run --name adapter115 \
-p 8081:8081 \
--link mysql5736:mysql5736 \
--link canal115:canal115 \
--link es7162:es7162 \
-d slpcat/canal-adapter:v1.1.5
# 但我启动时由于服务器内存不足导致启动失败，因此我选择在另一台虚拟机上执行下面的命令
# 2.我在这里选择放到了另一台虚拟机上运行，那么就不再链接，后面修改配置文件时直接写前三个容器的所在虚拟机IP
sudo docker run --name adapter115 \
-p 8081:8081 \
-d slpcat/canal-adapter:v1.1.5

# 3.进入容器
sudo docker exec -it adapter115 /bin/bash
# 4.进入 conf 目录
cd conf/
# 5.删除原配置文件
rm -rf application.yml
# 6.新建配置文件
vi application.yml

🍀 与 mysql、es、canal-server 依旧是在一台服务器时的application.yml信息配置：

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: 0
  timeout:
  accessKey:
  secretKey:
  consumerProperties:
    # canal tcp consumer
    canal.tcp.server.host: canal115:11111
    canal.tcp.zookeeper.hosts:
    canal.tcp.batch.size: 500
    canal.tcp.username:
    canal.tcp.password:
  srcDataSources:
    defaultDS:
      # 注意mysql容器中需要存在es_demo数据库，否则重启会报错
      url: jdbc:mysql://mysql5736:3306/es_demo?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: es7162:9200 # 127.0.0.1:9200 for rest mode
        properties:
          mode: rest
          # security.auth: test:123456 #  only used for rest mode
          cluster.name: fox-es

🍀 与 mysql、es、canal-server 依旧不在同一台服务器时的application.yml信息配置：

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: 0
  timeout:
  accessKey:
  secretKey:
  consumerProperties:
    # canal tcp consumer，这个192.168.65.133就是我前三个容器所在虚拟机的IP
    canal.tcp.server.host: 192.168.65.133:11111
    canal.tcp.zookeeper.hosts:
    canal.tcp.batch.size: 500
    canal.tcp.username:
    canal.tcp.password:
  srcDataSources:
    defaultDS:
      # 这个192.168.65.133就是我前三个容器所在虚拟机的IP，注意mysql容器中需要存在es_demo数据库，否则重启会报错
      url: jdbc:mysql://192.168.65.133:3306/es_demo?useUnicode=true
      username: root
      password: 123456
  canalAdapters:
  - instance: example # canal instance Name or mq topic name
    groups:
    - groupId: g1
      outerAdapters:
      - name: logger
      - name: es7
        # 这个192.168.65.133就是我前三个容器所在虚拟机的IP
        hosts: 192.168.65.133:9200 # 127.0.0.1:9200 for rest mode
        properties:
          mode: rest
          # security.auth: test:123456 #  only used for rest mode
          cluster.name: fox-es

# 7.进入到 es7 目录下，注意此时依旧在容器中
cd /opt/canal-adapter/conf/es7
# 8.删除该目录下所有配置文件
rm -rf biz_order.yml customer.yml mytest_user.yml
# 9.新建配置文件
vi es_demo_collect.yml

dataSourceKey: defaultDS
destination: example
groupId: g1
esMapping:
  _index: es_demo_collect
  _id: _id
  _type: _doc
  upsert: true
#  pk: id
# 这里的select语句会与数据库表设计的SQL相关联，我会创建两张表user和blog，创建的SQL我会在后面的操作中写到
  sql: "
SELECT 
  b.id AS _id,
  b.user_id AS userId,
  b.title AS title,
  b.tags AS tags,
  b.introduce AS introduce,
  b.content AS content,
  b.create_time AS createTime,
  b.update_time AS updateTime,
  u.icon AS userIcon,
  u.username AS username
FROM 
  blog b
LEFT JOIN user u
ON b.user_id = u.id
"
#  objFields:
#    _labels: array:;
#   etlCondition: "where c.c_time>={}"
  commitBatch: 3000

注意对于时间类型，在后端一定要使用LocalDateTime或者LocalDate类型，如果是Date类型，需要自己手动设置格式。

# 10.退出容器
exit
# 11.重启容器
sudo docker restart adapter115
# 12.查看日志信息
sudo docker logs --tail 100 adapter115

8.实战开发-创建es索引与映射

使用Apipost建立索引与映射：【PUT】 http://192.168.65.133:9200/es_demo_collect

{
  "settings": {
    "number_of_shards": 1,
    "number_of_replicas": 0
  },
  "mappings": {
    "properties": {
      "userId": {
        "type": "long"
      },
      "username":{
        "analyzer": "ik_max_word",
        "search_analyzer": "ik_smart",
        "type": "text"  
      },
      "userIcon":{
        "type": "keyword",
        "index": false
      },
      "title": {
        "analyzer": "ik_max_word",
        "search_analyzer": "ik_smart",
        "type": "text"
      },
      "tags": {
        "analyzer": "ik_max_word",
        "search_analyzer": "ik_smart",
        "type": "text"
      },
      "introduce":{
        "analyzer": "ik_max_word",
        "search_analyzer": "ik_smart",
        "type": "text"
      },
      "content":{
        "analyzer": "ik_max_word",
        "search_analyzer": "ik_smart",
        "type": "text"
      },
      "createTime":{
        "format": "date_optional_time||epoch_millis",
        "type": "date"
      },
      "updateTime":{
        "format": "date_optional_time||epoch_millis",
        "type": "date"
      }
    }
  }
}

删除索引：DELETE http://192.168.65.133:9200/es_demo_collect

9.效果测试

9.1 新增记录测试

在 canal-adapter 容器所在虚拟器上执行：

# 实时检测 adapter 信息
sudo docker logs --tail 100  -f adapter115

对数据库进行如下SQL操作：

# 在建表时，我们两张表的自增主键id都是从1000开始的，因此在插入时 user_id 的测试数据也是从1000开始
INSERT INTO `user` VALUES(NULL,'狐狸半面添','https://sangxin-fox');
INSERT INTO `user` VALUES(NULL,'逐浪者','https://zhulang-fox');
INSERT INTO `blog`(`user_id`,`title`,`tags`,`introduce`,`content`) VALUES 
(1001,'Java语言','编程,java,语言','Java的起源','Java最初是由任职于太阳微系统的詹姆斯-高斯林（James-Gosling）等人于1990年代初开发。最初被命名为Oak；当时发现Oak被其他公司注册了，不得不重新起名，当时他们正在咖啡馆喝着印尼爪哇（JAVA）岛出产的咖啡，有人提议就叫JAVA怎么样，然后就这样定下来了。'),
(1000,'C语言','编程,C,语言','C的起源','C语言是由美国贝尔实验室的Dennis Ritchie于1972年设计发明的，最初在UNIX操作系统的DEC PDP-11计算机上使用。它由早期的编程语言BCPL（Basic Combind Programming Language）发展演变而来。');

查看 adapter 日志信息：

查看 es 信息，使用 apipost 发送请求：【GET】http://192.168.65.133:9200/es_demo_collect/_search

// 以下是该请求需要携带的json数据，表示查询es_demo_collect索引中的全部文档数据
{
    "query": {
        "match_all": {}
    }
}

docker环境安装mysql、canal、elasticsearch，基于binlog利用canal实现mysql的数据同步到elasticsearch中（二）

5.安装canal-server

6.实战开发-数据库设计

6.1 开发背景

6.2 数据库表设计

7.安装canal-adapter

8.实战开发-创建es索引与映射

9.效果测试

9.1 新增记录测试

热门文章

最新文章

相关课程

相关电子书

推荐镜像

探索云世界

热门

云计算

大数据

云原生

人工智能

数据库

开发与运维

活动广场

任务中心

训练营

直播

乘风者计划

下载

镜像站

技术资料

docker环境安装mysql、canal、elasticsearch，基于binlog利用canal实现mysql的数据同步到elasticsearch中（二）

5.安装canal-server

6.实战开发-数据库设计

6.1 开发背景

6.2 数据库表设计

7.安装canal-adapter

8.实战开发-创建es索引与映射

9.效果测试

9.1 新增记录测试

热门文章

最新文章

相关课程

相关电子书

推荐镜像