Canal——增量同步MySQL数据到ElasticSearch

log-bin=mysql-bin # 开启 binlog
binlog-format=ROW # 选择 ROW 模式
server_id= # 配置 MySQL replaction 需要定义，不要和 canal 的 slaveId 重复

mysql> show variables like 'log_bin%';
+---------------------------------+----------------------------------+
| Variable_name | Value |
+---------------------------------+----------------------------------+
| log_bin | ON |
| log_bin_basename | /data/mysql/data/mysql-bin |
| log_bin_index | /data/mysql/data/mysql-bin.index |
| log_bin_trust_function_creators | OFF |
| log_bin_use_v1_row_events | OFF |
+---------------------------------+----------------------------------+
rows in set (0.00 sec)
mysql> show variables like 'binlog_format%';
+---------------+-------+
| Variable_name | Value |
+---------------+-------+
| binlog_format | ROW |
+---------------+-------+
row in set (0.00 sec)

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

直接下载访问：https://github.com/alibaba/canal/releases ，会列出所有历史的发布版本包 下载方式，比如以1.1.4版本为例子：
wget https://github.com/alibaba/canal/releases/download/canal-1.1.4/canal.deployer-1.1.4.tar.gz自己编译
git clone git@github.com:alibaba/canal.git
cd canal;
mvn clean install -Dmaven.test.skip -Denv=release
编译完成后，会在根目录下产生target/canal.deployer-$version.tar.gz

mkdir /usr/local/canal
tar zxvf canal.deployer-1.1..tar.gz -C /usr/local/canal

cd /usr/local/canal/conf
vi canal.properties

#################################################
######### common argument #############
#################################################
# tcp bind ip
canal.ip =
# register ip to zookeeper
canal.register.ip =#运行canal-server服务的主机IP，可以不用配置，他会自动绑定一个本机的IP
canal.port = 11111 #canal-server监听的端口（TCP模式下，非TCP模式不监听1111端口）
canal.metrics.pull.port = 11112#canal-server metrics.pull监听的端口
# canal instance user/passwd
# canal.user = canal
# canal.passwd = E3619321C1A937C46A0D8BD1DAC39F93B27D4458# canal admin config
#canal.admin.manager = 127.0.0.1:
canal.admin.port =
canal.admin.user = admin
canal.admin.passwd = 4ACFE3202A5FF5CF467898FC58AAB1D615029441canal.zkServers =#集群模式下要配置zookeeper进行协调配置，单机模式可以不用配置
# flush data to zk
canal.zookeeper.flush.period =
canal.withoutNetty = false
# tcp, kafka, RocketMQ
canal.serverMode = tcp#canal-server运行的模式，TCP模式就是直连客户端，不经过中间件。kafka和mq是消息队列的模式
# flush meta cursor/parse position to file
canal.file.data.dir = ${canal.conf.dir}#存放数据的路径
canal.file.flush.period =
## memory store RingBuffer size, should be Math.pow(,n)
canal.instance.memory.buffer.size =
## memory store RingBuffer used memory unit size , default 1kb#下面是一些系统参数的配置，包括内存、网络等
canal.instance.memory.buffer.memunit =
## meory store gets mode used MEMSIZE or ITEMSIZE
canal.instance.memory.batch.mode = MEMSIZE
canal.instance.memory.rawEntry = true## detecing config#这里是心跳检查的配置，做HA时会用到
canal.instance.detecting.enable = false
#canal.instance.detecting.sql = insert into retl.xdual values(,now()) on duplicate key update x=now()
canal.instance.detecting.sql = select
canal.instance.detecting.interval.time =
canal.instance.detecting.retry.threshold =
canal.instance.detecting.heartbeatHaEnable = false# support maximum transaction size, more than the size of the transaction will be cut into multiple transactions delivery
canal.instance.transaction.size =
# mysql fallback connected to new master should fallback times
canal.instance.fallbackIntervalInSeconds = # network config
canal.instance.network.receiveBufferSize =
canal.instance.network.sendBufferSize =
canal.instance.network.soTimeout = # binlog filter config#binlog过滤的配置，指定过滤那些SQL
canal.instance.filter.druid.ddl = true
canal.instance.filter.query.dcl = false
canal.instance.filter.query.dml = false
canal.instance.filter.query.ddl = false
canal.instance.filter.table.error = false
canal.instance.filter.rows = false
canal.instance.filter.transaction.entry = false# binlog format/image check#binlog格式检测，使用ROW模式，非ROW模式也不会报错，但是同步不到数据
canal.instance.binlog.format = ROW,STATEMENT,MIXED
canal.instance.binlog.image = FULL,MINIMAL,NOBLOB# binlog ddl isolation
canal.instance.get.ddl.isolation = false# parallel parser config
canal.instance.parser.parallel = true#并行解析配置，如果是单个CPU就把下面这个true改为false
## concurrent thread number, default % available processors, suggest not to exceed Runtime.getRuntime().availableProcessors()
#canal.instance.parser.parallelThreadSize =
## disruptor ringbuffer size, must be power of
canal.instance.parser.parallelBufferSize = # table meta tsdb info
canal.instance.tsdb.enable = true
canal.instance.tsdb.dir = ${canal.file.data.dir:../conf}/${canal.instance.destination:}
canal.instance.tsdb.url = jdbc:h2:${canal.instance.tsdb.dir}/h2;CACHE_SIZE=;MODE=MYSQL;
canal.instance.tsdb.dbUsername = canal
canal.instance.tsdb.dbPassword = canal
# dump snapshot interval, default hour
canal.instance.tsdb.snapshot.interval =
# purge snapshot expire , default hour( days)
canal.instance.tsdb.snapshot.expire = # aliyun ak/sk , support rds/mq
canal.aliyun.accessKey =
canal.aliyun.secretKey =#################################################
######### destinations #############
#################################################
canal.destinations = example #canal-server创建的实例，在这里指定你要创建的实例的名字，比如test1，test2等，逗号隔开
# conf root dir
canal.conf.dir = ../conf
# auto scan instance dir add/remove and start/stop instance
canal.auto.scan = true
canal.auto.scan.interval = canal.instance.tsdb.spring.xml = classpath:spring/tsdb/h2-tsdb.xml
#canal.instance.tsdb.spring.xml = classpath:spring/tsdb/mysql-tsdb.xmlcanal.instance.global.mode = spring
canal.instance.global.lazy = false
canal.instance.global.manager.address = ${canal.admin.manager}
#canal.instance.global.spring.xml = classpath:spring/memory-instance.xml
canal.instance.global.spring.xml = classpath:spring/file-instance.xml
#canal.instance.global.spring.xml = classpath:spring/default-instance.xml##################################################
######### MQ #############
##################################################
canal.mq.servers = 127.0.0.1:
canal.mq.retries =
canal.mq.batchSize =
canal.mq.maxRequestSize =
canal.mq.lingerMs =
canal.mq.bufferMemory =
canal.mq.canalBatchSize =
canal.mq.canalGetTimeout =
canal.mq.flatMessage = true
canal.mq.compressionType = none
canal.mq.acks = all
#canal.mq.properties. =
canal.mq.producerGroup = test
# Set this value to "cloud", if you want open message trace feature in aliyun.
canal.mq.accessChannel = local
# aliyun mq namespace
#canal.mq.namespace =##################################################
######### Kafka Kerberos Info #############
##################################################
canal.mq.kafka.kerberos.enable = false
canal.mq.kafka.kerberos.krb5FilePath = "../conf/kerberos/krb5.conf"
canal.mq.kafka.kerberos.jaasFilePath = "../conf/kerberos/jaas.conf"

[root@aliyun conf]# vim canal.properties
...
canal.destinations = user_order,delivery_info
...我们需要在根配置的同级目录下创建这两个实例
[root@aliyun conf]# pwd
/usr/local/canal-server/conf
[root@aliyun conf]# cp -a example/ user_order
[root@aliyun conf]# cp -a example/ delivery_info

vi /usr/local/canal/conf/example/instance.properties

###################################################
mysql serverId , v1.0.26+ will autoGencanal.instance.mysql.slaveId=11
# enable gtid use true/false
canal.instance.gtidon=false# position info
canal.instance.master.address=172.16.10.26:3306 #指定要读取binlog的MySQL的IP地址和端口
canal.instance.master.journal.name= #从指定的binlog文件开始读取数据
canal.instance.master.position= #指定偏移量，做过主从复制的应该都理解这两个参数。
#tips：binlog和偏移量也可以不指定，则canal-server会从当前的位置开始读取。我建议不设置
canal.instance.master.timestamp=
canal.instance.master.gtid=# rds oss binlog
canal.instance.rds.accesskey=
canal.instance.rds.secretkey=
canal.instance.rds.instanceId=# table meta tsdb info
canal.instance.tsdb.enable=true
#canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb
#canal.instance.tsdb.dbUsername=canal
#canal.instance.tsdb.dbPassword=canal

#这几个参数是设置高可用配置的，可以配置mysql从库的信息
#canal.instance.standby.address =
#canal.instance.standby.journal.name =
#canal.instance.standby.position =
#canal.instance.standby.timestamp =
#canal.instance.standby.gtid=# username/password
canal.instance.dbUsername=canal #指定连接mysql的用户密码
canal.instance.dbPassword=Aa123456.
canal.instance.connectionCharset = UTF-8 #字符集
# enable druid Decrypt database password
canal.instance.enableDruid=false
#canal.instance.pwdPublicKey=MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBALK4BUxdDltRRE5/zXpVEVPUgunvscYFtEip3pmLlhrWpacX7y7GCMo2/JM6LeHmiiNdH1FWgGCpUfircSwlWKUCAwEAAQ==# table regex
#canal.instance.filter.regex=.*\\..*
canal.instance.filter.regex=risk.canal,risk.cwx #这个是比较重要的参数，匹配库表白名单，比如我只要test库的user表的增量数据，则这样写 test.user
# table black regex
canal.instance.filter.black.regex=
# table field filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2)
#canal.instance.filter.field=test1.t_product:id/subject/keywords,test2.t_company:id/name/contact/ch
# table field black filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2)
#canal.instance.filter.black.field=test1.t_product:subject/product_image,test2.t_company:id/name/contact/ch# mq config
canal.mq.topic=example
# dynamic topic route by schema or table regex
#canal.mq.dynamicTopic=mytest1.user,mytest2\\..*,.*\\..*
canal.mq.partition=
# hash partition config
#canal.mq.partitionsNum=
#canal.mq.partitionHash=test.table:id^name,.*\\..*
#################################################

bin/startup.sh

vi logs/canal/canal.log

-- ::27.967 [main] INFO com.alibaba.otter.canal.deployer.CanalLauncher - ## start the canal server.
-- ::28.113 [main] INFO com.alibaba.otter.canal.deployer.CanalController - ## start the canal server[10.1.29.120:]
-- ::28.210 [main] INFO com.alibaba.otter.canal.deployer.CanalLauncher - ## the canal server is running now ......

vi logs/example/example.log

-- ::45.636 [main] INFO c.a.o.c.i.spring.support.PropertyPlaceholderConfigurer - Loading properties file from class path resource [canal.properties]
-- ::45.641 [main] INFO c.a.o.c.i.spring.support.PropertyPlaceholderConfigurer - Loading properties file from class path resource [example/instance.properties]
-- ::45.803 [main] INFO c.a.otter.canal.instance.spring.CanalInstanceWithSpring - start CannalInstance for -example
-- ::45.810 [main] INFO c.a.otter.canal.instance.spring.CanalInstanceWithSpring - start successful....

bin/stop.sh

访问：https://github.com/alibaba/canal/releases ，会列出所有历史的发布版本包 下载方式，比如以1.1.4版本为例子：
wget https://github.com/alibaba/canal/releases/download/canal-1.1.4/canal.adapter-1.1.4.tar.gz

mkdir /usr/local/canal-adapter
tar zxvf canal.adapter-1.1..tar.gz -C /usr/local/canal-adapter

cd /usr/local/canal-adapter
vim conf/application.yml

server:
port:
spring:
jackson:
date-format: yyyy-MM-dd HH:mm:ss
time-zone: GMT+
default-property-inclusion: non_nullcanal.conf:
mode: tcp #模式
canalServerHost: 127.0.0.1:11111 #指定canal-server的地址和端口
# zookeeperHosts: slave1:
# mqServers: 127.0.0.1: #or rocketmq
# flatMessage: true
batchSize:
syncBatchSize:
retries:
timeout:
accessKey:
secretKey:
srcDataSources: #数据源配置，从哪里获取数据
defaultDS: #指定一个名字，在ES的配置中会用到，唯一
url: jdbc:mysql://172.16.10.26:3306/risk?useUnicode=true
username: root
password: 123456
canalAdapters:
- instance: example # canal instance Name or mq topic name #指定在canal-server配置的实例
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:
# - name: rdb
# key: oracle1
# properties:
# jdbc.driverClassName: oracle.jdbc.OracleDriver
# jdbc.url: jdbc:oracle:thin:@localhost::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:
# threads:
# commitSize:
# - name: hbase
# properties:
# hbase.zookeeper.quorum: 127.0.0.1
# hbase.zookeeper.property.clientPort:
# zookeeper.znode.parent: /hbase
- name: es #输出到哪里，指定es
hosts: 172.16.99.2:40265 #指定es的地址，注意端口为es的传输端口9300
properties:
# mode: transport # or rest
# security.auth: test: # only used for rest mode
cluster.name: log-es-cluster #指定es的集群名称

[root@aliyun es]# pwd
/usr/local/canal-adapter/conf/es
[root@aliyun es]# ll
total
-rwxrwxrwx root root Apr : biz_order.yml#这三个配置文件是自带的，可以删除，不过最好不要删除，因为可以参考他的格式
-rwxrwxrwx root root Apr : customer.yml
-rwxrwxrwx root root Apr : mytest_user.yml

cp customer.yml canal.yml
vim conf/es/canal.yml

dataSourceKey: defaultDS #指定数据源，这个值和adapter的application.yml文件中配置的srcDataSources值对应。
destination: example #指定canal-server中配置的某个实例的名字，注意：我们可能配置多个实例，你要清楚的知道每个实例收集的是那些数据，不要瞎搞。
groupId: g1 #组ID，默认就好
esMapping: #ES的mapping（映射）
_index: canal #要同步到的ES的索引名称（自定义），需要自己在ES上创建哦！
_type: _doc #ES索引的类型名称（自定义）
_id: _id #ES标示文档的唯一标示，通常对应数据表中的主键ID字段，注意我这里写成的是"_id"，有个下划线哦！
#pk: id #如果不需要_id, 则需要指定一个属性为主键属性
sql: "select t.id as _id, t.name, t.sex, t.age, t.amount, t.email, t.occur_time from canal t" #这里就是数据表中的每个字段到ES索引中叫什么名字的sql映射，注意映射到es中的每个字段都要是唯一的，不能重复。
#etlCondition: "where t.occur_time>='{0}'"
commitBatch: 3000

CREATE TABLE `canal` (
id int() NOT NULL AUTO_INCREMENT,
name varchar() NULL COMMENT '名称',
sex varchar() NULL COMMENT '性别',
age int NULL COMMENT '年龄',
amount decimal(,) NULL COMMENT '资产',
email varchar() NULL COMMENT '邮箱',
occur_time timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (`ID`)
) ENGINE=InnoDB AUTO_INCREMENT= DEFAULT CHARSET=utf8;

{
"mappings": {
"_doc": {
"properties": {
"id": {
"type": "long"
},
"name": {
"type": "text"
},
"sex": {
"type": "text"
},
"age": {
"type": "long"
},
"amount": {
"type": "text"
},
"email": {
"type": "text"
},
"occur_time": {
"type": "date"
}
}
}
}
}

cd /usr/local/canal-adapter
./bin/startup.sh

cat logs/adapter/adapter.log

insert into canal(id,name,sex,age,amount,email,occur_time) values(null,'cwx','男',,,'249299170@qq.com',now());

update canal set name='cwx1',sex='女',age=,amount=,email='asdf',occur_time=now() where id=;

delete from canal where id=;