我说下我的版本,首先要安装
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

image.png

创建一个给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

image.png

下载完成后,上传到服务器解压,分别为解压文件位置

image.png

第四步: 配置cancel 关键步骤!!!!
1 解压canal.deployer-1.1.5 到 /opt/canal-server 后目录为

image.png

修改配置文件conf/example/instance.properties，按如下配置即可，主要是修改数据库相关配置；

image.png

由于我都是安装在同一台机器上,所以为127.0.0.1:3306
进入 当前的bin文件下 运行命令,启动canal服务

sh startup.sh

2 解压 canal.adapter-1.1.5 到 /opt/canal-adpter 目录为

image.png

修改配置文件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

下面是配置说明

image.png

image.png

配置重点 一个是 jdbc:mysql://127.0.0.1:3306/canal-test?useUnicode=true 中的 数据库名称
第二个是es集群名称根据自己的实际的配置,我的是 docker-cluster
第三个 - name: es7 这个很重要一会儿要用

image.png

其他全部照搬即可
添加配置文件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,

image.png

执行 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"
            }
        }
    }
}

访问截图

image.png

与数据库字段一一对应,所以名称与之前配置的 canal_product 必须一致
至此全部配置完毕
接下来,看着canal-adapter 的日志

image.png

在mysql数据库中反复执行 删除数据,添加数据操作
可以看到 canal-adapter 中已经接受到了mysql 的dml操作语句!!!!!!!

image.png

接下来,通过命令触发,让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

image.png

可以看到es中已经同步到了mysql数据;
哇哇哇~~~~~~~~~~~~~~~~~~~~~~~~~~~至此,大功告成!!!!!!!!!!!!!!!!!!!!!!!!!

说实话这次实践遇到了很多问题
1 docker 中vim是没有的要安装
2 es-header 和es链接是要配置跨域的
3 canal 的 数据同步不是自动的,需要触发!!!!!!!!!!!!!
3 canal 的配置很复杂~~一个不小心就会导致同步数据的时候错误,要很小心
4 当看到canal-adapter的日志中有mysql的dml日志的时候,说明已经离成功很近了
5 当触发同步的时候报错, 找不到任务,仔细检查触发中与配置相关的每一个,一定是哪里错了