目录

一、简介

二、准备

三、canal-admin

四、canal-server

五、canal-adapter


 

一、简介

操作系统: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、测试和配置

访问http://ip地址:8089/

账号admin,初始密码123456(登陆密码并不是admin,连接密码才是admin)

 

 

canal-admin的核心模型主要有:

  1. instance,对应canal-server里的instance,一个最小的订阅mysql的队列
  2. server,对应canal-server,一个server里可以包含多个instance
  3. 集群,对应一组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版本的,推荐使用迅雷下载,还挺快。

https://github.com/alibaba/canal/releases/download/canal-1.1.5-alpha-1/canal.adapter-1.1.5-SNAPSHOT.tar.gz

在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,终于完成了。撒花~

Logo

权威|前沿|技术|干货|国内首个API全生命周期开发者社区

更多推荐