基于 cancal 进行 mysql 与 es 的数据同步

2022-09-23

other

字数统计: 1.7k | 阅读时长≈ 9 分钟

基于 cancal 进行 mysql 与 elasticsearch 的数据同步

1、mysql 安装

2、建库、建表

3、elasticsearch 安装

4、canal-server 安装

5、联合测试

mysql 安装

# 安装 mysql 5.7.36
docker pull 5.7.36
docker run --name mysql5736 -p 3306:3306 -e MYSQL_ROOT_PASSWORD=root -d mysql:5.7.36

# 将配置文件拷出，进行修改
docker cp docker cp mysql5736:/etc/mysql/mysql.conf.d/mysqld.cnf /docker/mysql5736/config

#修改配置文件

# binlog setting
# 开启 logbin
log-bin=mysql-bin
# binlog 日志格式
binlog-format=ROW
# mysql 主从备份 serverId,canal 中不能与此相同
server-id=1

# 进行配置文件的挂载
docker run --name mysql5736 -p 3306:3306 -e MYSQL_ROOT_PASSWORD=root -v /docker/mysql5736/config/mysqld.cnf:/etc/mysql/mysql.conf.d/mysqld.cnf -v /docker/mysql5736/data:/var/lib/mysql -d mysql:5.7.36

docker exec -it mysql5736 /bin/bash

mysql 容器中操作

# mysql: [Warning] World-writable config file '/etc/mysql/mysql.conf.d/mysqld.cnf' is ignored. 遇到这个问题 修改文件的权限  chmod 644 /etc/mysql/mysql.conf.d/mysqld.cnf

# 登录
mysql -uroot -p
# binlog日志文件
show master status;
# 重启日志
reset master
# 查看日志文件格式
show variables like 'binlog_format';

建库、建表

-- 新建 dailyhub 数据库
-- 运行建表 sql

DROP TABLE IF EXISTS `m_collect`;
CREATE TABLE `m_collect` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  `collected` date DEFAULT NULL,
  `created` datetime(6) DEFAULT NULL,
  `note` varchar(255) DEFAULT NULL,
  `personal` int(11) DEFAULT NULL,
  `title` varchar(255) DEFAULT NULL,
  `url` varchar(255) DEFAULT NULL,
  `user_id` bigint(20) DEFAULT NULL,
  PRIMARY KEY (`id`),
  KEY `FK6yx2mr7fgvv204y8jw5ubsn7h` (`user_id`),
  CONSTRAINT `FK6yx2mr7fgvv204y8jw5ubsn7h` FOREIGN KEY (`user_id`) REFERENCES `m_user` (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=7 DEFAULT CHARSET=utf8mb4;

DROP TABLE IF EXISTS `m_user`;
CREATE TABLE `m_user` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  `avatar` varchar(255) DEFAULT NULL,
  `created` datetime(6) DEFAULT NULL,
  `lasted` datetime(6) DEFAULT NULL,
  `open_id` varchar(255) DEFAULT NULL,
  `username` varchar(255) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=4 DEFAULT CHARSET=utf8mb4;

elasticsearch 安装

docker pull elasticsearch:7.16.2

docker run -p 9200:9200 -p 9300:9300 -e "discovery.type=single-node" --name='es7162' -d elasticsearch:7.16.2


# 将配置文件拷出进行修改
docker cp es7162:/usr/share/elasticsearch/config/elasticsearch.yml /docker/elasticsearch/config

1、yml 配置文件修改

cluster.name: dailyhub-es
network.host: 0.0.0.0

node.name: node-1
http.port: 9200
http.cors.enabled: true
http.cors.allow-origin: "*"
node.master: true
node.data: true

2、安装 ik 分词器

搜索 ik分词器，下载对应 elasticsearch 版本的 ik 分词器，解压置于本地目录中，便于下一步挂载 plugins 目录

3、启动

1
2

# 删除之前的镜像重新启动
docker run -p 9200:9200 -p 9300:9300 -e "discovery.type=single-node" -v /docker/elasticsearch/config/elasticsearch.yml:/usr/share/elasticsearch/config/elasticsearch.yml -v /docker/elasticsearch/plugins:/usr/share/elasticsearch/plugins --name='es7162' -d elasticsearch:7.16.2

4、测试

1 2	<!-- 测试地址 --> http://localhost:9200/_mapping?pretty=true

安装 canal-server

# 拉取镜像
docker pull canal/canal-server:v1.1.5

# 网络桥接 mysql
docker run --name canal115 -p 11111:11111 --link mysql5736:mysql5736 -id canal/canal-server:v1.1.5

docker cp  canal115:/home/admin/canal-server/conf/example/instance.properties /docker/canal/config

5、修改配置文件

# 不和 mysql 中的 id 相同
canal.instance.mysql.slaveId=10
# mysql5736 是 mysql 镜像的链接别名
canal.instance.master.address=mysql5736:3306
canal.instance.dbUsername=root
canal.instance.dbPassword=root

6、重新运行容器

docker rm -f canal115
# 挂载配置文件
docker run --name canal115 -p 11111:11111 -v /docker/canal/config/instance.properties:/home/admin/canal-server/conf/example/instance.properties --link mysql5736:mysql5736 -id canal/canal-server:v1.1.5

docker exec -it canal115 /bin/bash
# 查看日志是否连接成功
tail -100f canal-server/logs/example/example.log

canal-adapter 安装

# 拉取镜像，非官方源
docker pull slpcat/canal-adapter:v1.1.5

docker run --name adapter115 -p 8081:8081 --link mysql5736:mysql5736 --link canal115:canal115 --link es7162:es7162 -d slpcat/canal-adapter:v1.1.5

docker cp adapter115:/opt/canal-adapter/conf/application.yml /docker/canaladapter/config

docker cp adapter115:/opt/canal-adapter/conf/es7/mytest_user.yml /docker/canaladapter/es7

7、修改配置

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:
      url: jdbc:mysql://mysql5736:3306/dailyhub?useUnicode=true
      username: root
      password: root
  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 127.0.0.1:9300 for transport mode
        properties:
          mode: rest #transport or rest
          # security.auth: test:123456 #  only used for rest mode
          cluster.name: dailyhub-es

修改完之后，可重命名文件名

dataSourceKey: defaultDS
destination: example
groupId: g1
esMapping:
  _index: dailyhub_collect
  _id: _id
 upsert: true
#  pk: id
  sql: "
SELECT
	c.id AS _id,
	c.title AS title,
	c.user_id AS userId,
	c.note AS note,
	c.url AS url,
	c.created AS created,
	c.collected AS collected,
	c.personal AS personal,
	u.username AS username,
	u.avatar AS userAvatar
FROM
	m_collect c
	LEFT JOIN m_user u ON c.user_id = u.id"
#  objFields:
#    _labels: array:;
  etlCondition: "where a.c_time>={}"
  commitBatch: 3000

8、重新运行容器

1
2

# 删除原来的容器重新运行容器
docker run --name adapter115 -p 8081:8081 --link mysql5736:mysql5736 --link canal115:canal115 --link es7162:es7162 -v /docker/canaladapter/config/application.yml:/opt/canal-adapter/conf/application.yml -v /docker/canaladapter/es7:/opt/canal-adapter/conf/es7 -d slpcat/canal-adapter:v1.1.5

联合测试

9、新建索引

使用 postman 发送请求到 es 新建索引

PUT http://192.168.31.21:9200/dailyhub_collect

附带 json 数据如下（主要描述各字段类型）：

{
    "mappings":{
        "properties":{
            "collected":{
                "type":"date",
                "format":"date_optional_time||epoch_millis"
            },
            "created":{
                "type":"date",
                "format":"date_optional_time||epoch_millis"
            },
            "note":{
                "type":"text",
                "analyzer":"ik_max_word",
                "search_analyzer":"ik_smart"
            },
            "personal":{
                "type":"integer"
            },
            "title":{
                "type":"text",
                "analyzer":"ik_max_word",
                "search_analyzer":"ik_smart"
            },
            "url":{
                "type":"text"
            },
            "userAvatar":{
                "type":"text"
            },
            "userId":{
                "type":"long"
            },
            "username":{
                "type":"keyword"
            }
        }
    }
}

10、同步数据测试

数据库对相应的表插入字段

查询 es 中的数据

GET http://192.168.31.21:9200/dailyhub_collect/_search

返回结果：

{
    "took": 617,
    "timed_out": false,
    "_shards": {
        "total": 1,
        "successful": 1,
        "skipped": 0,
        "failed": 0
    },
    "hits": {
        "total": {
            "value": 1,
            "relation": "eq"
        },
        "max_score": 1.0,
        "hits": [
            {
                "_index": "dailyhub_collect",
                "_type": "_doc",
                "_id": "2",
                "_score": 1.0,
                "_source": {
                    "title": "笔记111",
                    "userId": 1,
                    "note": "这是个笔记",
                    "url": "111",
                    "created": "2022-01-30T21:49:17+08:00",
                    "collected": "2022-01-30",
                    "personal": 0,
                    "username": "hqz",
                    "userAvatar": "https://gis-visualization.oss-cn-beijing.aliyuncs.com/typoraImg/typora-icon.png"
                }
            }
        ]
    }
}

总结

本来是用 windows 上的 docker 进行操作的，自己电脑的虚拟机没装 centos，但是在装了 mysql 和 elasticsearch 之后，安装 canal-server 出现了问题，不管怎么启动，容器都是无提示退出，google 了一下，在 github 看到有这样的问题提交，遗憾的是，这个问题并没有人来解决，这真不知道是阿里的问题还是微软的问题了。很气，也没办法，只能在自己电脑上重新部署一套 linux 环境。
看网上视频时，作者修改配置文件是进入容器中去修改的，这样还是挺不好的，还是拷贝出配置文件再和容器挂载比较好，万一容器因为配置文件配的不对，无法启动，也无法进入容器去进行修改配置，就很麻烦。