(十二)docker安装canal同步mysql8与elasticsearch7数据
docker pull canal/canal-server:v1.1.4docker run -d --name canal-server -p 11111:11111 -e canal.instance.master.address=47.98.183.103:3306 \-e canal.instance.dbUsername=root \-e canal.instance.dbPasswo
·
目录
一、简介
操作系统：Linux CentOS 7.3 64位
docker版本：19.03.8
canal-admin版本：1.1.4
canal-server版本：1.1.4
canal-adapter版本： v1.1.5-alpha-1(之所以安装这个版本，是因为这个版本支持elasticsearch7)
mysql版本：8.0.19，docker安装mysql
elasticsearch版本：7.6.2，docker安装elasticsearch
zookeeper版本：3.6.1，docker安装zookeeper
kafka版本：2.5，docker安装kafka
不了解canal的，建议前往github查看了解canal github，文档写得很详细，简单来说canal的原理是根据mysql的主从复制原理实现的，canal伪装成slave库从而向master库读取binlog日志获取增量日志。
canal-admin(非必须但推荐使用)：为canal提供整体配置管理、节点运维等面向运维的功能，提供相对友好的WebUI操作界面，方便更多用户快速和安全的操作。
canal-server：服务端，从mysql读取binlog日志获取增量日志，可以通过tcp、kafka、RocketMQ等方式与客户端通信；通过 zookeeper搭建集群。
canal-adapter：客户端，根据canal-server获取的增量日志执行适配到其他诸如elasticsearch、redis、mysql等端，实现数据同步。
在本次实践中，
zookeeper，kafka，elasticsearch，mysql的IP地址都是47.98.183.103
canal-admin，canal-server，canal-adapter的IP地址都是192.168.45.129
二、准备
1、使用canal-server需要先准备mysql，对于自建 MySQL , 需要先开启 Binlog 写入功能，配置 binlog-format 为 ROW 模式，my.cnf 中配置如下
[mysqld]
log-bin=mysql-bin # 开启 binlog
binlog-format=ROW # 选择 ROW 模式
server_id=1 # 配置 MySQL replaction 需要定义，不要和 canal 的 slaveId 重复由于我用docker安装的mysql8.0.19，默认配置好了，因此不用做配置。且如果是用的阿里云 RDS for MySQL，也不需要配置。
2、授权 canal 链接 MySQL 账号具有作为 MySQL slave 的权限, 如果已有账户可直接 grant。以下是mysql8的sql语句。
create user 'canal'@'%%' identified by 'canal';
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%%';
//mysql8需要执行这句，将加密规则还原成mysql_native_password
ALTER USER 'canal'@'%%' IDENTIFIED WITH mysql_native_password BY 'canal';
FLUSH PRIVILEGES;
show grants for 'canal'@'%%';
3、创建数据库mytest和role表、user表
CREATE TABLE `user` (
`id` int unsigned NOT NULL AUTO_INCREMENT,
`name` varchar(255) DEFAULT NULL,
`role_id` int DEFAULT NULL,
`c_time` timestamp NULL DEFAULT NULL,
PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=49 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
CREATE TABLE `role` (
`id` int unsigned NOT NULL AUTO_INCREMENT,
`role_name` varchar(255) DEFAULT NULL,
PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
三、canal-admin
1、拉取镜像
docker pull canal/canal-admin:v1.1.4
2、运行
笔者写这篇文章时，官方的运行脚本run_admin.sh下载不下来，应该是被墙了，且该脚本执行的镜像默认是latest版本，而docker仓库没有latest版本，只有v1.1.4版本；综合以上原因，这里我自己创建run_admin.sh文件，并将执行的镜像版本修改为v1.1.4，内容如下：
#!/bin/bash
function usage() {
echo "Usage:"
echo " run_admin.sh [CONFIG]"
echo "example :"
echo " run_admin.sh -e server.port=8089 \\"
echo " -e canal.adminUser=admin \\"
echo " -e canal.adminPasswd=admin"
exit
}
function check_port() {
local port=$1
local TL=$(which telnet)
if [ -f $TL ]; then
data=`echo quit | telnet 127.0.0.1 $port| grep -ic connected`
echo $data
return
fi
local NC=$(which nc)
if [ -f $NC ]; then
data=`nc -z -w 1 127.0.0.1 $port | grep -ic succeeded`
echo $data
return
fi
echo "0"
return
}
function getMyIp() {
case "`uname`" in
Darwin)
myip=`echo "show State:/Network/Global/IPv4" | scutil | grep PrimaryInterface | awk '{print $3}' | xargs ifconfig | grep inet | grep -v inet6 | awk '{print $2}'`
;;
*)
myip=`ip route get 1 | awk '{print $NF;exit}'`
;;
esac
echo $myip
}
CONFIG=${@:1}
#VOLUMNS="-v $DATA:/home/admin/canal-admin/logs"
PORTLIST="8089"
PORTS=""
for PORT in $PORTLIST ; do
#exist=`check_port $PORT`
exist="0"
if [ "$exist" == "0" ]; then
PORTS="$PORTS -p $PORT:$PORT"
else
echo "port $PORT is used , pls check"
exit 1
fi
done
NET_MODE=""
case "`uname`" in
Darwin)
bin_abs_path=`cd $(dirname $0); pwd`
;;
Linux)
bin_abs_path=$(readlink -f $(dirname $0))
NET_MODE="--net=host"
PORTS=""
;;
*)
NET_MODE="--net=host"
PORTS=""
bin_abs_path=`cd $(dirname $0); pwd`
;;
esac
BASE=${bin_abs_path}
DATA="$BASE/data"
mkdir -p $DATA
if [ $# -eq 0 ]; then
usage
elif [ "$1" == "-h" ] ; then
usage
elif [ "$1" == "help" ] ; then
usage
fi
MEMORY="-m 1024m"
LOCALHOST=`getMyIp`
cmd="docker run -d -it -h $LOCALHOST $CONFIG --name=canal-admin $VOLUMNS $NET_MODE $PORTS $MEMORY canal/canal-admin:v1.1.4"
echo $cmd
eval $cmd
执行语句如下：
# 以8089端口启动canal-admin
sh run_admin.sh -e server.port=8089 \
-e canal.adminUser=admin \
-e canal.adminPasswd=admin查看容器
3、测试和配置
账号admin，初始密码123456(登陆密码并不是admin，连接密码才是admin)
canal-admin的核心模型主要有：
- instance，对应canal-server里的instance，一个最小的订阅mysql的队列
- server，对应canal-server，一个server里可以包含多个instance
- 集群，对应一组canal-server，组合在一起面向高可用HA的运维
这里我们创建集群
接着点击操作->主配置，这个主配置其实就是个配置文件，集群里的canal-server都会统一使用这个配置文件。
点击载入模板，最终修改内容为：
#################################################
######### common argument #############
#################################################
# tcp bind ip
canal.ip =
# register ip to zookeeper 运行canal-server服务的主机IP，可以不用配置，他会自动绑定一个本机的IP
canal.register.ip =
canal.port = 11111
canal.metrics.pull.port = 11112
# canal instance user/passwd
#canal.user = canal
#canal.passwd = E3619321C1A937C46A0D8BD1DAC39F93B27D4458
# canal admin config
canal.admin.manager = 192.168.45.129:8089
canal.admin.port = 11110
canal.admin.user = admin
# 这个密码是admin，这是密码加密后的字符串
canal.admin.passwd = 4ACFE3202A5FF5CF467898FC58AAB1D615029441
#zookeeper集群地址，多个地址用逗号隔开
canal.zkServers = 47.98.183.103:2181
# flush data to zk
canal.zookeeper.flush.period = 1000
canal.withoutNetty = false
# tcp, kafka, RocketMQ 这里我们选择使用kafka
canal.serverMode = kafka
# flush meta cursor/parse position to file
canal.file.data.dir = ${canal.conf.dir}
canal.file.flush.period = 1000
## memory store RingBuffer size, should be Math.pow(2,n)
canal.instance.memory.buffer.size = 16384
## memory store RingBuffer used memory unit size , default 1kb
canal.instance.memory.buffer.memunit = 1024
## meory store gets mode used MEMSIZE or ITEMSIZE
canal.instance.memory.batch.mode = MEMSIZE
canal.instance.memory.rawEntry = true
## detecing config
canal.instance.detecting.enable = false
#canal.instance.detecting.sql = insert into retl.xdual values(1,now()) on duplicate key update x=now()
canal.instance.detecting.sql = select 1
canal.instance.detecting.interval.time = 3
canal.instance.detecting.retry.threshold = 3
canal.instance.detecting.heartbeatHaEnable = false
# support maximum transaction size, more than the size of the transaction will be cut into multiple transactions delivery
canal.instance.transaction.size = 1024
# mysql fallback connected to new master should fallback times
canal.instance.fallbackIntervalInSeconds = 60
# network config
canal.instance.network.receiveBufferSize = 16384
canal.instance.network.sendBufferSize = 16384
canal.instance.network.soTimeout = 30
# binlog filter config
canal.instance.filter.druid.ddl = true
canal.instance.filter.query.dcl = false
canal.instance.filter.query.dml = false
canal.instance.filter.query.ddl = false
canal.instance.filter.table.error = false
canal.instance.filter.rows = false
canal.instance.filter.transaction.entry = false
# binlog format/image check
canal.instance.binlog.format = ROW,STATEMENT,MIXED
canal.instance.binlog.image = FULL,MINIMAL,NOBLOB
# binlog ddl isolation
canal.instance.get.ddl.isolation = false
# parallel parser config
canal.instance.parser.parallel = true
## concurrent thread number, default 60% available processors, suggest not to exceed Runtime.getRuntime().availableProcessors()
#canal.instance.parser.parallelThreadSize = 16
## disruptor ringbuffer size, must be power of 2
canal.instance.parser.parallelBufferSize = 256
# table meta tsdb info
canal.instance.tsdb.enable = true
canal.instance.tsdb.dir = ${canal.file.data.dir:../conf}/${canal.instance.destination:}
canal.instance.tsdb.url = jdbc:h2:${canal.instance.tsdb.dir}/h2;CACHE_SIZE=1000;MODE=MYSQL;
canal.instance.tsdb.dbUsername = canal
canal.instance.tsdb.dbPassword = canal
# dump snapshot interval, default 24 hour
canal.instance.tsdb.snapshot.interval = 24
# purge snapshot expire , default 360 hour(15 days)
canal.instance.tsdb.snapshot.expire = 360
# aliyun ak/sk , support rds/mq
canal.aliyun.accessKey =
canal.aliyun.secretKey =
#################################################
######### destinations #############
#################################################
canal.destinations = example
# conf root dir
canal.conf.dir = ../conf
# auto scan instance dir add/remove and start/stop instance
canal.auto.scan = true
canal.auto.scan.interval = 5
canal.instance.tsdb.spring.xml = classpath:spring/tsdb/h2-tsdb.xml
#canal.instance.tsdb.spring.xml = classpath:spring/tsdb/mysql-tsdb.xml
canal.instance.global.mode = manager
canal.instance.global.lazy = false
canal.instance.global.manager.address = ${canal.admin.manager}
#canal.instance.global.spring.xml = classpath:spring/memory-instance.xml
canal.instance.global.spring.xml = classpath:spring/file-instance.xml
#canal.instance.global.spring.xml = classpath:spring/default-instance.xml
##################################################
######### MQ #############
##################################################
#这里因为我们选择的是kafka，所以填kafka集群地址，多个地址用逗号隔开
canal.mq.servers = 47.98.183.103:9092
canal.mq.retries = 0
canal.mq.batchSize = 16384
canal.mq.maxRequestSize = 1048576
canal.mq.lingerMs = 100
canal.mq.bufferMemory = 33554432
canal.mq.canalBatchSize = 50
canal.mq.canalGetTimeout = 100
canal.mq.flatMessage = true
canal.mq.compressionType = none
canal.mq.acks = all
#canal.mq.properties. =
canal.mq.producerGroup = test
# Set this value to "cloud", if you want open message trace feature in aliyun.
canal.mq.accessChannel = local
# aliyun mq namespace
#canal.mq.namespace =
##################################################
######### Kafka Kerberos Info #############
##################################################
canal.mq.kafka.kerberos.enable = false
canal.mq.kafka.kerberos.krb5FilePath = "../conf/kerberos/krb5.conf"
canal.mq.kafka.kerberos.jaasFilePath = "../conf/kerberos/jaas.conf"
需要注意的是：
①可以注意到，配置中admin的密码是一串长长的字符串，这是加密后的密码。我们用docker安装启动canal-admin后，就会在本地创建一个数据库，ip是容器所在服务器本地ip，端口3306，这里我的是192.168.45.129:3306，账号密码是canal。我们可以用navicat连接。
然后可以通过以下语句获取到加密密码
②可以注意到，我们配置了canal.destinations = example
这个example是我们接下来要创建的instance名称。
需要注意的是，经测试，后续再创建多几个instance，不需要在此修改canal.destinations配置。
ok，接下来我们开始创建instance。
接着同样是点击载入模板，然后进行配置
最终配置内容如下：
#################################################
## mysql serverId , v1.0.26+ will autoGen
# canal.instance.mysql.slaveId=0
# enable gtid use true/false
canal.instance.gtidon=false
# position info mysql地址
canal.instance.master.address=47.98.183.103:3306
canal.instance.master.journal.name=
canal.instance.master.position=
canal.instance.master.timestamp=
canal.instance.master.gtid=
# rds oss binlog
canal.instance.rds.accesskey=
canal.instance.rds.secretkey=
canal.instance.rds.instanceId=
# table meta tsdb info
canal.instance.tsdb.enable=true
#canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb
#canal.instance.tsdb.dbUsername=canal
#canal.instance.tsdb.dbPassword=canal
#canal.instance.standby.address =
#canal.instance.standby.journal.name =
#canal.instance.standby.position =
#canal.instance.standby.timestamp =
#canal.instance.standby.gtid=
# username/password mysql账号密码
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal
canal.instance.connectionCharset = UTF-8
# enable druid Decrypt database password
canal.instance.enableDruid=false
#canal.instance.pwdPublicKey=MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBALK4BUxdDltRRE5/zXpVEVPUgunvscYFtEip3pmLlhrWpacX7y7GCMo2/JM6LeHmiiNdH1FWgGCpUfircSwlWKUCAwEAAQ==
# table regex mysql表设置，这里我们设置读取mytest数据库的user表
canal.instance.filter.regex=mytest\\.user
# table black regex
canal.instance.filter.black.regex=
# table field filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2)
#canal.instance.filter.field=test1.t_product:id/subject/keywords,test2.t_company:id/name/contact/ch
# table field black filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2)
#canal.instance.filter.black.field=test1.t_product:subject/product_image,test2.t_company:id/name/contact/ch
# mq config 主题名称
canal.mq.topic=user_topic
# dynamic topic route by schema or table regex
#canal.mq.dynamicTopic=mytest1.user,mytest2\\..*,.*\\..*
canal.mq.partition=0
# hash partition config
#canal.mq.partitionsNum=3
#canal.mq.partitionHash=test.table:id^name,.*\\..*
#################################################
创建成功后，instance是停止状态，我们不用管它，等有canal-server连接我们的online集群时，它就会变成启动状态。
四、canal-server
1、拉取镜像
docker pull canal/canal-server:v1.1.4
2、运行
出于跟canal-admin相似的原因，这里我创建执行脚本文件run.sh，内容如下
#!/bin/bash
function usage() {
echo "Usage:"
echo " run.sh [CONFIG]"
echo "example 1 :"
echo " run.sh -e canal.instance.master.address=127.0.0.1:3306 \\"
echo " -e canal.instance.dbUsername=canal \\"
echo " -e canal.instance.dbPassword=canal \\"
echo " -e canal.instance.connectionCharset=UTF-8 \\"
echo " -e canal.instance.tsdb.enable=true \\"
echo " -e canal.instance.gtidon=false \\"
echo " -e canal.instance.filter.regex=.*\\\\\\..* "
echo "example 2 :"
echo " run.sh -e canal.admin.manager=127.0.0.1:8089 \\"
echo " -e canal.admin.port=11110 \\"
echo " -e canal.admin.user=admin \\"
echo " -e canal.admin.passwd=4ACFE3202A5FF5CF467898FC58AAB1D615029441"
exit
}
function check_port() {
local port=$1
local TL=$(which telnet)
if [ -f $TL ]; then
data=`echo quit | telnet 127.0.0.1 $port| grep -ic connected`
echo $data
return
fi
local NC=$(which nc)
if [ -f $NC ]; then
data=`nc -z -w 1 127.0.0.1 $port | grep -ic succeeded`
echo $data
return
fi
echo "0"
return
}
function getMyIp() {
case "`uname`" in
Darwin)
myip=`echo "show State:/Network/Global/IPv4" | scutil | grep PrimaryInterface | awk '{print $3}' | xargs ifconfig | grep inet | grep -v inet6 | awk '{print $2}'`
;;
*)
myip=`ip route get 1 | awk '{print $NF;exit}'`
;;
esac
echo $myip
}
CONFIG=${@:1}
#VOLUMNS="-v $DATA:/home/admin/canal-server/logs"
PORTLIST="11110 11111 11112 9100"
PORTS=""
for PORT in $PORTLIST ; do
#exist=`check_port $PORT`
exist="0"
if [ "$exist" == "0" ]; then
PORTS="$PORTS -p $PORT:$PORT"
else
echo "port $PORT is used , pls check"
exit 1
fi
done
NET_MODE=""
case "`uname`" in
Darwin)
bin_abs_path=`cd $(dirname $0); pwd`
;;
Linux)
bin_abs_path=$(readlink -f $(dirname $0))
NET_MODE="--net=host"
PORTS=""
;;
*)
bin_abs_path=`cd $(dirname $0); pwd`
NET_MODE="--net=host"
PORTS=""
;;
esac
BASE=${bin_abs_path}
DATA="$BASE/data"
mkdir -p $DATA
if [ $# -eq 0 ]; then
usage
elif [ "$1" == "-h" ] ; then
usage
elif [ "$1" == "help" ] ; then
usage
fi
MEMORY="-m 4096m"
LOCALHOST=`getMyIp`
cmd="docker run -d -it -h $LOCALHOST $CONFIG --name=canal-server $VOLUMNS $NET_MODE $PORTS $MEMORY canal/canal-server:v1.1.4"
echo $cmd
eval $cmd执行语句如下：
sh run.sh -e canal.admin.manager=192.168.45.129:8089 \
-e canal.admin.port=11110 \
-e canal.admin.user=admin \
-e canal.admin.passwd=4ACFE3202A5FF5CF467898FC58AAB1D615029441 \
-e canal.admin.register.cluster=online
查看容器和容器日志，可以发现启动成功了
如果没有显示SUCCESSFUL，可以进入canal-server的容器去查看日志，我这边遇到过一个错误，日志如下：
com.alibaba.otter.canal.common.CanalException: load manager config failed.
Caused by: com.alibaba.otter.canal.common.CanalException: requestGet for canal config error: auth :admin is failed
2020-05-19 15:41:01.542 [main] INFO com.alibaba.otter.canal.deployer.CanalLauncher - ## set default uncaught exception handler
2020-05-19 15:41:01.647 [main] INFO com.alibaba.otter.canal.deployer.CanalLauncher - ## load canal configurations
2020-05-19 15:41:04.095 [main] ERROR com.alibaba.otter.canal.deployer.CanalLauncher - ## Something goes wrong when starting up the canal Server:
意思是admin账号密码错误了，原因是密码我用的是123456加密字符串，而不是admin加密字符串。
3、测试
ok，这时候我们去canal-admin界面查看
可以看到我们的server连接到我们的admin了，且instance也从停止状态变成启动状态。
我们点击server的操作->日志，内容如下：
可以看出，server启动成功，且通知example启动成功。
再去instance列表点击example的操作->日志，内容如下：
可以看出，成功的连接到了我们的数据库，并且开始读取binlog日志。
ok，接下来我们就可以测试一下改动我们的user表数据，看会不会读取到。
首先我们先进入kafka容器运行消费者来消费user_topic消息，
之后我们往user表插入一条数据，再次查看控制台，内容如下：
可以看到，已经被读取到了。(记得按Ctrl+C键退出)
如此，我们的服务端就大工告成。接下来就是客户端了，我们可以根据需要将数据同步适配到elasticsearch、redis、mysql等端，这些都是在客户端设置的。这里我们选择适配到elasticsearch端。
五、canal-adapter
由于目前canal-adapter没有官方docker镜像，所以我们自己构建一个镜像。大家也可以直接拉取我的镜像
docker pull liazhan/canal-adapter:v1.1.5
1、首先下载tar包，这里我们下载的是1.5.1版本的，推荐使用迅雷下载，还挺快。
在linux服务器上创建一个文件夹adapter，然后将tar包上传到adapter文件夹。
2、接着在adapter文件夹中创建Dockerfile文件，内容如下：
FROM openjdk:8-jre-alpine
RUN echo "Asia/Shanghai" > /etc/timezone
RUN mkdir -p /opt/canal/adapter
ADD canal.adapter-1.1.5-SNAPSHOT.tar.gz /opt/canal/adapter
WORKDIR /opt/canal/adapter
CMD ["sh", "-c", "sh /opt/canal/adapter/bin/startup.sh && tail -F logs/adapter/adapter.log"]
3、然后在adapter目录下运行指令构建镜像
docker build -t liazhan/canal-adapter:v1.1.5 .
4、构建完成后，输入以下指令运行
docker run -d --name adapter liazhan/canal-adapter:v1.1.5
如此便完成了，但是需要注意的是，这时候查看容器日志，会出现很多报错信息。这是因为我们还没有进行相关的配置。
这样子构建镜像还有个缺点是当我们需要关闭容器时，需要先在容器中的bin目录运行stop.sh停掉canal-adapter，否则我们停掉容器后再次启动容器会报错found adapter.pid , Please run stop.sh first ,then startup.sh
5、进入容器修改配置
docker exec -it 08e641f610bb sh
平常的容器一般都可以执行/bin/bash，很是alpine没有，这里用的是sh
然后我们进入conf目录去修改配置文件
这里我们先修改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: kafka # kafka rocketMQ
canalServerHost: 192.168.45.129:11111
zookeeperHosts: 47.98.183.103:2181
mqServers: 47.98.183.103:9092 #or rocketmq
# flatMessage: true
batchSize: 500
syncBatchSize: 1000
retries: 0
timeout:
accessKey:
secretKey:
username:
password:
vhost:
srcDataSources:
defaultDS:
url: jdbc:mysql://47.98.183.103:3306/mytest?useUnicode=true
username: canal
password: canal
canalAdapters:
- instance: user_topic # 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
# - 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: 47.98.183.103: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
由于我们用的是kafka，所以instance配置的是kafka的topic，即user_topic
es的cluster.name可以访问es的ip:9200查看
conf文件夹下边的es6和es7文件夹表示的是es版本，我们用的是es7版本，文件夹下面放的是相关配置文件
里面有三个配置文件，我们修改mytest_user.yml的配置文件内容如下：
dataSourceKey: defaultDS
destination: user_topic
groupId: g1
esMapping:
_index: mytest_user
_id: _id
# upsert: true
# pk: id
sql: "select a.id as _id, a.name, a.role_id, b.role_name,
a.c_time from user a
left join role b on b.id=a.role_id"
# objFields:
# _labels: array:;
etlCondition: "where a.c_time>={}"
commitBatch: 3000
destination配置的是kafka的topic；_index是es的索引，我们还没创建，等下我们再创建；sql中查询的列名对应es索引的字段。
如果有多个instance，也意味着有多个topic，那么我们需要创建多几个配置文件来一一对应，比如mytest_user.yml对应user_topic。
ok，修改完配置文件我们到bin目录执行重新启动canal-adapter
sh ./restart.sh
然后到logs目录下查看日志文件最后部分
可以看到已经启动成功了。
接着我们使用postman给es添加索引
然后我们往数据库user表插入一条记录，然后再次查看日志
可以看到已经显示同步到索引mytest_user了，我们用postman查看一下
同步成功啦，其中id为4的数据是我之前测试添加的。而且我们可以看到数据中有个role_name是java，是因为在role表中我添加了一条数据
同步的时候sql语句通过left join获取到的。
ok，终于完成了。撒花~
权威｜前沿｜技术｜干货｜国内首个API全生命周期开发者社区
更多推荐
6
0- 0
已为社区贡献1条内容
所有评论(0)