devbox
爱喝兽奶帝天荒
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

数据采集工具之Canal

作者：爱喝兽奶帝天荒 | 2024-08-09 17:54:15

赞

踩

数据采集工具之Canal

本文主要介绍canal采集mysql数据的tcp、datahub(kafka)模式如何实现

1、下载canal

https://aliyun-datahub.oss-cn-hangzhou.aliyuncs.com/tools/canal.deployer-1.1.5-SNAPSHOT.tar.gz

canal的原理类似于mysql的主从复制，canal模拟的是从节点拉取主节点的binlog数据 

2、TCP模式的实现

a、canal.properties

打开看看即可，不需要调整

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

 b、example/instance.properties

canal.instance.master.address=192.168.140.1:3306  ###修改为自己的mysql信息

  1. #################################################
  2. ## mysql serverId , v1.0.26+ will autoGen
  3. # canal.instance.mysql.slaveId=0
  4.  
  5. # enable gtid use true/false
  6. canal.instance.gtidon=false
  7.  
  8. # position info
  9. canal.instance.master.address=192.168.140.1:3306  ###修改为自己的mysql信息
  10. canal.instance.master.journal.name=
  11. canal.instance.master.position=
  12. canal.instance.master.timestamp=
  13. canal.instance.master.gtid=
  14.  
  15. # rds oss binlog
  16. canal.instance.rds.accesskey=
  17. canal.instance.rds.secretkey=
  18. canal.instance.rds.instanceId=
  19.  
  20. # table meta tsdb info
  21. canal.instance.tsdb.enable=true
  22. #canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb
  23. #canal.instance.tsdb.dbUsername=canal
  24. #canal.instance.tsdb.dbPassword=canal
  25.  
  26. #canal.instance.standby.address =
  27. #canal.instance.standby.journal.name =
  28. #canal.instance.standby.position =
  29. #canal.instance.standby.timestamp =
  30. #canal.instance.standby.gtid=
  31.  
  32. # username/password
  33. canal.instance.dbUsername=flink
  34. canal.instance.dbPassword=flink
  35. canal.instance.connectionCharset = UTF-8
  36. # enable druid Decrypt database password
  37. canal.instance.enableDruid=false
  38. #canal.instance.pwdPublicKey=MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBALK4BUxdDltRRE5/zXpVEVPUgunvscYFtEip3pmLlhrWpacX7y7GCMo2/JM6LeHmiiNdH1FWgGCpUfircSwlWKUCAwEAAQ==
  39.  
  40. # table regex
  41. canal.instance.filter.regex=.*\\..*
  42. # table black regex
  43. canal.instance.filter.black.regex=
  44. # table field filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2)
  45. #canal.instance.filter.field=test1.t_product:id/subject/keywords,test2.t_company:id/name/contact/ch
  46. # table field black filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2)
  47. #canal.instance.filter.black.field=test1.t_product:subject/product_image,test2.t_company:id/name/contact/ch
  48.  
  49. # mq config
  50. canal.mq.topic=example
  51. # dynamic topic route by schema or table regex
  52. #canal.mq.dynamicTopic=mytest1.user,mytest2\\..*,.*\\..*
  53. canal.mq.partition=0
  54. # hash partition config
  55. #canal.mq.partitionsNum=3
  56. #canal.mq.partitionHash=test.table:id^name,.*\\..*
  57. #################################################

启动：bin/startup.sh

jps:

到此，canal服务端配置完成 

 c、canal客户端开发

依赖

  1. <dependency>
  2.       <groupId>com.alibaba.otter</groupId>
  3.       <artifactId>canal.client</artifactId>
  4.       <version>1.1.2</version>
  5.     </dependency>

开发代码：

  1. package com.tbea;
  2.  
  3. import com.alibaba.fastjson.JSONObject;
  4. import com.alibaba.otter.canal.client.CanalConnector;
  5. import com.alibaba.otter.canal.client.CanalConnectors;
  6. import com.alibaba.otter.canal.protocol.CanalEntry;
  7. import com.alibaba.otter.canal.protocol.Message;
  8. import com.google.protobuf.ByteString;
  9. import com.google.protobuf.InvalidProtocolBufferException;
  10.  
  11. import java.net.InetSocketAddress;
  12. import java.util.List;
  13.  
  14. public class CanalClient {
  15.     public static void main(String[] args) throws InterruptedException, InvalidProtocolBufferException {
  16.         //获取链接
  17.         CanalConnector canalConnector = CanalConnectors.newSingleConnector(
  18.                 new InetSocketAddress("192.168.140.129", 11111)
  19.                 , "example"
  20.                 , "", ""
  21.         );
  22.  
  23.         //尝试获取新数据
  24.         while (true){
  25.             //todo 连接
  26.             canalConnector.connect();
  27.             //todo 订阅数据库
  28.             canalConnector.subscribe("flinkcdc.*");
  29.             //todo 批量拉取数据
  30.             Message message = canalConnector.get(100);
  31.             //todo 获取entry
  32.             List<CanalEntry.Entry> entries = message.getEntries();
  33.             //todo 遍历 判断集合状态
  34.             if (entries.size()<=0){
  35.                 System.out.println("当次未抓取到数据，休息一会~~~~~~~");
  36.                 Thread.sleep(1000);
  37.             }else {
  38.                 //遍历解析
  39.                 for (CanalEntry.Entry entry:entries){
  40.                     //1.获取表名
  41.                     String tableName = entry.getHeader().getTableName();
  42.                     //2.获取类型
  43.                     CanalEntry.EntryType entryType = entry.getEntryType();
  44.                     //3.获取序列化数据
  45.                     ByteString storeValue = entry.getStoreValue();
  46.                     //4.判断entry是否rowdata类型
  47.                     if (CanalEntry.EntryType.ROWDATA.equals(entryType)){
  48.                         //5.反序列化数据
  49.                         CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(storeValue);
  50.                         //6.获取当前事件操作类型
  51.                         CanalEntry.EventType eventType = rowChange.getEventType();
  52.                         //7.获取数据集
  53.                         List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();
  54.  
  55.                         //8.遍历rowdatalist，并打印数据集
  56.                         for (CanalEntry.RowData rowData:rowDatasList){
  57.                             JSONObject beforeData = new JSONObject();
  58.                             JSONObject afterData = new JSONObject();
  59.                             List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList();
  60.                             List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
  61.                             for (CanalEntry.Column column:beforeColumnsList){
  62.                                 beforeData.put(column.getName(),column.getValue());
  63.                             }
  64.                             for (CanalEntry.Column column:afterColumnsList){
  65.                                 afterData.put(column.getName(),column.getValue());
  66.                             }
  67.                             System.out.println("Table:"+tableName+
  68.                                     ",EventType:"+eventType+
  69.                                     ",Before:"+beforeData+
  70.                                     ",After:"+afterData);
  71.                         }
  72.                     }else {
  73.                         System.out.println("数据类型不是所需要的");
  74.                     }
  75.                 }
  76.             }
  77.  
  78.  
  79.         }
  80.  
  81.     }
  82. }

执行：

到此，我们可以实时获取到mysql数据库的各种操作日志，接下来需要将数据写到哪里 可以按需实现。

3、kafka模式的实现

a.canal.properties

修改：canal.serverMode = kafka

    kafka信息：kafka.bootstrap.servers = 192.168.140.128:9092

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

b、example/instance.properties

修改：canal.instance.master.address=192.168.140.1:3306

canal.instance.dbUsername=flink

canal.instance.dbPassword=flink

canal.mq.topic=mysql_binlogs

  1. #################################################
  2. ## mysql serverId , v1.0.26+ will autoGen
  3. # canal.instance.mysql.slaveId=0
  4.  
  5. # enable gtid use true/false
  6. canal.instance.gtidon=false
  7.  
  8. # position info
  9. canal.instance.master.address=192.168.140.1:3306
  10. canal.instance.master.journal.name=
  11. canal.instance.master.position=
  12. canal.instance.master.timestamp=
  13. canal.instance.master.gtid=
  14.  
  15. # rds oss binlog
  16. canal.instance.rds.accesskey=
  17. canal.instance.rds.secretkey=
  18. canal.instance.rds.instanceId=
  19.  
  20. # table meta tsdb info
  21. canal.instance.tsdb.enable=true
  22. #canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb
  23. #canal.instance.tsdb.dbUsername=canal
  24. #canal.instance.tsdb.dbPassword=canal
  25.  
  26. #canal.instance.standby.address =
  27. #canal.instance.standby.journal.name =
  28. #canal.instance.standby.position =
  29. #canal.instance.standby.timestamp =
  30. #canal.instance.standby.gtid=
  31.  
  32. # username/password
  33. canal.instance.dbUsername=flink
  34. canal.instance.dbPassword=flink
  35. canal.instance.connectionCharset = UTF-8
  36. # enable druid Decrypt database password
  37. canal.instance.enableDruid=false
  38. #canal.instance.pwdPublicKey=MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBALK4BUxdDltRRE5/zXpVEVPUgunvscYFtEip3pmLlhrWpacX7y7GCMo2/JM6LeHmiiNdH1FWgGCpUfircSwlWKUCAwEAAQ==
  39.  
  40. # table regex
  41. canal.instance.filter.regex=.*\\..*
  42. # table black regex
  43. canal.instance.filter.black.regex=
  44. # table field filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2)
  45. #canal.instance.filter.field=test1.t_product:id/subject/keywords,test2.t_company:id/name/contact/ch
  46. # table field black filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2)
  47. #canal.instance.filter.black.field=test1.t_product:subject/product_image,test2.t_company:id/name/contact/ch
  48.  
  49. # mq config
  50. canal.mq.topic=mysql_binlogs
  51. # dynamic topic route by schema or table regex
  52. #canal.mq.dynamicTopic=mytest1.user,mytest2\\..*,.*\\..*
  53. canal.mq.partition=0
  54. # hash partition config
  55. #canal.mq.partitionsNum=3
  56. #canal.mq.partitionHash=test.table:id^name,.*\\..*
  57. #################################################

c、启动

startup.sh

d、查看数据生产情况

kafka-console-consumer.sh --bootstrap-server 192.168.140.128:9092 --topic mysql_binlogs --from-beginning

4、datahub兼容kafka的实现

配置如何实现啊

什么是Canal插件,如何使用Canal插件_数据总线 DataHub(DataHub)-阿里云帮助中心

看样子需要datahub兼容kafka的ip:port （私有云需要联系运维）

5、TCP模式+Datahub SDK实现

TCP配置及基本实现参考2

a、依赖

  1.     <dependency>
  2.       <groupId>com.aliyun.datahub</groupId>
  3.       <artifactId>aliyun-sdk-datahub</artifactId>
  4.       <version>2.25.1</version>
  5.     </dependency>

b、代码实现

  1. package com.tbea;
  2.  
  3. import com.alibaba.fastjson.JSONObject;
  4. import com.alibaba.otter.canal.client.CanalConnector;
  5. import com.alibaba.otter.canal.client.CanalConnectors;
  6. import com.alibaba.otter.canal.protocol.CanalEntry;
  7. import com.alibaba.otter.canal.protocol.Message;
  8. import com.aliyun.datahub.client.DatahubClient;
  9. import com.aliyun.datahub.client.DatahubClientBuilder;
  10. import com.aliyun.datahub.client.auth.AliyunAccount;
  11. import com.aliyun.datahub.client.common.DatahubConfig;
  12. import com.aliyun.datahub.client.example.examples.Constant;
  13. import com.aliyun.datahub.client.exception.*;
  14. import com.aliyun.datahub.client.http.HttpConfig;
  15. import com.aliyun.datahub.client.model.*;
  16. import com.google.protobuf.ByteString;
  17. import com.google.protobuf.InvalidProtocolBufferException;
  18.  
  19. import java.net.InetSocketAddress;
  20. import java.util.ArrayList;
  21. import java.util.List;
  22.  
  23. import static com.aliyun.datahub.client.example.examples.Constant.*;
  24.  
  25. public class CanalClient {
  26.  
  27.     static DatahubClient datahubClient = DatahubClientBuilder.newBuilder().setDatahubConfig(
  28.                 //Protocol可不设置，不设置默认使用PROTOBUF传输协议
  29.                 new DatahubConfig("https://datahub.cn-beijing-tbdg-d01.dh.res.bigdata.tbea.com",
  30.                         new AliyunAccount("2Z8tAOpDPBm5LEkA", "Tlupsw2G0PdKGCRyPLucHjeESqoCla"))
  31.         ).setHttpConfig(new HttpConfig().setCompressType(CompressType.LZ4)).build();
  32.  
  33.     public static void main(String[] args) throws InterruptedException, InvalidProtocolBufferException {
  34.         //获取链接
  35.         CanalConnector canalConnector = CanalConnectors.newSingleConnector(
  36.                 new InetSocketAddress("192.168.140.129", 11111)
  37.                 , "example"
  38.                 , "", ""
  39.         );
  40.  
  41.         //尝试获取新数据
  42.         while (true){
  43.             //todo 连接
  44.             canalConnector.connect();
  45.             //todo 订阅数据库
  46.             canalConnector.subscribe("flinkcdc.*");
  47.             //todo 批量拉取数据
  48.             Message message = canalConnector.get(100);
  49.             //todo 获取entry
  50.             List<CanalEntry.Entry> entries = message.getEntries();
  51.             //todo 遍历 判断集合状态
  52.             if (entries.size()<=0){
  53.                 System.out.println("当次未抓取到数据，休息一会~~~~~~~");
  54.                 Thread.sleep(1000);
  55.             }else {
  56.                 //遍历解析
  57.                 for (CanalEntry.Entry entry:entries){
  58.                     //1.获取表名
  59.                     String tableName = entry.getHeader().getTableName();
  60.                     String schemaName = entry.getHeader().getSchemaName();
  61.                     //2.获取类型
  62.                     CanalEntry.EntryType entryType = entry.getEntryType();
  63.                     //3.获取序列化数据
  64.                     ByteString storeValue = entry.getStoreValue();
  65.                     //4.判断entry是否rowdata类型
  66.                     if (CanalEntry.EntryType.ROWDATA.equals(entryType)){
  67.                         //5.反序列化数据
  68.                         CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(storeValue);
  69.                         //6.获取当前事件操作类型
  70.                         CanalEntry.EventType eventType = rowChange.getEventType();
  71.                         //7.获取数据集
  72.                         List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();
  73.                         JSONObject beforeData = new JSONObject();
  74.                         JSONObject afterData = new JSONObject();
  75.                         //8.遍历rowdatalist，并打印数据集
  76.                         for (CanalEntry.RowData rowData:rowDatasList){
  77.  
  78.                             List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList();
  79.                             List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
  80.                             for (CanalEntry.Column column:beforeColumnsList){
  81.                                 beforeData.put(column.getName(),column.getValue());
  82.                             }
  83.                             for (CanalEntry.Column column:afterColumnsList){
  84.                                 afterData.put(column.getName(),column.getValue());
  85.                             }
  86.                             System.out.println("Table:"+tableName+
  87.                                     ",EventType:"+eventType+
  88.                                     ",Before:"+beforeData+
  89.                                     ",After:"+afterData);
  90.  
  91.                         }
  92.                         Binlog binlog = new Binlog(eventType.toString(), tableName, schemaName, beforeData.toJSONString(), afterData.toJSONString());
  93.                         tupleExample("bigdata","tcp_canal",3,binlog);
  94.                     }else {
  95.                         System.out.println("数据类型不是所需要的");
  96.                     }
  97.                 }
  98.             }
  99.         }
  100.     }
  101.     // 写入Tuple型数据
  102.     public static void tupleExample(String project,String topic,int retryTimes,Binlog binlog) {
  103.         DatahubClient datahubClient = DatahubClientBuilder.newBuilder().setDatahubConfig(
  104.                 //Protocol可不设置，不设置默认使用PROTOBUF传输协议
  105.                 new DatahubConfig("https://datahub.cn-*****************.com",
  106.                         new AliyunAccount("************", "*************"))
  107.         ).setHttpConfig(new HttpConfig().setCompressType(CompressType.LZ4)).build();
  108.         // 获取schema
  109.         RecordSchema recordSchema = datahubClient.getTopic(project,topic).getRecordSchema();
  110.         // 生成十条数据
  111.         List<RecordEntry> recordEntries = new ArrayList<>();
  112.  
  113.         RecordEntry recordEntry = new RecordEntry();
  114.         // 对每条数据设置额外属性，例如ip 机器名等。可以不设置额外属性，不影响数据写入
  115.         recordEntry.addAttribute("key1", "value1");
  116.         TupleRecordData data = new TupleRecordData(recordSchema);
  117.         data.setField("operate", binlog.getOperater());
  118.         data.setField("tablename", binlog.getTableName());
  119.         data.setField("databasename", binlog.getDatabaseName());
  120.         data.setField("before", binlog.getBefore());
  121.         data.setField("after", binlog.getAfter());
  122.         recordEntry.setRecordData(data);
  123.         recordEntries.add(recordEntry);
  124.  
  125.         try {
  126.             PutRecordsResult result = datahubClient.putRecords(project, topic, recordEntries);
  127.             int i = result.getFailedRecordCount();
  128.             if (i > 0) {
  129.                 retry(datahubClient, result.getFailedRecords(), retryTimes, project, topic);
  130.             }
  131.         }  catch (DatahubClientException e) {
  132.             System.out.println("requestId:" + e.getRequestId() + "\tmessage:" + e.getErrorMessage());
  133.         }
  134.     }
  135.     //重试机制
  136.     public static void retry(DatahubClient client, List<RecordEntry> records, int retryTimes, String project, String topic) {
  137.         boolean suc = false;
  138.         while (retryTimes != 0) {
  139.             retryTimes = retryTimes - 1;
  140.             PutRecordsResult recordsResult = client.putRecords(project, topic, records);
  141.             if (recordsResult.getFailedRecordCount() > 0) {
  142.                 retry(client,recordsResult.getFailedRecords(),retryTimes,project,topic);
  143.             }
  144.             suc = true;
  145.             break;
  146.         }
  147.         if (!suc) {
  148.             System.out.println("retryFailure");
  149.         }
  150.     }
  151. }

注意topic和代码的映射关系

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/爱喝兽奶帝天荒/article/detail/954474
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号