canal同步mysql到elasticsearch

1、mysql配置

1、編輯mysql配置文件

docker exec -it mysql5.7 /bin/bash    #mysql5.7為容器名稱
cd etc
vi my.cnf

新增如下配置：

log-bin=mysql-bin          #添加這一行就ok
binlog-format=ROW          #選擇row模式
server_id=1                #配置mysql replaction需要定義，不能和canal的slaveId重復
expire_logs_days=5         #日志過期時間為5天      

2、新建數據庫用戶，並賦予相應權限

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

3、新建測試數據庫

新建mytest數據庫，並新建兩張表

SET NAMES utf8mb4;
SET FOREIGN_KEY_CHECKS = 0;

-- ----------------------------
-- Table structure for role
-- ----------------------------
DROP TABLE IF EXISTS `role`;
CREATE TABLE `role`  (
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  `role_name` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NULL DEFAULT NULL,
  PRIMARY KEY (`id`) USING BTREE
) ENGINE = InnoDB AUTO_INCREMENT = 3 CHARACTER SET = latin1 COLLATE = latin1_swedish_ci ROW_FORMAT = Dynamic;

-- ----------------------------
-- Table structure for user
-- ----------------------------
DROP TABLE IF EXISTS `user`;
CREATE TABLE `user`  (
  `id` bigint(255) NOT NULL AUTO_INCREMENT,
  `name` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NULL DEFAULT NULL,
  `role_id` bigint(20) NULL DEFAULT NULL,
  `c_time` datetime(0) NULL DEFAULT NULL,
  PRIMARY KEY (`id`) USING BTREE
) ENGINE = InnoDB AUTO_INCREMENT = 6 CHARACTER SET = latin1 COLLATE = latin1_swedish_ci ROW_FORMAT = Dynamic;

SET FOREIGN_KEY_CHECKS = 1;　

2、安裝canal-server和canal-admin

1、安裝

具體安裝操作可以參考canal-server安裝和canal-admin安裝　　

主要有幾個注意點：

1. 如果兩個都要裝，建議先安裝canal-admin,我在實際操作過程中遇到了先安裝canal-server之后再安裝canal-admin,canal-admin的server管理中發現不了之前安裝的服務
2. 如果先安裝了canal-server，可以修改conf/canal_local.properties文件中的配置，這樣canal-admin可以發現canal-server

2、配置

打開canal-admin,可以在server管理頁面下看到安裝好的server

打開Instance管理頁面新增實例,並修改相應配置

 

3、在ElasticSearch中新建兩個索引

4、運行canal-adapter

1、下載canel-1.1.4版本源碼

git clone -b canal-1.1.4 https://github.com/alibaba/canal.git

2、打開其中的client-adapter

 

3、修改launcher的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: 192.168.1.223:11111    #修改為canal-server的IP，端口不變
  batchSize: 500
  syncBatchSize: 1000
  retries: 0
  timeout:
  accessKey:
  secretKey:
  srcDataSources:
    defaultDS:
      url: jdbc:mysql://192.168.1.223:3306/mytest?useUnicode=true 
      username: canal
      password: canal
  canalAdapters:
  - instance: lietou # canal instance Name or mq topic name 此處的實例名為上文創建時候的實例名稱
    groups:
    - groupId: g1
      outerAdapters:
      - name: es
        hosts: 192.168.1.223:9200 # 127.0.0.1:9200 for rest mode
        properties:
          mode: rest # or rest|transport 此處默認是transport,修改為rest
          cluster.name: docker-cluster #此處的名稱可以通過上面配置的hosts的地址在瀏覽器訪問查看

4、在launcher下的的resources下新建es文件夾

新增mytest_role.yml

dataSourceKey: defaultDS #與application.yml中的srcDataSources對應
destination: lietou #此處的實例名為上文配置的名稱
groupId: g1
esMapping:
  _index: mytest_role
  _type: _doc
  _id: _id
  upsert: true
  #  pk: id
  sql: "select a.id as _id, a.role_name as _role_name from role a"
  #  objFields:
  #    _labels: array:;
  etlCondition: "where a.id>={}"
  commitBatch: 3000

  新增mytest_user.yml

dataSourceKey: defaultDS #與application.yml中的srcDataSources對應
destination: lietou  #此處的實例名為上文配置的名稱
groupId: g1
esMapping:
  _index: mytest_user
  _type: _doc
  _id: _id
  upsert: true
  #  pk: id
  sql: "select a.id as _id, a.name as _name, a.role_id as _role_id, b.role_name as _role_name,
        a.c_time as _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

5、啟動launher項目

然后在數據庫中新增數據查看效果是否成功

查詢elasticsearch是否成功寫入

 

6、全量同步

curl  -X POST http://192.168.1.14:8081/etl/es/mytest_sysuser.yml   #此處的ip地址為運行adapter機器的IP

　　

 

至此，使用canel同步mysql到elasticsearch成功！