canal+kafka实践——实时etl

忘是亡心i 2022-05-14 07:21 576阅读 0赞

canal解析sql数据库的binlog并格式化数据，然后同步到kafka消息，可以用来实现实时etl

yml：

  1. spring:
  2.   application:
  3.     name: canal
  4.   canal:
  5.     topic-prefix: etl_timely.
  6.     destination:
  7.       example: 0
  8.     username:
  9.     password:
  10.     dealy-limit: 2000
  11.   kafka:
  12.     bootstrap-servers: 127.0.0.1:9092
  13.     producer:
  14.       acks: 1
  15.       batch-size: 100
  16.       client-id: canal
  17.       retries: 3
  18.       key-serializer: org.apache.kafka.common.serialization.StringSerializer
  19.       value-serializer: org.apache.kafka.common.serialization.StringSerializer
  20.       bootstrap-servers: 127.0.0.1:9092
  21.     consumer:
  22.       enable-auto-commit: true
  23.       group-id: etl
  24.       bootstrap-servers: 127.0.0.1:9092
  25.     template:
  26.       default-topic: etl_canal

pom：

  1. <dependency>
  2.             <groupId>org.springframework.boot</groupId>
  3.             <artifactId>spring-boot-starter-web</artifactId>
  4.         </dependency>
  6.         <dependency>
  7.             <groupId>org.springframework.boot</groupId>
  8.             <artifactId>spring-boot-starter-test</artifactId>
  9.             <scope>test</scope>
  10.         </dependency>
  12.         <!-- https://mvnrepository.com/artifact/com.alibaba.otter/canal.common -->
  13.         <dependency>
  14.             <groupId>com.alibaba.otter</groupId>
  15.             <artifactId>canal.client</artifactId>
  16.             <version>1.0.25</version>
  17.         </dependency>
  18.         <dependency>
  19.             <groupId>org.bouncycastle</groupId>
  20.             <artifactId>bcprov-jdk15on</artifactId>
  21.             <version>RELEASE</version>
  22.             <scope>compile</scope>
  23.         </dependency>
  25.         <dependency>
  26.             <groupId>org.springframework.boot</groupId>
  27.             <artifactId>spring-boot-configuration-processor</artifactId>
  28.             <optional>true</optional>
  29.         </dependency>
  32.         <!--kafka-->
  33.         <dependency>
  34.             <groupId>org.apache.kafka</groupId>
  35.             <artifactId>kafka_2.11</artifactId>
  36.         </dependency>
  37.         <dependency>
  38.             <groupId>org.apache.kafka</groupId>
  39.             <artifactId>kafka-clients</artifactId>
  40.         </dependency>
  42.         <dependency>
  43.             <groupId>org.springframework.kafka</groupId>
  44.             <artifactId>spring-kafka</artifactId>
  45.         </dependency>
  46.         <dependency>
  47.             <groupId>org.bouncycastle</groupId>
  48.             <artifactId>bcprov-jdk15on</artifactId>
  49.             <version>RELEASE</version>
  50.             <scope>compile</scope>
  51.         </dependency>
  53.         <!-- https://mvnrepository.com/artifact/org.projectlombok/lombok -->
  54.         <dependency>
  55.             <groupId>org.projectlombok</groupId>
  56.             <artifactId>lombok</artifactId>
  57.             <version>1.18.0</version>
  58.             <scope>provided</scope>
  59.         </dependency>

CanalClient：

  1. package com.kexin.canal.client;
  3. import com.alibaba.otter.canal.client.CanalConnector;
  4. import com.alibaba.otter.canal.client.CanalConnectors;
  5. import com.alibaba.otter.canal.common.utils.AddressUtils;
  6. import com.alibaba.otter.canal.protocol.Message;
  7. import com.kexin.canal.config.DisruptorConfig;
  8. import com.kexin.canal.service.KafkaService;
  9. import com.lmax.disruptor.BlockingWaitStrategy;
  10. import com.lmax.disruptor.EventFactory;
  11. import com.lmax.disruptor.RingBuffer;
  12. import com.lmax.disruptor.dsl.Disruptor;
  13. import com.lmax.disruptor.dsl.ProducerType;
  14. import lombok.Getter;
  15. import lombok.Setter;
  16. import org.springframework.beans.factory.annotation.Autowired;
  17. import org.springframework.boot.CommandLineRunner;
  18. import org.springframework.boot.context.properties.ConfigurationProperties;
  19. import org.springframework.stereotype.Component;
  21. import java.net.InetSocketAddress;
  22. import java.util.Map;
  23. import java.util.concurrent.Executors;
  24. import java.util.concurrent.ThreadFactory;
  26. /**
  27.  * @Author KeXin
  28.  * @Date 2018/7/20 下午5:16
  29.  **/
  30. @Component
  31. @ConfigurationProperties(prefix = "spring.canal")
  32. public class CanalClient implements CommandLineRunner{
  34.     @Getter
  35.     @Setter
  36.     private String username;
  38.     @Getter
  39.     @Setter
  40.     private String password;
  42.     @Getter
  43.     @Setter
  44.     private Map<String, String> destination;
  46.     @Autowired
  47.     KafkaService kafkaService;
  48.     @Override
  49.     public void run(String... args) {
  50.         int port = 11111;
  51.         startClient(port,destination,username,password);
  52.     }
  54.     /**
  55.      * 监测数据库变化，使用disruptor处理消息
  56.      */
  57.     public void startClient(int port,Map<String, String> canalDestination,String canalUsername,String canalPassword){
  58.         canalDestination.forEach((database,partition)->{
  59.             //启动不加锁队列
  60.             RingBuffer<DisruptorConfig.Element> buffer = getDisruptor().start();
  61.             new Thread(()->{
  62.                 // 创建链接
  63.                 CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(AddressUtils.getHostIp(),
  64.                         port), database, canalUsername, canalPassword);
  65.                 int batchSize = 100;
  66.                 try {
  67.                     connector.connect();
  68.                     connector.subscribe();
  69.                     connector.rollback();
  70.                     while (true) {
  71.                         Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
  72.                         long batchId = message.getId();
  73.                         int size = message.getEntries().size();
  74.                         if (batchId == -1 || size == 0) {
  75.                             try {
  76.                                 Thread.sleep(1000);
  77.                             } catch (InterruptedException e) {
  78.                                 e.printStackTrace();
  79.                             }
  80.                         } else {
  81.                               //发送至kafka
  82.                             kafkaService.sendMessage(message.getEntries());
  83.                         }
  85.                         connector.ack(batchId); // 提交确认
  86.                         // connector.rollback(batchId); // 处理失败, 回滚数据
  87.                     }
  89.                 } finally {
  90.                     connector.disconnect();
  91.                 }
  92.             }).start();
  93.     });
  95. }

解析binlog为自己需要的格式：

  1. package com.kexin.canal.service;
  3. import com.alibaba.fastjson.JSON;
  4. import com.alibaba.otter.canal.protocol.CanalEntry;
  5. import org.springframework.beans.factory.annotation.Autowired;
  6. import org.springframework.beans.factory.annotation.Value;
  7. import org.springframework.kafka.core.KafkaTemplate;
  8. import org.springframework.stereotype.Component;
  10. import java.util.HashMap;
  11. import java.util.List;
  12. import java.util.Map;
  14. /**
  15.  * @Author KeXin
  16.  * @Date 2018/7/20 下午4:56
  17.  **/
  18. @Component
  19. public class KafkaService {
  20.     @Autowired
  21.     KafkaTemplate kafkaTemplate;
  23.     @Value("${spring.canal.topic-prefix}")
  24.     private String canalTopicPrefix;
  26.     /**
  27.      * 发送消息
  28.      * @param entrys
  29.      */
  30.     public void sendMessage(List<CanalEntry.Entry> entrys) {
  31.         for (CanalEntry.Entry entry : entrys) {
  32.             if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
  33.                 continue;
  34.             }
  36.             CanalEntry.RowChange rowChange = null;
  37.             try {
  38.                 rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
  39.             } catch (Exception e) {
  40.                 throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
  41.                         e);
  42.             }
  44.             CanalEntry.EventType eventType = rowChange.getEventType();
  45.             String tableName = entry.getHeader().getTableName();
  46.             String schemaName = entry.getHeader().getSchemaName();
  47.             long executeTime = entry.getHeader().getExecuteTime();
  49.             //根据binlog的filename和position来定位
  50.             System.out.println(String.format("================> binlog[%s:%s] , name[%s,%s] , eventType : %s",
  51.                     entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
  52.                     entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
  53.                     eventType));
  55.             for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
  57.                 Map<String, Object> map = new HashMap<>();
  59.                 map.put("event_timestamp", executeTime);
  60.                 map.put("table_name", tableName);
  61.                 map.put("database_name", schemaName);
  62.                 Map<String, Object> map_info = new HashMap<>();
  64.                 if (eventType == CanalEntry.EventType.DELETE) {
  65.                     map.put("event_op_type", "delete");
  66.                     for(CanalEntry.Column column : rowData.getBeforeColumnsList()){
  67.                         if(column.getValue()!=null&&!column.getValue().equals(""))
  68.                             map_info.put(column.getName(), column.getValue());
  69.                     }
  70.                 } else if(eventType == CanalEntry.EventType.INSERT){
  71.                     map.put("event_op_type", "insert");
  72.                     for(CanalEntry.Column column : rowData.getAfterColumnsList()){
  73.                             map_info.put(column.getName(), column.getValue());
  74.                     }
  75.                 }else {
  76.                     map.put("event_op_type", "update");
  77.                     for(CanalEntry.Column column : rowData.getAfterColumnsList()){
  78.                         map_info.put(column.getName(), column.getValue());
  79.                     }
  81.                     Map<String, Object> beforeMap = new HashMap<>();
  83.                     for(CanalEntry.Column column : rowData.getBeforeColumnsList()){
  84.                         if(column.getValue()!=null&&!column.getValue().equals(""))
  85.                             beforeMap.put(column.getName(), column.getValue());
  86.                     }
  87.                     map.put("beforeColumns", beforeMap);
  88.                 }
  89.                 map.put("map_info",map_info);
  90.                 System.out.println(map);
  91.                 kafkaTemplate.send( canalTopicPrefix + tableName, JSON.toJSONString(map));
  93.             }
  94.         }
  95.     }
  97. }

发表评论

表情：
评论列表 （有 0 条评论，576人围观）

还没有评论，来说两句吧...

相关阅读

    相关 Flink：实时ETL案例

    抽取（extract）、转换（transform）、加载（load） 需求 将国家编号变成地区编号（在跨国业务常用） 形式 数据源 \{"dt":"2020-

    男娘i男娘i/ 0 赞/ 108 阅读

    相关 ETL

    做数据仓库系统，ETL是关键的一环。说大了，ETL是数据整合解决方案，说小了，就是倒数据的工具。回忆一下工作这么些年来，处理数据迁移、转换的工作倒还真的不少。但是那些工作基本上

    忘是亡心i忘是亡心i/ 0 赞/ 234 阅读