• canal1.1.5不能同步update(修改)类型的数据到ES中


    目录

    1、canal->deployer配置

     1.1、canal.properties文件具体配置

     2、启动deployer

    3、设置canal-adapter 配置文件

     4、配置canal到ES的映射

     5、启动cana-adapter

    问题:如果修改数据库某张表的某个字段时出现如下日志:

    6、再次测试


    1、canal->deployer配置

     1.1、canal.properties文件具体配置

            案例中文件的详细配置

    1. # tcp bind ip
    2. canal.ip =
    3. # register ip to zookeeper
    4. canal.register.ip =
    5. canal.port = 11111
    6. canal.metrics.pull.port = 11112
    7. # canal instance user/passwd
    8. # canal.user = canal
    9. # canal.passwd = E3619321C1A937C46A0D8BD1DAC39F93B27D4458
    10. # canal admin config
    11. #canal.admin.manager = 127.0.0.1:8089
    12. canal.admin.port = 11110
    13. canal.admin.user = admin
    14. canal.admin.passwd = 4ACFE3202A5FF5CF467898FC58AAB1D615029441
    15. # admin auto register
    16. #canal.admin.register.auto = true
    17. #canal.admin.register.cluster =
    18. #canal.admin.register.name =
    19. canal.zkServers =
    20. # flush data to zk
    21. canal.zookeeper.flush.period = 1000
    22. canal.withoutNetty = false
    23. # tcp, kafka, rocketMQ, rabbitMQ
    24. canal.serverMode = tcp
    25. # flush meta cursor/parse position to file
    26. canal.file.data.dir = ${canal.conf.dir}
    27. canal.file.flush.period = 1000
    28. ## memory store RingBuffer size, should be Math.pow(2,n)
    29. canal.instance.memory.buffer.size = 16384
    30. ## memory store RingBuffer used memory unit size , default 1kb
    31. canal.instance.memory.buffer.memunit = 1024
    32. ## meory store gets mode used MEMSIZE or ITEMSIZE
    33. canal.instance.memory.batch.mode = MEMSIZE
    34. canal.instance.memory.rawEntry = true
    35. ## detecing config
    36. canal.instance.detecting.enable = false
    37. #canal.instance.detecting.sql = insert into retl.xdual values(1,now()) on duplicate key update x=now()
    38. canal.instance.detecting.sql = select 1
    39. canal.instance.detecting.interval.time = 3
    40. canal.instance.detecting.retry.threshold = 3
    41. canal.instance.detecting.heartbeatHaEnable = false
    42. # support maximum transaction size, more than the size of the transaction will be cut into multiple transactions delivery
    43. canal.instance.transaction.size = 1024
    44. # mysql fallback connected to new master should fallback times
    45. canal.instance.fallbackIntervalInSeconds = 60
    46. # network config
    47. canal.instance.network.receiveBufferSize = 16384
    48. canal.instance.network.sendBufferSize = 16384
    49. canal.instance.network.soTimeout = 30
    50. # binlog filter config
    51. canal.instance.filter.druid.ddl = true
    52. canal.instance.filter.query.dcl = false
    53. canal.instance.filter.query.dml = false
    54. canal.instance.filter.query.ddl = false
    55. canal.instance.filter.table.error = false
    56. canal.instance.filter.rows = false
    57. canal.instance.filter.transaction.entry = false
    58. canal.instance.filter.dml.insert = false
    59. canal.instance.filter.dml.update = false
    60. canal.instance.filter.dml.delete = false
    61. # binlog format/image check
    62. canal.instance.binlog.format = ROW,STATEMENT,MIXED
    63. #canal.instance.binlog.format = ROW
    64. icanal.instance.binlog.image = FULL,MINIMAL,NOBLOB
    65. #canal.instance.binlog.image = FULL
    66. # binlog ddl isolation
    67. canal.instance.get.ddl.isolation = false
    68. # parallel parser config
    69. canal.instance.parser.parallel = true
    70. ## concurrent thread number, default 60% available processors, suggest not to exceed Runtime.getRuntime().availableProcessors()
    71. #canal.instance.parser.parallelThreadSize = 16
    72. ## disruptor ringbuffer size, must be power of 2
    73. canal.instance.parser.parallelBufferSize = 256
    74. # table meta tsdb info
    75. canal.instance.tsdb.enable = true
    76. canal.instance.tsdb.dir = ${canal.file.data.dir:../conf}/${canal.instance.destination:}
    77. canal.instance.tsdb.url = jdbc:h2:${canal.instance.tsdb.dir}/h2;CACHE_SIZE=1000;MODE=MYSQL;
    78. canal.instance.tsdb.dbUsername = canal
    79. canal.instance.tsdb.dbPassword = canal
    80. # dump snapshot interval, default 24 hour
    81. canal.instance.tsdb.snapshot.interval = 24
    82. # purge snapshot expire , default 360 hour(15 days)
    83. canal.instance.tsdb.snapshot.expire = 360
    84. #################################################
    85. ######### destinations #############
    86. #################################################
    87. canal.destinations = example
    88. # conf root dir
    89. canal.conf.dir = ../conf
    90. # auto scan instance dir add/remove and start/stop instance
    91. canal.auto.scan = true
    92. canal.auto.scan.interval = 5
    93. # set this value to 'true' means that when binlog pos not found, skip to latest.
    94. # WARN: pls keep 'false' in production env, or if you know what you want.
    95. canal.auto.reset.latest.pos.mode = false
    96. canal.instance.tsdb.spring.xml = classpath:spring/tsdb/h2-tsdb.xml
    97. #canal.instance.tsdb.spring.xml = classpath:spring/tsdb/mysql-tsdb.xml
    98. canal.instance.global.mode = spring
    99. canal.instance.global.lazy = false
    100. canal.instance.global.manager.address = ${canal.admin.manager}
    101. #canal.instance.global.spring.xml = classpath:spring/memory-instance.xml
    102. canal.instance.global.spring.xml = classpath:spring/file-instance.xml
    103. #canal.instance.global.spring.xml = classpath:spring/default-instance.xml
    104. ##################################################
    105. ######### MQ Properties #############
    106. ##################################################
    107. # aliyun ak/sk , support rds/mq
    108. canal.aliyun.accessKey =
    109. canal.aliyun.secretKey =
    110. canal.aliyun.uid=
    111. canal.mq.flatMessage = true
    112. canal.mq.canalBatchSize = 50
    113. canal.mq.canalGetTimeout = 100
    114. # Set this value to "cloud", if you want open message trace feature in aliyun.
    115. canal.mq.accessChannel = local
    116. canal.mq.database.hash = true
    117. canal.mq.send.thread.size = 30
    118. canal.mq.build.thread.size = 8
    119. ##################################################
    120. ######### Kafka #############
    121. ##################################################
    122. kafka.bootstrap.servers = 127.0.0.1:9092
    123. kafka.acks = all
    124. kafka.compression.type = none
    125. kafka.batch.size = 16384
    126. kafka.linger.ms = 1
    127. kafka.max.request.size = 1048576
    128. kafka.buffer.memory = 33554432
    129. kafka.max.in.flight.requests.per.connection = 1
    130. kafka.retries = 0
    131. kafka.kerberos.enable = false
    132. kafka.kerberos.krb5.file = "../conf/kerberos/krb5.conf"
    133. kafka.kerberos.jaas.file = "../conf/kerberos/jaas.conf"
    134. ##################################################
    135. ######### RocketMQ #############
    136. ##################################################
    137. rocketmq.producer.group = test
    138. rocketmq.enable.message.trace = false
    139. rocketmq.customized.trace.topic =
    140. rocketmq.namespace =
    141. rocketmq.namesrv.addr = 127.0.0.1:9876
    142. rocketmq.retry.times.when.send.failed = 0
    143. rocketmq.vip.channel.enabled = false
    144. rocketmq.tag =
    145. ##################################################
    146. ######### RabbitMQ #############
    147. ##################################################
    148. rabbitmq.host =
    149. rabbitmq.virtual.host =
    150. rabbitmq.exchange =
    151. rabbitmq.username =
    152. rabbitmq.password =
    153. 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发现已经同步成功

    到此问题处理完成 

  • 相关阅读:
    TypeScript学习一(基础类型)
    【STM32F407+CUBEMX+FreeRTOS+lwIP之UDP记录】
    GFS分布式存储
    灰色关联度分析-详细代码和说明
    OpenJudge NOI题库 1.7 编程基础之字符串
    【二分】Pythagorean Triples—CF1487D
    使用Oracle VM VirtualBox安装Unbuntu虚拟机并安装增强功能(实现双向复制粘贴)
    ElementUI之动态树+数据表格+分页->动态树,动态表格
    蛋白质基础组成结构
    数据库并发问题及四种隔离级别
  • 原文地址:https://blog.csdn.net/he_xin2009/article/details/126503112