canal同步mysql數據至elasticSearch

時間 2019-12-23

標籤 canal 同步 mysql 數據 elasticsearch 欄目 MySQL 简体版

原文原文鏈接

數據庫帳號受權mysql

由於canal是模擬mysql的slave去偷取msql的binlog操做日誌達到同步的效果，因此須要給帳號受權；sql

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

用戶名：root 密碼: 123456 指向ip(容許訪問的ip): %表明全部Ip,此處也能夠輸入Ip來指定Ip數據庫

開啓mysql binlog模式windows

找到mysql配置文件，windows通常在C:\ProgramData\MySQL\MySQL Server 5.7app

Linux通常在/etc/my.cnf；添加以下內容：url

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

而後重啓mysql，查看binlog是否開啓：show variables like '%log_bin%'; log_bin爲on即開啓成功。spa

配置canaldePloyer日誌

進入conf/example/instance.properties配置文件並更改以下配置code

canal.instance.master.address=127.0.0.1:3306
canal.instance.dbUsername=root   （第一步所受權的帳號）
canal.instance.dbPassword=123456
canal.instance.defaultDatabaseName=test
canal.instance.filter.regex=test.tzg_course_t,test.tzg_school_t （表過濾：設置要同步的表）

配置完成便可啓動；orm

配置canalAdapter

配置application.yml

進入conf/application.yml;

將defaultDS改成本身的數據庫配置

srcDataSources:
    defaultDS:
      url: jdbc:mysql://127.0.0.1:3306/test?useUnicode=true
      username: root
      password: 123456

配置本身的es

- name: es
        hosts: 127.0.0.1:9300 #transport端口
        properties:
          cluster.name: tzg  #集羣名

配置數據抽取文件

上述步驟配置完成保存後回到上級目錄，再進入es目錄，建立一個yml配置文件相似以下：

dataSourceKey: defaultDS
destination: example
groupId: g1
esMapping:
  _index: tzg_business.tzg_course  #待同步es index，要提早建好
  _type: _doc  #文檔類型
  _id: _id   #文檔_id和相同
  upsert: true 
#  pk: id
  sql: "select c.id as _id,c.id as id,s.area_id as areaId,s.area as area,s.district_id as districtId,s.district as district,s.city_id as cityId,s.city      as city,c.category as category,c.category_id as categoryId,c.title as title ,c.school_id as schoolId,c.description as description,
  c.score as score,c.tags as tags,c.cover_img_id as coverImgId,c.recommended as recommended,c.promote as promote,
  c.last_modify_time as lastModifyTime,c.create_time as createTime from tzg_course_t c left join tzg_school_t s on c.school_id= s.id"
#  objFields:
#    _labels: array:;
#  etlCondition: "where a.c_time>='{0}'"
  commitBatch: 3000

配置完成啓動canalAdapter便可；增長修改數據庫數據看是否同步es；

deployer：