本文主要讲解如何从使用canal 将mysql 中的数据同步到elasticsearch中。

本文使用的版本：canal1.1.4+MYSQL5.7.31+ES6.8.0

实现步骤

• （1）安装mysql
  （2）开启mysql binlog row模式，并启动mysql
  （3）安装jdk
  （4）安装Elasticsearch并启动(我安装的是6.8.0,主要目前canal adapter1.1.4还不支持7.0.0的版本)
  （5）安装kibana并启动
  （6）安装并启动canal-server
  （7）安装并启动canal-adapter

1.安装MYSQL

（1）去官网查看最新的安装包
https://dev.mysql.com/downloa…
（2）下载mysql源安装包

wget http://dev.mysql.com/get/mysql57-community-release-el7-11.noarch.rpm

目前版本已经很高了，但是我使用的是57
安装mysql源

yum -y install mysql57-community-release-el7-11.noarch.rpm

查看效果：

yum repolist enabled | grep mysql.*

（3）安装mysql服务器

yum install mysql-community-server

（4）启动mysql服务

systemctl start mysqld.service

（5）查看初始化密码

grep "password" /var/log/mysqld.log

登录：

mysql -u root -p

(6)修改密码
修改密码策略

修改密码长度策略

修改root用户的密码

（6）数据库授权（切记这一步一定要做，我为了方便后面使用的都是root账号，没有说新建一个canal账号）
数据库没有授权，只支持localhost本地访问

GRANT ALL PRIVILEGES ON *.* TO 'root'@'%' IDENTIFIED BY '123456' WITH GRANT OPTION;
FLUSH PRIVILEGES;

2.开启mysql binlog模式

找到my.cnf文件，我本地目录是/etc/my.cnf
添加即可

log-bin=mysql-bin
binlog-format=ROW
server-id=1

然后重启mysql，检查一下binlog是否正确启动

show variables like '%log_bin%';

3、安装jdk

这一步较为简单，如果不会可以百度 or google.

4、安装并启动Elasticsearch

官网地址：https://www.elastic.co/downlo…
执行如下命令，对于安装包也可以手动下载之后上传

cd /usr/local
wget https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-6.8.0-linux-x86_64.tar.gz
tar -xzvf elasticsearch-6.8.0-linux-x86_64.tar.gz
mv elasticsearch-6.8.0-linux-x86_64 elasticsearch
rm -rf elasticsearch-6.8.0-linux-x86_64.tar.gz

命令执行完成之后在/usr/local目录下就会生成一个elasticsearch目录
由于elasticsearch不能使用root账户启动。
下面执行如下命令：

useradd es
chown -R es /usr/local/elasticsearch
su es

使用elasticsearch用户来启动ES
但是这个时候启动会报如下错误，需要修改一些linux 参数

ERROR: [2] bootstrap checks failed
[1]: max file descriptors [4096] for elasticsearch process is too low, increase to at least [65535]
[2]: max virtual memory areas vm.max_map_count [65530] is too low, increase to at least [262144]

vim /etc/security/limits.conf
增加：

* soft nofile 65536
* hard nofile 65536
* soft nproc 2048
* hard nproc 4096
#锁住swapping因此需要在这个配置文件下再增加两行代码
elasticsearch soft memlock unlimited
elasticsearch hard memlock unlimited

vim /etc/sysctl.conf
增加：

vm.max_map_count=655360
fs.file-max=655360

注意：之后需要执行一句命令sysctl -p使系统配置生效(使用root用户)

（2）修改ES配置文件
修改config/elasticsearch.yml

将host的ip换成自己的
network.host: 0.0.0.0

（3）启动elasticsearch

./bin/elasticsearch -d

检查是否启动成功：

curl http://127.0.0.1:9200

5、安装并启动kibana

官网地址：https://www.elastic.co/downlo…
执行如下命令，对于安装包也可以手动下载之后上传

wget https://artifacts.elastic.co/downloads/kibana/kibana-6.8.0-linux-x86_64.tar.gz
tar -xzvf kibana-6.8.0-linux-x86_64.tar.gz
mv kibana-6.8.0-linux-x86_64 kibana
rm -rf kibana-6.8.0-linux-x86_64.tar.gz

修改kibana配置文件
config/kibana.yml

server.host: “0.0.0.0”
启动kibana
nohup ./bin/kibana &
检查是否启动成功
在浏览器中打开http://127.0.0.1:5601

6、安装并启动canal-server

详情请查询官网文档：
https://github.com/alibaba/canal/wiki/QuickStart

（1）下载canal

这里我使用的是1.1.4的版本
（2）解压
下载下来进行解压，形成下面的目录：

（3）修改配置
打开conf/example/instance.properties
slave id 可以填写一个数字，不和master的id重复即可

canal.instance.mysql.slaveId=1234

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

# position info
canal.instance.master.address=127.0.0.1:3306
 # username/password
canal.instance.dbUsername=root
canal.instance.dbPassword=111111

（4）启动canal-server

./bin/startup.sh

7、安装并启动canal-adapter

（1）下载canal-adapter
访问：https://github.com/alibaba/canal/releases ，会列出所有历史的发布版本包 下载方式，比如以1.1.14版本为例子：
wget https://github.com/alibaba/canal/releases/download/canal-1.1.4/canal.adapter-1.1.4.tar.gz
（2）解压缩
使用tar xzvf 进行解压

（3）修改配置

vim 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 # kafka rocketMQ
  canalServerHost: 127.0.0.1:11111
#  zookeeperHosts: slave1:2181
#  mqServers: 127.0.0.1:9092 #or rocketmq
#  flatMessage: true
  batchSize: 500
  syncBatchSize: 1000
  retries: 0
  timeout:
  accessKey:
  secretKey:
  srcDataSources:
    defaultDS:
      url: jdbc:mysql://127.0.0.1:3306/canal_test?useUnicode=true
      username: root
      password: 111111
  canalAdapters:
  - instance: example # 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: 111111
#      - 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: es
        hosts: 127.0.0.1:9200
        properties:
          cluster.name: my-application

这里需要配置的是数据库的schema
jdbc:mysql://127.0.0.1:3306/mytest2?useUnicode=true
以及数据库的账号密码
jdbc.username: root
jdbc.password: 111111

并取消es的注释，意思是输出到es中

接着配置mytest_user.yml文件

vim conf/es/mytest_user.yml

dataSourceKey: defaultDS
destination: example
groupId: g1
esMapping:
  _index: mytest_user
  _type: _doc
  _id: _id
  upsert: true
#  pk: id
  sql: "select a.id as _id, a.name, a.role_id, a.c_time from user a"
#  objFields:
#    _labels: array:;
#  etlCondition: "where a.c_time>='{0}'"
  commitBatch: 3000

（4）先创建mysql表user以及索引students，否则启动canal-adapter会报错

create database canal_test;
use canal_test;
create table students(
    `id` int(10) NOT NULL,
    `name` varchar(100) DEFAULT NULL,
     `age` int(10) DEFAULT NULL,
     `c_time` timestamp NULL DEFAULT CURRENT_TIMESTAMP,
      `c_utime` timestamp NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
       PRIMARY KEY (`id`)
   );

PUT /students
{
  "mappings": {
    "_doc": {
      "properties": {
        "name": {
          "type": "text",
          "fields": {
            "keyword": {
              "type": "keyword"
            }
          }
        },
        "age": {
          "type": "integer"
        },
        "c_time": {
          "type": "date"
        }
      }
    }
  }
}

（5）启动canal-adapter

./bin/startup.sh

查看日志：

cat logs/adapter/adapter.log

（6）测试是否增量同步数据成功
没有数据更新前

GET /students/_search
{
  "took": 1,
  "timed_out": false,
  "_shards": {
    "total": 5,
    "successful": 5,
    "skipped": 0,
    "failed": 0
  },
  "hits": {
    "total": 0,
    "max_score": null,
    "hits": []
  }
}

插入一条数据：

insert students (id, name, age) values(1, "canal", 25);

ES中查询

GET /students/_search

结果是：

{
  "took" : 2,
  "timed_out" : false,
  "_shards" : {
    "total" : 5,
    "successful" : 5,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : 1,
    "max_score" : 1.0,
    "hits" : [
      {
        "_index" : "students",
        "_type" : "_doc",
        "_id" : "1",
        "_score" : 1.0,
        "_source" : {
          "name" : "canal",
          "age" : 25,
          "c_time" : "2020-08-28T14:02:25+08:00"
        }
      }
    ]
  }
}

更新一条数据：

update students set name = 'canal_2' where id = 1;

ES中查询

GET /students/_search

{
  "took" : 2,
  "timed_out" : false,
  "_shards" : {
    "total" : 5,
    "successful" : 5,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : 1,
    "max_score" : 1.0,
    "hits" : [
      {
        "_index" : "students",
        "_type" : "_doc",
        "_id" : "1",
        "_score" : 1.0,
        "_source" : {
          "name" : "canal_2",
          "age" : 25,
          "c_time" : "2020-08-28T14:02:25+08:00"
        }
      }
    ]
  }
}

再插入一条数据：

insert studentsr(id, name, age) values(2, "canal_3", 24);

ES中查询

GET /students/_search

{
  "took" : 1,
  "timed_out" : false,
  "_shards" : {
    "total" : 5,
    "successful" : 5,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : 2,
    "max_score" : 1.0,
    "hits" : [
      {
        "_index" : "students",
        "_type" : "_doc",
        "_id" : "2",
        "_score" : 1.0,
        "_source" : {
          "name" : "canal_3",
          "age" : 24,
          "c_time" : "2020-08-28T14:14:57+08:00"
        }
      },
      {
        "_index" : "students",
        "_type" : "_doc",
        "_id" : "1",
        "_score" : 1.0,
        "_source" : {
          "name" : "canal_2",
          "age" : 25,
          "c_time" : "2020-08-28T14:02:25+08:00"
        }
      }
    ]
  }
}

删除一条数据：

delete from students where id =2 ;

ES中查询

GET /students/_search

{
  "took" : 0,
  "timed_out" : false,
  "_shards" : {
    "total" : 5,
    "successful" : 5,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : 1,
    "max_score" : 1.0,
    "hits" : [
      {
        "_index" : "students",
        "_type" : "_doc",
        "_id" : "1",
        "_score" : 1.0,
        "_source" : {
          "name" : "canal_2",
          "age" : 25,
          "c_time" : "2020-08-28T14:02:25+08:00"
        }
      }
    ]
  }
}

可以看到操作都成功了。

在测试过程中，其实并没有这么的一帆风顺，但是通过查找canal-deployer和canal-adapter的日志可以查看报错进行解决。

canal 也提供了一种全量更新的方式

curl http://127.0.0.1:8081/etl/es/mytest_user.yml -X POST

就可以进行全量更新。

参考文章

https://segmentfault.com/a/1190000019066098?utm_source=tag-newest