Canal realizes real-time synchronization of data from Mysql to es

1. Environmental preparation ：

1.1  mysql:5.7

1.2 elasticsearch:7.4.2

1.3  kibana:7.4.2

1.4 Server side ：canal-deployer1.1.5

1.5 client ：canal-adapter 1.1.5

2. Download and install MySQL（ notes ： My projects are all docker install   The installation steps are not shown here   See Baidu or see my other articles for details ） modify mysql The configuration file Turn on binlog journal , And take ROW The way , Turn on master-slave mode as well as logbin File location of as follows ：

server_id=101
binlog-ignore-db=mysql
log-bin=mall-mysql-bin
binlog_cache_size=1M
binlog_format=row
expire_logs_days=7
slave_skip_errors=1062

2.1  Remember to restart mysql    see MySQL Configure whether to turn on

SHOW VARIABLES LIKE 'binlog-format'; -- The result should be ROW
SHOW VARIABLES LIKE 'log_bin'; -- The result should be ON
SHOW VARIABLES LIKE '%log%'; -- all binlog Information

 2.2  And then to canal Create a canal Account   The order is as follows ：

CREATE USER canal IDENTIFIED BY 'canal';  
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
FLUSH PRIVILEGES;

2.3 Create a canal-test Kuhe sys_log The test table

DROP TABLE IF EXISTS `sys_log`;
CREATE TABLE `sys_log`  (
  `id` varchar(64) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT ' Number ',
  `type` char(1) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT '1' COMMENT ' Log type ',
  `title` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT '' COMMENT ' Journal Title ',
  `create_by` varchar(64) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT ' The creator ',
  `create_date` datetime NULL DEFAULT NULL COMMENT ' Creation time ',
  `remote_addr` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT ' operation IP Address ',
  `user_agent` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT ' The user agent ',
  `request_uri` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT ' request URI',
  `method` varchar(5) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT ' Mode of operation ',
  `params` text CHARACTER SET utf8 COLLATE utf8_general_ci NULL COMMENT ' Operate the submitted data ',
  `exception` text CHARACTER SET utf8 COLLATE utf8_general_ci NULL COMMENT ' Abnormal information ',
  PRIMARY KEY (`id`) USING BTREE
) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_general_ci COMMENT = ' Log table ' ROW_FORMAT = Dynamic;

3. Download and install ES and Kibana  It should be noted here that the two versions need to be consistent     I won't introduce it here     I installed the following ：

4. download canal Server side of   canal-deployer1.1.5 And the client  canal-adapter 1.1.5  Download at ：https://github.com/alibaba/canal/releases

5. Unzip these two projects command ：

tar -zxvf  canal.adapter-1.1.5.tar.gz

tar -zxvf  canal.deployer-1.1.5.tar.gz

6. Modify the server  canal.deployer-1.1.5 Next conf Directory name example by orgdeer-cui  command ：mv  example  orgdeer-cui    Then change the configuration conf/orgdeer-cui/instance.properties     It is mainly to modify the database configuration     Change the following three places     as follows ：

6. Start the server side canal.deployer-1.1.5   

Toggle directory ：cd  /home/canal/canal.deployer-1.1.5/bin

start-up ：./startup.sh

7. Check whether the log is started successfully ：

tail -f /home/canal/canal.deployer-1.1.5/logs/canal/canal.log 

8. Modify the client   canal.adapter-1.1.5 Configuration below  application.yml     The main thing is to modify canal-server To configure 、 Data source configuration and client adapter configuration     Mainly change the annotated part as follows ：

server:
  port: 8081
logging:
  level:
    com.alibaba.otter.canal.client.adapter.es: DEBUG
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
  # IMPORTANT!!! KEEP ADAPER SYNC CONSISTANT
  retries: -1
  timeout:
  accessKey:
  secretKey:
  consumerProperties:
    # canal tcp consumer
    canal.tcp.server.host: 127.0.0.1:11111  # Server address
    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&characterEncoding=UTF-8&serverTimezone=GMT%2B8&useSSL=false&useInformationSchema=false&allowPublicKeyRetrieval=true   #mysql Link address
      username: root   # user name
      password: hzp123456  # password
  canalAdapters:
  - instance: orgdeer-cui # Corresponding to the configuration in the server mq topic name
    groups:
    - groupId: g1
      outerAdapters:
      - name: logger  # Log print adapter
      - name: es7  #ES Synchronization adapter
        hosts: 127.0.0.1:9300  #ESl Link address
        properties:
          mode: transport  # Mode options transport（9300） perhaps rest(9200)
          cluster.name: elasticsearch  #ES Cluster name
HOSTNAME%%.*:
PWD/#$HOME/~:

9. Start client  canal.adapter-1.1.5   

Toggle directory ：cd /home/canal/canal.adapter-1.1.5/bin

start-up ：./startup.sh

10 Check whether the client log is started successfully  

tail -f  /home/canal/canal.adapter-1.1.5/logs/adapter/adapter.log 

11. Add configuration file canal-adapter/conf/es7/sys_log.yml, Used for configuration MySQL The table in and Elasticsearch The mapping relation of index in （ Be careful destination name orgdeer-cui To follow the server side mq topic name Corresponding ） Then put the file sys_log.yml Put it in /home/canal/canal.adapter-1.1.5/conf/es7 Under the table of contents  

dataSourceKey: defaultDS
destination: orgdeer-cui
groupId: g1
esMapping:
  _index: sys_log
  _id: _id
  sql: "select
        id as _id, type,
         title,
         create_by as createBy,
         UNIX_TIMESTAMP(create_date) as createDate,
        remote_addr as remoteAddr,
        user_agent as userAgent,
        request_uri as requestUri,method,
        params,exception
        from sys_log"
  etlCondition: "where create_date>={}"
  commitBatch: 3000

12. stay Kibana Create... In the console sys_log Indexes

PUT sys_log
{
  "mappings": {
      "properties":{
        "type":{
          "type":"keyword"
        },
        "title":{
          "type":"text"
        },
        "createBy":{
          "type":"keyword"
        },
        "remoteAddr":{
          "type": "text"
        },
        "userAgent":{
          "type":"text"
        },
        "requestUri": {
          "type": "text"
        },
        "method": {
          "type": "keyword"
        },
        "params": {
          "type": "text"
        },
       "exception": {
          "type": "text"
        },
        "createDate": {
           "type": "long"
        }
      }
  }
}

13. Triggered by command , Give Way canal-adapter Read the dml journal , Synchronize to es Library   The order is as follows ：

curl -X POST http://127.0.0.1:8081/etl/es7/sys_log.yml

see ES Data synchronization in ：

  Delete manually   Add data    ES The data will also be synchronized