canal同步mysql数据至elasticSearch

数据库帐号受权

由于canal是模拟mysql的slave去偷取msql的binlog操做日志达到同步的效果，因此须要给帐号受权；

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

用户名：root      密码: 123456     指向ip(容许访问的ip): %表明全部Ip,此处也能够输入Ip来指定Ip

开启mysql binlog模式

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

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

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

而后重启mysql，查看binlog是否开启：show variables like '%log_bin%'; log_bin为on即开启成功。

配置canaldePloyer

进入conf/example/instance.properties配置文件并更改以下配置

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 （表过滤：设置要同步的表）

配置完成便可启动；

配置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：

adapter：