【Canal从入门到放弃】(三)Canal 推送 ElasticSearch | Eddie'Blog
【Canal从入门到放弃】(三)Canal 推送 ElasticSearch

【Canal从入门到放弃】(三)Canal 推送 ElasticSearch

eddie 427 2022-07-12

目录

介绍 Canal 把 MySQL 数据同步到 ElasticSearch 的关键细节

版本说明

  • canal-1.1.6
  • MySQL:5.7
  • ElasticSearch:7.10.1
  • kibana:7.10.1

参考文档

准备工作

需要的程序文件, 下载:Canal-1.1.6 xxxx

[root@localhost module]# pwd
/opt/module

[root@localhost module]# ll
total 0
drwxr-xr-x 7 root root 66 Jul  6 14:48 canal-adapter
drwxr-xr-x 6 root root 52 Jul  6 15:53 canal-admin
drwxr-xr-x 7 root root 66 Jul  6 14:49 canal-server   ## canal-deployer

Canal-Server

文件与配置讲解

*.bak 是备份文件,可以忽略。主要是新版本1.1.6 出现 canal_local.properties ,默认就好

[root@localhost conf]# pwd
/opt/module/canal-server/conf

[root@localhost conf]# ll
total 24
-rwxrwxrwx 1 root root  319 Jun 22  2021 canal_local.properties
-rwxr-xr-x 1 root root  319 Jul  6 15:30 canal_local.properties.bak
-rwxr-xr-x 1 root root 6509 Jul  6 15:29 canal.properties.bak
drwxr-xr-x 2 root root   38 Jul  7 14:07 es-instance
drwxrwxrwx 2 root root   33 Jul 11 14:00 example
-rwxrwxrwx 1 root root 4429 May 20 15:26 logback.xml
drwxrwxrwx 2 root root   39 Jul  6 14:49 metrics
drwxrwxrwx 3 root root  149 Jul  6 14:49 spring

启动服务

[root@localhost canal-server]# pwd
/opt/module/canal-server

[root@localhost canal-server]# bin/startup.sh 

## 还包含重启、停止
[root@localhost canal-server]# ll bin
total 20
-rw-r--r-- 1 root root    5 Jul 11 16:56 canal.pid
-rwxr-xr-x 1 root root  226 Jun 22  2021 restart.sh
-rwxr-xr-x 1 root root 1244 Oct  9  2021 startup.bat
-rwxr-xr-x 1 root root 3765 May 23 17:36 startup.sh
-rwxr-xr-x 1 root root 1356 Jun 22  2021 stop.sh

Canal-Admin

[root@localhost canal-admin]# cd /opt/module/canal-admin/conf/ 

[root@localhost conf]# ll
total 24
-rwxrwxrwx 1 root root  495 Jul  6 16:20 application.yml              ## 需要修改DB信息
-rwxrwxrwx 1 root root 3898 Jun 22  2021 canal_manager.sql            ## 需要复制里面语句
-rwxrwxrwx 1 root root 6271 Jun 22  2021 canal-template.properties
-rwxrwxrwx 1 root root 2036 Jun 22  2021 instance-template.properties
-rwxrwxrwx 1 root root 1568 Jun 22  2021 logback.xml
drwxrwxrwx 3 root root   72 Jul  6 15:53 public

application.yml 配置

访问db账户,需要有读写权限

server:
  port: 8089
spring:
  jackson:
    date-format: yyyy-MM-dd HH:mm:ss
    time-zone: GMT+8

spring.datasource:
  address: 127.0.0.1:3306
  database: tmp_canal_manager
  username: *****
  password: *****
  driver-class-name: com.mysql.jdbc.Driver
  url: jdbc:mysql://${spring.datasource.address}/${spring.datasource.database}?useUnicode=true&characterEncoding=UTF-8&useSSL=false
  hikari:
    maximum-pool-size: 30
    minimum-idle: 1

canal:
  adminUser: admin
  adminPasswd: admin

canal_manager.sql 语句

直接运行DDL语句

CREATE DATABASE /*!32312 IF NOT EXISTS*/ `tmp_canal_manager` /*!40100 DEFAULT CHARACTER SET utf8 COLLATE utf8_bin */;

USE `tmp_canal_manager`;

SET NAMES utf8;
SET FOREIGN_KEY_CHECKS = 0;

-- ----------------------------
-- Table structure for canal_adapter_config
-- ----------------------------
DROP TABLE IF EXISTS `canal_adapter_config`;
CREATE TABLE `canal_adapter_config` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  `category` varchar(45) NOT NULL,
  `name` varchar(45) NOT NULL,
  `status` varchar(45) DEFAULT NULL,
  `content` text NOT NULL,
  `modified_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

-- ----------------------------
-- Table structure for canal_cluster
-- ----------------------------
DROP TABLE IF EXISTS `canal_cluster`;
CREATE TABLE `canal_cluster` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  `name` varchar(63) NOT NULL,
  `zk_hosts` varchar(255) NOT NULL,
  `modified_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

-- ----------------------------
-- Table structure for canal_config
-- ----------------------------
DROP TABLE IF EXISTS `canal_config`;
CREATE TABLE `canal_config` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  `cluster_id` bigint(20) DEFAULT NULL,
  `server_id` bigint(20) DEFAULT NULL,
  `name` varchar(45) NOT NULL,
  `status` varchar(45) DEFAULT NULL,
  `content` text NOT NULL,
  `content_md5` varchar(128) NOT NULL,
  `modified_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  UNIQUE KEY `sid_UNIQUE` (`server_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

-- ----------------------------
-- Table structure for canal_instance_config
-- ----------------------------
DROP TABLE IF EXISTS `canal_instance_config`;
CREATE TABLE `canal_instance_config` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  `cluster_id` bigint(20) DEFAULT NULL,
  `server_id` bigint(20) DEFAULT NULL,
  `name` varchar(45) NOT NULL,
  `status` varchar(45) DEFAULT NULL,
  `content` text NOT NULL,
  `content_md5` varchar(128) DEFAULT NULL,
  `modified_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  UNIQUE KEY `name_UNIQUE` (`name`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

-- ----------------------------
-- Table structure for canal_node_server
-- ----------------------------
DROP TABLE IF EXISTS `canal_node_server`;
CREATE TABLE `canal_node_server` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  `cluster_id` bigint(20) DEFAULT NULL,
  `name` varchar(63) NOT NULL,
  `ip` varchar(63) NOT NULL,
  `admin_port` int(11) DEFAULT NULL,
  `tcp_port` int(11) DEFAULT NULL,
  `metric_port` int(11) DEFAULT NULL,
  `status` varchar(45) DEFAULT NULL,
  `modified_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

-- ----------------------------
-- Table structure for canal_user
-- ----------------------------
DROP TABLE IF EXISTS `canal_user`;
CREATE TABLE `canal_user` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  `username` varchar(31) NOT NULL,
  `password` varchar(128) NOT NULL,
  `name` varchar(31) NOT NULL,
  `roles` varchar(31) NOT NULL,
  `introduction` varchar(255) DEFAULT NULL,
  `avatar` varchar(255) DEFAULT NULL,
  `creation_date` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

SET FOREIGN_KEY_CHECKS = 1;

-- ----------------------------
-- Records of canal_user
-- ----------------------------
BEGIN;
INSERT INTO `canal_user` VALUES (1, 'admin', '6BB4837EB74329105EE4568DDA7DC67ED2CA2AD9', 'Canal Manager', 'admin', NULL, NULL, '2019-07-14 00:05:28');
COMMIT;

SET FOREIGN_KEY_CHECKS = 1;

启动服务

启动后,就可以访问:http://127.0.0.1:8089 admin/123456

[root@localhost canal-admin]# pwd
/opt/module/canal-admin

[root@localhost canal-admin]# bin/startup.sh 

## 其余操作脚本
[root@localhost canal-admin]# ll bin
total 20
-rw-r--r-- 1 root root    5 Jul 11 10:36 admin.pid
-rwxr-xr-x 1 root root  205 Jun 22  2021 restart.sh
-rwxr-xr-x 1 root root  747 Jun 22  2021 startup.bat
-rwxr-xr-x 1 root root 2866 May 23 17:39 startup.sh
-rwxr-xr-x 1 root root 1362 Jun 22  2021 stop.sh

Web 操作

主页 / Canal Server / Server 管理

canal-admin01选择配置,通常第一次 canal.properties 是载入模板, 或者复制我提供的:

#################################################
######### 		common argument		#############
#################################################
# tcp bind ip
canal.ip =
# register ip to zookeeper
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 = 127.0.0.1:8089
canal.admin.port = 11110
canal.admin.user = admin
canal.admin.passwd = 4ACFE3202A5FF5CF467898FC58AAB1D615029441
# admin auto register
#canal.admin.register.auto = true
#canal.admin.register.cluster =
#canal.admin.register.name =

canal.zkServers =
# flush data to zk
canal.zookeeper.flush.period = 1000
canal.withoutNetty = false
# tcp, kafka, rocketMQ, rabbitMQ
canal.serverMode = tcp
# 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
canal.instance.filter.dml.insert = false
canal.instance.filter.dml.update = false
canal.instance.filter.dml.delete = 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

#################################################
######### 		destinations		#############
#################################################
canal.destinations = 
# 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
# set this value to 'true' means that when binlog pos not found, skip to latest.
# WARN: pls keep 'false' in production env, or if you know what you want.
canal.auto.reset.latest.pos.mode = false

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 Properties      #############
##################################################
# aliyun ak/sk , support rds/mq
canal.aliyun.accessKey =
canal.aliyun.secretKey =
canal.aliyun.uid=

canal.mq.flatMessage = true
canal.mq.canalBatchSize = 50
canal.mq.canalGetTimeout = 100
# Set this value to "cloud", if you want open message trace feature in aliyun.
canal.mq.accessChannel = local

canal.mq.database.hash = true
canal.mq.send.thread.size = 30
canal.mq.build.thread.size = 8

##################################################
######### 		     Kafka 		     #############
##################################################
kafka.bootstrap.servers = 127.0.0.1:9092
kafka.acks = all
kafka.compression.type = none
kafka.batch.size = 16384
kafka.linger.ms = 1
kafka.max.request.size = 1048576
kafka.buffer.memory = 33554432
kafka.max.in.flight.requests.per.connection = 1
kafka.retries = 0

kafka.kerberos.enable = false
kafka.kerberos.krb5.file = "../conf/kerberos/krb5.conf"
kafka.kerberos.jaas.file = "../conf/kerberos/jaas.conf"

##################################################
######### 		    RocketMQ	     #############
##################################################
rocketmq.producer.group = test
rocketmq.enable.message.trace = false
rocketmq.customized.trace.topic =
rocketmq.namespace =
rocketmq.namesrv.addr = 127.0.0.1:9876
rocketmq.retry.times.when.send.failed = 0
rocketmq.vip.channel.enabled = false
rocketmq.tag = 

##################################################
######### 		    RabbitMQ	     #############
##################################################
rabbitmq.host =
rabbitmq.virtual.host =
rabbitmq.exchange =
rabbitmq.username =
rabbitmq.password =
rabbitmq.deliveryMode =

主页 /Canal Server / Instance 管理 点击【新建 Instance】,第一次 es-instance/instance.propertios 可以载入配置

#################################################
## mysql serverId , v1.0.26+ will autoGen
canal.instance.mysql.slaveId=20                  ## 必需改,必需不能与MySQL service-id 一样

# enable gtid use true/false
canal.instance.gtidon=false

# position info
canal.instance.master.address=127.0.0.1:3306       ## 必需的,修改成MySQL DB连接
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 root的账户
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
canal.instance.filter.regex=.*\\..*    ## 默认就好
#canal.instance.filter.regex=.\..
# 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=example_es   ## 使用MQ就填写, 现在ES就不用了
# 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,.*\\..*
#################################################

这两个服务都可以在页面上启动与停止

Canal-Adapter

结构

[root@localhost canal-adapter]# pwd
/opt/module/canal-adapter

[root@localhost canal-adapter]# ll
total 12
drwxr-xr-x 2 root root   95 Jul 12 09:59 bin
drwxrwxrwx 9 root root  161 Jul 11 16:53 conf
drwxr-xr-x 2 root root 4096 Jul  6 14:49 lib
drwxrwxrwx 3 root root   21 Jul  7 13:41 logs
drwxrwxrwx 2 root root 4096 Jun 13 09:45 plugin

[root@localhost canal-adapter]# ll bin/
total 20
-rw-r--r-- 1 root root    6 Jul 12 09:59 adapter.pid
-rwxr-xr-x 1 root root  205 Jun 22  2021 restart.sh
-rwxr-xr-x 1 root root  793 Jun 22  2021 startup.bat
-rwxr-xr-x 1 root root 2887 May 23 17:40 startup.sh
-rwxr-xr-x 1 root root 1370 Jun 22  2021 stop.sh

[root@localhost canal-adapter]# ll conf/
total 12
-rwxrwxrwx 1 root root 3240 Jul 11 16:44 application.yml
-rwxrwxrwx 1 root root  283 Jul 11 10:40 bootstrap.yml        ## DB连接信息
drwxr-xr-x 2 root root   70 Jul  6 14:49 es6
drwxr-xr-x 2 root root   70 Jul 12 09:59 es7
drwxr-xr-x 2 root root   32 Jul  6 14:49 hbase
drwxr-xr-x 2 root root   31 Jul  6 14:49 kudu
-rwxrwxrwx 1 root root 3106 May 20 15:26 logback.xml
drwxrwxrwx 2 root root   30 Jul  6 14:49 META-INF
drwxrwxrwx 2 root root   29 Jun 22  2021 rdb
drwxrwxrwx 2 root root   22 Oct  9  2021 tablestore

bootstrap.yml

[root@localhost canal-adapter]# vim conf/bootstrap.yml 

canal:
  manager:
    jdbc:
    ## Canal-Admin 使用的DB库信息
      url: jdbc:mysql://127.0.0.1:3306/tmp_canal_manager?useUnicode=true&characterEncoding=UTF-8
      username: canal
      password: canal

application.yml

[root@localhost canal-adapter]# cat 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: tcp #tcp kafka rocketMQ rabbitMQ
  flatMessage: true
  zookeeperHosts:
  syncBatchSize: 1000
  retries: -1
  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/test?useUnicode=true    ## 需要查询的库
      username: canal
      password: canal
  canalAdapters:
  ## Canal-Admin的instance名称
  - instance: es-instance # 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
#          druid.stat.enable: false
#          druid.stat.slowSqlMillis: 1000
#      - 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  ## 对应这 ./canal-adapter/conf/es7 这个文件夹名称
        hosts: http://127.0.0.1:9200 # for rest mode / 127.0.0.1:9300
        properties:
          mode: rest # or rest=9200  transport=9300
          # security.auth: test:123456 #  only used for rest mode
          cluster.name: my-application  # curl 127.0.0.1:9200 显示的 cluster_name
#      - name: kudu
#        key: kudu
#        properties:
#          kudu.master.address: 127.0.0.1 # ',' split multi address
#      - name: phoenix
#        key: phoenix
#        properties:
#          jdbc.driverClassName: org.apache.phoenix.jdbc.PhoenixDriver
#          jdbc.url: jdbc:phoenix:127.0.0.1:2181:/hbase/db
#          jdbc.username:
#          jdbc.password:

创建关于ES索引的yaml文件

[root@localhost ~]# cd /opt/module/canal-adapter/conf/es7/
[root@localhost es7]# vim mytest_user.yml   ## 文件名和之后创建ES索引对应起来

mytest_user.yml

dataSourceKey: defaultDS ## 源数据源的key, 对应上面配置的srcDataSources中的值
destination: es-instance ## cannal的instance或者MQ的topic
groupId: g1              ## 对应MQ模式下的groupId, 只会同步对应groupId的数据_search
esMapping:
  _index: mytest_user    ## es 的索引名称
  _id: _id               ## es 的_id, 如果不配置该项必须配置下面的pk项_id则会由es自动分配
#  upsert: true
#  字段注意大小段写,不然ES会出现null情况
  sql: "SELECT
        a.id AS _id,
        a.name AS username,
        a.sex AS sex,
        a.create_time AS createTime,
        a.update_time AS updateTime 
FROM
        user_info a"
  etlCondition: "where a.id>={}"
  commitBatch: 3000
~                   

指定同步数据的条件

curl http://127.0.0.1:9200/etl/es7/mytest_user.yml -X POST -d "id=1"

etl 固定的
es7 之前起的名字必须对应,后续配置文件也在这里面
product.yml 配置文件名称
-d “params=1” 同步数据的条件 1 入参

启动服务

[root@localhost canal-admin]# pwd
/opt/module/canal-admin

[root@localhost canal-admin]# ll bin/
total 20
-rw-r--r-- 1 root root    5 Jul 11 10:36 admin.pid
-rwxr-xr-x 1 root root  205 Jun 22  2021 restart.sh
-rwxr-xr-x 1 root root  747 Jun 22  2021 startup.bat
-rwxr-xr-x 1 root root 2866 May 23 17:39 startup.sh
-rwxr-xr-x 1 root root 1362 Jun 22  2021 stop.sh

ElasticSearch 容器化部署

docker-compose.yml

docker-compose up -d

version: '2.2'

services:
  es01:
    image: docker.elastic.co/elasticsearch/elasticsearch:7.10.1
    container_name: es01
    environment:
      # 节点名称
      - node.name=es01
      # 集群名称
      - cluster.name=my-application
      #指定主机名称
      - discovery.seed_hosts=es02,es03
      # 从哪里选举主节点
      - cluster.initial_master_nodes=es01,es02,es03
      # 是否锁住内存,避免交换(swapped)带来的性能损失
      - bootstrap.memory_lock=true
      - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
    ulimits:
      memlock:
        soft: -1
        hard: -1
    volumes:
      - ./es01/data:/usr/share/elasticsearch/data
    ports:
      - 9200:9200
      - 9300:9300
    networks:
      - elastic
 
  es02:
    image: docker.elastic.co/elasticsearch/elasticsearch:7.10.1
    container_name: es02
    environment:
      - node.name=es02
      - cluster.name=my-application
      - discovery.seed_hosts=es01,es03
      - cluster.initial_master_nodes=es01,es02,es03
      - bootstrap.memory_lock=true
      - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
    ulimits:
      memlock:
        soft: -1
        hard: -1
    volumes:
      - ./es02/data:/usr/share/elasticsearch/data
    networks:
      - elastic
 
  es03:
    image: docker.elastic.co/elasticsearch/elasticsearch:7.10.1
    container_name: es03
    environment:
      - node.name=es03
      - cluster.name=my-application
      - discovery.seed_hosts=es01,es02
      - cluster.initial_master_nodes=es01,es02,es03
      - bootstrap.memory_lock=true
      - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
    ulimits:
      memlock:
        soft: -1
        hard: -1
    volumes:
      - ./es03/data:/usr/share/elasticsearch/data
    networks:
      - elastic
 
  kib01:
    image: docker.elastic.co/kibana/kibana:7.10.1
    container_name: kib01
    ports:
      - 5601:5601
    environment:
      ELASTICSEARCH_URL: http://es01:9200
      ELASTICSEARCH_HOSTS: '["http://es01:9200","http://es02:9200","http://es03:9200"]'
    networks:
      - elastic
 
  elastichd:
    image: containerize/elastichd:latest
    container_name: elastichd
    networks:
    - elastic
    ports:
      - 9800:9800
    depends_on:
      - es01
      - es02
      - es03

volumes:
  data01:
    driver: local
  data02:
    driver: local
  data03:
    driver: local
 
networks:
  elastic:
    driver: bridge

MySQL DDL 创建表

CREATE TABLE `user_info` (
  `id` bigint(255) NOT NULL AUTO_INCREMENT,
  `name` varchar(255) DEFAULT NULL,
  `sex` varchar(255) DEFAULT NULL,
  `create_time` datetime DEFAULT CURRENT_TIMESTAMP,
  `update_time` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=4 DEFAULT CHARSET=utf8mb4;

创建ES索引 (使用PostMan)

PUT http://127.0.0.1:9200/mytest_user

{
    "mappings": {
        "properties": {
            "username": {
                "type": "text"
            },
            "sex": {
                "type": "text"
            },
            "createTime": {
                "type": "date"
            },
            "updateTime": {
                "type": "date"
            }
        }
    }
}

对应 mytest_user.yml

Elastichd 界面

ip:9800
在这里插入图片描述

Kibana 界面与测试

MySQL

INSERT INTO `test`.`user_info` ( `name`, `sex` )
VALUES
	( '张三', 'M' );
INSERT INTO `test`.`user_info` ( `name`, `sex` )
VALUES
	( '李四', 'F' );
INSERT INTO `test`.`user_info` ( `name`, `sex` )
VALUES
	( '王五', 'M' );

访问 ip:5601
在这里插入图片描述
GET /mytest_user/_search
在这里插入图片描述

Canal-Adapter 日志打印

正常是出现两条差不多一样数据,但是一条是获取一条是插入es
在这里插入图片描述