canal1.1.5不能同步update（修改）类型的数据到ES中
canal
alibaba/canal: Canal 是由阿里巴巴开源的分布式数据库同步系统，主要用于实现MySQL数据库的日志解析和实时增量数据订阅与消费，广泛应用于数据库变更消息的捕获、数据迁移、缓存更新等场景。
项目地址：https://gitcode.com/gh_mirrors/ca/canal
免费下载资源
·
目录
1、canal->deployer配置
1.1、canal.properties文件具体配置
案例中文件的详细配置
# tcp bind ip
canal.ip =
# register ip to zookeeper
canal.register.ip =
canal.port = 11111
canal.metrics.pull.port = 11112
# canal instance user/passwd
# canal.user = canal
# canal.passwd = E3619321C1A937C46A0D8BD1DAC39F93B27D4458
# canal admin config
#canal.admin.manager = 127.0.0.1:8089
canal.admin.port = 11110
canal.admin.user = admin
canal.admin.passwd = 4ACFE3202A5FF5CF467898FC58AAB1D615029441
# admin auto register
#canal.admin.register.auto = true
#canal.admin.register.cluster =
#canal.admin.register.name =
canal.zkServers =
# flush data to zk
canal.zookeeper.flush.period = 1000
canal.withoutNetty = false
# tcp, kafka, rocketMQ, rabbitMQ
canal.serverMode = tcp
# flush meta cursor/parse position to file
canal.file.data.dir = ${canal.conf.dir}
canal.file.flush.period = 1000
## memory store RingBuffer size, should be Math.pow(2,n)
canal.instance.memory.buffer.size = 16384
## memory store RingBuffer used memory unit size , default 1kb
canal.instance.memory.buffer.memunit = 1024
## meory store gets mode used MEMSIZE or ITEMSIZE
canal.instance.memory.batch.mode = MEMSIZE
canal.instance.memory.rawEntry = true
## detecing config
canal.instance.detecting.enable = false
#canal.instance.detecting.sql = insert into retl.xdual values(1,now()) on duplicate key update x=now()
canal.instance.detecting.sql = select 1
canal.instance.detecting.interval.time = 3
canal.instance.detecting.retry.threshold = 3
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 = 1024
# mysql fallback connected to new master should fallback times
canal.instance.fallbackIntervalInSeconds = 60
# network config
canal.instance.network.receiveBufferSize = 16384
canal.instance.network.sendBufferSize = 16384
canal.instance.network.soTimeout = 30
# binlog filter config
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
canal.instance.filter.dml.insert = false
canal.instance.filter.dml.update = false
canal.instance.filter.dml.delete = false
# binlog format/image check
canal.instance.binlog.format = ROW,STATEMENT,MIXED
#canal.instance.binlog.format = ROW
icanal.instance.binlog.image = FULL,MINIMAL,NOBLOB
#canal.instance.binlog.image = FULL
# binlog ddl isolation
canal.instance.get.ddl.isolation = false
# parallel parser config
canal.instance.parser.parallel = true
## concurrent thread number, default 60% available processors, suggest not to exceed Runtime.getRuntime().availableProcessors()
#canal.instance.parser.parallelThreadSize = 16
## disruptor ringbuffer size, must be power of 2
canal.instance.parser.parallelBufferSize = 256
# 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=1000;MODE=MYSQL;
canal.instance.tsdb.dbUsername = canal
canal.instance.tsdb.dbPassword = canal
# dump snapshot interval, default 24 hour
canal.instance.tsdb.snapshot.interval = 24
# purge snapshot expire , default 360 hour(15 days)
canal.instance.tsdb.snapshot.expire = 360
#################################################
######### destinations #############
#################################################
canal.destinations = example
# 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 = 5
# set this value to 'true' means that when binlog pos not found, skip to latest.
# WARN: pls keep 'false' in production env, or if you know what you want.
canal.auto.reset.latest.pos.mode = false
canal.instance.tsdb.spring.xml = classpath:spring/tsdb/h2-tsdb.xml
#canal.instance.tsdb.spring.xml = classpath:spring/tsdb/mysql-tsdb.xml
canal.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 Properties #############
##################################################
# aliyun ak/sk , support rds/mq
canal.aliyun.accessKey =
canal.aliyun.secretKey =
canal.aliyun.uid=
canal.mq.flatMessage = true
canal.mq.canalBatchSize = 50
canal.mq.canalGetTimeout = 100
# Set this value to "cloud", if you want open message trace feature in aliyun.
canal.mq.accessChannel = local
canal.mq.database.hash = true
canal.mq.send.thread.size = 30
canal.mq.build.thread.size = 8
##################################################
######### Kafka #############
##################################################
kafka.bootstrap.servers = 127.0.0.1:9092
kafka.acks = all
kafka.compression.type = none
kafka.batch.size = 16384
kafka.linger.ms = 1
kafka.max.request.size = 1048576
kafka.buffer.memory = 33554432
kafka.max.in.flight.requests.per.connection = 1
kafka.retries = 0
kafka.kerberos.enable = false
kafka.kerberos.krb5.file = "../conf/kerberos/krb5.conf"
kafka.kerberos.jaas.file = "../conf/kerberos/jaas.conf"
##################################################
######### RocketMQ #############
##################################################
rocketmq.producer.group = test
rocketmq.enable.message.trace = false
rocketmq.customized.trace.topic =
rocketmq.namespace =
rocketmq.namesrv.addr = 127.0.0.1:9876
rocketmq.retry.times.when.send.failed = 0
rocketmq.vip.channel.enabled = false
rocketmq.tag =
##################################################
######### RabbitMQ #############
##################################################
rabbitmq.host =
rabbitmq.virtual.host =
rabbitmq.exchange =
rabbitmq.username =
rabbitmq.password =
rabbitmq.deliveryMode =
1.2 vim conf/example/instance.properties
本案例使用example实例进行测试，文件的详细配置如下：
2、启动deployer
进入到canal的deployer下的bin目录启动 命令为./bin/startup.sh
./bin/startup.sh
3、设置canal-adapter 配置文件
使用命令修改application.yml文件
vim conf/application.yml
4、配置canal到ES的映射
新增t_transport.yml文件并编辑
vim conf/es7/t_transport.yml
5、启动cana-adapter
./bin/startup.sh
问题：如果修改数据库某张表的某个字段时出现如下日志：
2022-08-24 11:37:37.735 [pool-2-thread-1] INFO c.a.o.canal.client.adapter.logger.LoggerAdapterExample - DML: {"data":[{"project_name":"黔西南水泥项目（示例）88888"}],"database":"fleet","destination":"example","es":1661312257000,"groupId":"g1","isDdl":false,"old":null,"pkNames":["id"],"sql":"","table":"t_transport","ts":1661312257735,"type":"UPDATE"}
从日志中看到old属性为null，说明没有获取到源数据信息，从而修改无效无法同步修改后的数据到ES中。解决此问题需要做如下设置
设置源数据库MySql的my.cnf的binlog配置。把两个关键的参数做如下设置
binlog_format = ROW
binlog_row_image = FULL
使云数据库修改后立即生效，其他模式可能需要重启
6、再次测试
修改某条数据的某个字段验证是否修改同步成功：查看日志
2022-08-24 11:37:37.735 [pool-2-thread-1] INFO c.a.o.canal.client.adapter.logger.LoggerAdapterExample - DML: {"data":[{"id":1560555730631700481,"no":"YD2022081900013","status":1,"project_id":1560544760224141313,"project_name":"黔西南水泥项目（示例）88888","order_id":1560544948384813058,"order_no":"JH2022819163123","owner_id":1559007671070244865,"owner_name":"贵州省卓跃实业有限公司","owner_phone":"18188239125","unloading_person":"张三","unloading_person_phone":"15285458306","loading_place_title":"高科一号A栋","loading_place":"贵州省贵阳市观山湖区-长岭南路40号高科一号-高科一号A栋","loading_place_lon":106.646152,"loading_place_lat":26.618197,"unloading_place_title":"高科一号C栋","unloading_place":"贵州省贵阳市观山湖区-长岭南路160号C号楼-高科一号C栋","unloading_place_lon":106.647312,"unloading_place_lat":26.618864,"loading_person":null,"loading_person_phone":null,"sign_time":null,"settle_time":null,"take_time":null,"in_time":null,"out_time":null,"down_time":null,"driver_id":null,"driver_name":null,"driver_phone":null,"car_number":"贵ap0n33","car_type":"平板车-2轴","car_load":null,"device_id":null,"device_type":null,"remark":"","create_user":1559007673167446017,"create_time":1660900455000,"update_user":1559007673167446017,"update_time":1660897884000,"is_deleted":0,"tenant_id":"004981","order_type":1,"project_deadline":1661184000000,"up_time":null,"pound_affirm_time":null,"cargo_amount":0.0,"source_amount":0.0,"platform_amount":0.0,"transport_amount":0.0,"commission_amount":0.0,"driver_no":null,"exception_status":0,"weight":0.0,"distance":null,"expect_unloading_final_time":1661243483000,"expect_unloading_duration":"96","payee_id":1540231034833031169,"payee_name":"李四","payee_phone":"15285458305","transport_type":0,"start_time":null,"suggest_weight":10.0,"sign_user_id":null,"sign_user_name":null,"order_product_stock":null}],"database":"fleet","destination":"example","es":1661312257000,"groupId":"g1","isDdl":false,"old":[{"project_name":"黔西南水泥项目（示例）55555"}],"pkNames":["id"],"sql":"","table":"t_transport","ts":1661312257735,"type":"UPDATE"}
old已经有源数据信息，再查看ES发现已经同步成功
到此问题处理完成
GitHub 加速计划 / ca / canal
28.22 K
7.57 K
下载
alibaba/canal: Canal 是由阿里巴巴开源的分布式数据库同步系统，主要用于实现MySQL数据库的日志解析和实时增量数据订阅与消费，广泛应用于数据库变更消息的捕获、数据迁移、缓存更新等场景。
最近提交(Master分支：3 个月前 )
1e5b8a20 - 2 个月前
ff82fd65
2 个月前
旨在为数千万中国开发者提供一个无缝且高效的云端环境，以支持学习、使用和贡献开源项目。
更多推荐
2
0- 0
已为社区贡献3条内容
所有评论(0)