我说下我的版本,首先要安装
mysql5.6
es 7.12
es-head
canal-adapter 1.1.5
canal-deployer1.1.5
第一步: 安装mysql 修改mysql配置文件 开启binlog日志,并且以ROW方式,开启主从模式 以及logbin的文件位置
log_bin=ON
binlog_format=ROW
server-id=1000
log-bin=/var/lib/mysql/mysql-bin
记得重新启动mysql
创建一个给canal用的一个用户
CREATE USER canal IDENTIFIED BY 'canal';
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
FLUSH PRIVILEGES;
查看值得配置是否开启
SHOW VARIABLES LIKE 'binlog-format'; -- 结果应该是ROW
SHOW VARIABLES LIKE 'log_bin'; -- 结果应该是 ON
SHOW VARIABLES LIKE '%log%'; -- 所有binlog信息
创建一个库和一个测试表,为了去canal同步到es 的一个测试表
-- 库的名称为 canal-test
-- 表
CREATE TABLE `product` (
`id` bigint(20) NOT NULL AUTO_INCREMENT,
`title` varchar(255) DEFAULT NULL,
`sub_title` varchar(255) DEFAULT NULL,
`price` decimal(10,2) DEFAULT NULL,
`pic` varchar(255) DEFAULT NULL,
PRIMARY KEY (`id`) USING BTREE
) ENGINE=InnoDB AUTO_INCREMENT=11 DEFAULT CHARSET=utf8 ROW_FORMAT=DYNAMIC
-- 加入数据
INSERT INTO product ( id, title, sub_title, price, pic ) VALUES ( 7, '小米8', ' 全面屏游戏智能手机 6GB+64GB', 1999.00, NULL );
INSERT INTO product ( id, title, sub_title, price, pic ) VALUES ( 8, '小米8', ' 全面屏游戏智能手机 6GB+64GB', 1999.00, NULL );
INSERT INTO product ( id, title, sub_title, price, pic ) VALUES ( 9, '小米8', ' 全面屏游戏智能手机 6GB+64GB', 1999.00, NULL );
INSERT INTO product ( id, title, sub_title, price, pic ) VALUES ( 10, '小米8', ' 全面屏游戏智能手机 6GB+64GB', 1999.00, NULL );
第二步: 安装es 和es-header ,具体在之前的一篇文章中;https://www.jianshu.com/p/a542a11debf6
第三步: 下载canal 的两个文件,并上传的服务器解压
canal下载地址: https://github.com/alibaba/canal/releases
下载完成后,上传到服务器解压,分别为解压文件位置
第四步: 配置cancel 关键步骤!!!!
1 解压canal.deployer-1.1.5 到 /opt/canal-server 后目录为
修改配置文件conf/example/instance.properties，按如下配置即可，主要是修改数据库相关配置；
由于我都是安装在同一台机器上,所以为127.0.0.1:3306
进入 当前的bin文件下 运行命令,启动canal服务
sh startup.sh
2 解压 canal.adapter-1.1.5 到 /opt/canal-adpter 目录为
修改配置文件conf/application.yml，按如下配置即可，主要是修改canal-server配置、数据源配置和客户端适配器配置；
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: 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/canal-test?useUnicode=true
username: canal
password: canal
canalAdapters:
- instance: example # canal instance Name or mq topic name
groups:
- groupId: g1
outerAdapters:
- name: logger
# - name: es7
# 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
# - 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: 127.0.0.1:9200 # 127.0.0.1:9200 for rest mode
properties:
mode: rest # 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
下面是配置说明
配置重点 一个是 jdbc:mysql://127.0.0.1:3306/canal-test?useUnicode=true 中的 数据库名称
第二个是es集群名称根据自己的实际的配置,我的是 docker-cluster
第三个 - name: es7 这个很重要一会儿要用
其他全部照搬即可
添加配置文件canal-adapter/conf/es7/product.yml，用于配置MySQL中的表与Elasticsearch中索引的映射关系；
dataSourceKey: defaultDS
destination: example
groupId: g1
esMapping:
_index: canal_product
_id: id
sql: "SELECT
p.id,
p.title,
p.sub_title,
p.price,
p.pic
FROM`product` p"
etlCondition: "where p.id > {}"
commitBatch: 30
其中 _index: canal_product 为要在es中创建的索引名称,很重要
_id: id 其中的id很重要 需要与 sql语句中的 p.id 中的id一致
etlCondition: "where p.id > {} 这个 {} 为入参 后续同步重要接受参数
至此,配置完毕,启动canal-adapter,
执行 sh startup.sh
接下来,在es中创建相应索引
put http://10.0.59.161:9200/canal_product
入参body
{
"mappings":{
"properties":{
"title":{
"type":"text"
},
"sub_title":{
"type":"text"
},
"pic":{
"type":"text"
},
"price":{
"type":"double"
}
}
}
}
访问截图
与数据库字段一一对应,所以名称与之前配置的 canal_product 必须一致
至此全部配置完毕
接下来,看着canal-adapter 的日志
在mysql数据库中反复执行 删除数据,添加数据操作
可以看到 canal-adapter 中已经接受到了mysql 的dml操作语句!!!!!!!
接下来,通过命令触发,让canal-adapter读取到的dml日志,同步到es的库中;
curl http://127.0.0.1:8081/etl/es7/product.yml -X POST -d "params=1"
etl 固定的
es7 之前起的名字必须对应,后续配置文件也在这里面
product.yml 配置文件名称
-d "params=1" 同步数据的条件 1 入参
查看es中的数据
get http://10.0.59.161:9200/canal_product/_search
可以看到es中已经同步到了mysql数据;
哇哇哇~~~~~~~~~~~~~~~~~~~~~~~~~~~至此,大功告成!!!!!!!!!!!!!!!!!!!!!!!!!
说实话这次实践遇到了很多问题
1 docker 中vim是没有的要安装
2 es-header 和es链接是要配置跨域的
3 canal 的 数据同步不是自动的,需要触发!!!!!!!!!!!!!
3 canal 的配置很复杂~~一个不小心就会导致同步数据的时候错误,要很小心
4 当看到canal-adapter的日志中有mysql的dml日志的时候,说明已经离成功很近了
5 当触发同步的时候报错, 找不到任务,仔细检查触发中与配置相关的每一个,一定是哪里错了
