devbox
很楠不爱3
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

FlinkCDC简介_flink cdc

作者：很楠不爱3 | 2024-04-07 22:31:29

赞

踩

flink cdc

目录

一. Flink CDC介绍

二.Flink CDC 实操

2.1 MySQL配置

2.2 pom文件

2.3 Java代码

2.4 测试结果

一. Flink CDC介绍

CDC主要分为基于查询和基于Binlog两种方式，我们主要了解一下这两种之间的区别：

在这里插入图片描述
FlinkCDC其实和canal差不多，只不过就是flink社区开发的组件，用起来更方便一些。

  Flink在1.11版本中新增了CDC的特性，简称 改变数据捕获。名称来看有点乱，我们先从之前的数据架构来看CDC的内容。

  以上是之前的mysql binlog日志处理流程,例如canal监听binlog把日志写入到kafka中。而Apache Flink实时消费Kakfa的数据实现mysql数据的同步或其他内容等。拆分来说整体上可以分为以下几个阶段。

  1. mysql开启binlog
  2. canal同步binlog数据写入到kafka
  3. flink读取kakfa中的binlog数据进行相关的业务处理。

整体的处理链路较长，需要用到的组件也比较多。Apache Flink CDC可以直接从数据库获取到binlog供下游进行业务计算分析。简单来说链路会变成这样

也就是说数据不再通过canal与kafka进行同步，而flink直接进行处理mysql的数据。节省了canal与kafka的过程。

Flink 1.11中实现了mysql-cdc与postgre-CDC，也就是说在Flink 1.11中我们可以直接通过Flink来直接消费mysql,postgresql的数据进行业务的处理。

使用场景:

  1. 数据库数据的增量同步
  2. 数据库表之上的物理化视图
  3. 维表join
  4. 其他业务处理

二.Flink CDC 实操

2.1 MySQL配置

MySQL必须开启binlog
MySQL表必须有主键

  1. mysql> show variables like '%log_bin%';
  2. +---------------------------------+---------------------------------------------+
  3. | Variable_name                   | Value                                       |
  4. +---------------------------------+---------------------------------------------+
  5. | log_bin                         | ON                                          |
  6. | log_bin_basename                | /home/mysql/data/3306/10-31-1-122-bin       |
  7. | log_bin_index                   | /home/mysql/data/3306/10-31-1-122-bin.index |
  8. | log_bin_trust_function_creators | OFF                                         |
  9. | log_bin_use_v1_row_events       | OFF                                         |
  10. | sql_log_bin                     | ON                                          |
  11. +---------------------------------+---------------------------------------------+
  12. 6 rows in set (0.01 sec)

MySQL代码:

  1. create databases cdc_test;
  2.  
  3. create table test1(id int primary key,name varchar(50),create_datetime timestamp(0));
  4.  
  5. insert into test1(id,name,create_datetime) values (1,'abc',current_timestamp());
  6. insert into test1(id,name,create_datetime) values (2,'def',current_timestamp());
  7. insert into test1(id,name,create_datetime) values (3,'ghi',current_timestamp()); 
  8.  
  9. update test1 set name = 'aaa' where id = 1;
  10. delete from test1 where id = 1;
  11.  
  12. create table test2(id int primary key,name varchar(50),create_datetime timestamp(0)); 
  13. delete from test1 where id = 1;
  14. insert into test2(id,name,create_datetime) values (1,'abc',current_timestamp()); 
  15.  
  16. drop table test2;

2.2 pom文件

pom文件配置如下:

  1. <dependencies>
  2.     <dependency>
  3.       <groupId>junit</groupId>
  4.       <artifactId>junit</artifactId>
  5.       <version>4.11</version>
  6.       <scope>test</scope>
  7.     </dependency>
  8.     <dependency>
  9.       <groupId>com.alibaba.ververica</groupId>
  10.       <artifactId>flink-connector-mysql-cdc</artifactId>
  11.       <version>1.1.1</version>
  12.     </dependency>
  13.     <dependency>
  14.       <groupId>com.alibaba</groupId>
  15.       <artifactId>fastjson</artifactId>
  16.       <version>1.2.75</version>
  17.     </dependency>
  18.     <dependency>
  19.       <groupId>org.apache.flink</groupId>
  20.       <artifactId>flink-streaming-java_2.12</artifactId>
  21.       <version>1.12.0</version>
  22.     </dependency>
  23.     <dependency>
  24.       <groupId>org.apache.flink</groupId>
  25.       <artifactId>flink-clients_2.12</artifactId>
  26.       <version>1.12.0</version>
  27.     </dependency>
  28.     <dependency>
  29.       <groupId>org.apache.flink</groupId>
  30.       <artifactId>flink-java</artifactId>
  31.       <version>1.12.0</version>
  32.     </dependency>
  33.     <dependency>
  34.       <groupId>org.apache.flink</groupId>
  35.       <artifactId>flink-table-planner-blink_2.12</artifactId>
  36.       <version>1.12.0</version>
  37.       <type>test-jar</type>
  38.     </dependency>
  39.   </dependencies>

2.3 Java代码

CdcDwdDeserializationSchema

  1. package com.zqs.study.flink.cdc;
  2.  
  3. import com.alibaba.fastjson.JSONArray;
  4. import com.alibaba.fastjson.JSONObject;
  5. import com.alibaba.ververica.cdc.debezium.DebeziumDeserializationSchema;
  6. import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
  7. import org.apache.flink.api.common.typeinfo.TypeInformation;
  8. import org.apache.flink.util.Collector;
  9. import org.apache.kafka.connect.data.Field;
  10. import org.apache.kafka.connect.data.Schema;
  11. import org.apache.kafka.connect.data.Struct;
  12. import org.apache.kafka.connect.source.SourceRecord;
  13.  
  14. import java.util.List;
  15.  
  16.  
  17. public class CdcDwdDeserializationSchema implements DebeziumDeserializationSchema<JSONObject> {
  18.     private static final long serialVersionUID = -3168848963265670603L;
  19.  
  20.     public CdcDwdDeserializationSchema() {
  21.     }
  22.  
  23.     @Override
  24.     public void deserialize(SourceRecord record, Collector<JSONObject> out) {
  25.         Struct dataRecord = (Struct) record.value();
  26.  
  27.         Struct afterStruct = dataRecord.getStruct("after");
  28.         Struct beforeStruct = dataRecord.getStruct("before");
  29.         /*
  30.           todo 1，同时存在 beforeStruct 跟 afterStruct数据的话，就代表是update的数据
  31.                2,只存在 beforeStruct 就是delete数据
  32.                3，只存在 afterStruct数据 就是insert数据
  33.          */
  34.  
  35.         JSONObject logJson = new JSONObject();
  36.  
  37.         String canal_type = "";
  38.         List<Field> fieldsList = null;
  39.         if (afterStruct != null && beforeStruct != null) {
  40.             System.out.println("这是修改数据");
  41.             canal_type = "update";
  42.             fieldsList = afterStruct.schema().fields();
  43.             //todo 字段与值
  44.             for (Field field : fieldsList) {
  45.                 String fieldName = field.name();
  46.                 Object fieldValue = afterStruct.get(fieldName);
  47. //            System.out.println("*****fieldName=" + fieldName+",fieldValue="+fieldValue);
  48.                 logJson.put(fieldName, fieldValue);
  49.             }
  50.         } else if (afterStruct != null) {
  51.             System.out.println("这是新增数据");
  52.  
  53.             canal_type = "insert";
  54.             fieldsList = afterStruct.schema().fields();
  55.             //todo 字段与值
  56.             for (Field field : fieldsList) {
  57.                 String fieldName = field.name();
  58.                 Object fieldValue = afterStruct.get(fieldName);
  59. //            System.out.println("*****fieldName=" + fieldName+",fieldValue="+fieldValue);
  60.                 logJson.put(fieldName, fieldValue);
  61.             }
  62.         } else if (beforeStruct != null) {
  63.             System.out.println("这是删除数据");
  64.             canal_type = "detele";
  65.             fieldsList = beforeStruct.schema().fields();
  66.             //todo 字段与值
  67.             for (Field field : fieldsList) {
  68.                 String fieldName = field.name();
  69.                 Object fieldValue = beforeStruct.get(fieldName);
  70. //            System.out.println("*****fieldName=" + fieldName+",fieldValue="+fieldValue);
  71.                 logJson.put(fieldName, fieldValue);
  72.             }
  73.         } else {
  74.             System.out.println("一脸蒙蔽了");
  75.         }
  76.  
  77.  
  78.         //todo 拿到databases table信息
  79.         Struct source = dataRecord.getStruct("source");
  80.         Object db = source.get("db");
  81.         Object table = source.get("table");
  82.         Object ts_ms = source.get("ts_ms");
  83.  
  84.         logJson.put("canal_database", db);
  85.         logJson.put("canal_database", table);
  86.         logJson.put("canal_ts", ts_ms);
  87.         logJson.put("canal_type", canal_type);
  88.  
  89.         //todo 拿到topic
  90.         String topic = record.topic();
  91.         System.out.println("topic = " + topic);
  92.  
  93.         //todo 主键字段
  94.         Struct pk = (Struct) record.key();
  95.         List<Field> pkFieldList = pk.schema().fields();
  96.         int partitionerNum = 0;
  97.         for (Field field : pkFieldList) {
  98.             Object pkValue = pk.get(field.name());
  99.             partitionerNum += pkValue.hashCode();
  100.  
  101.         }
  102.         int hash = Math.abs(partitionerNum) % 3;
  103.         logJson.put("pk_hashcode", hash);
  104.         out.collect(logJson);
  105.     }
  106.  
  107.     @Override
  108.     public TypeInformation<JSONObject> getProducedType() {
  109.         return BasicTypeInfo.of(JSONObject.class);
  110.     }
  111. }

FlinkCDCSQLTest

  1. package com.zqs.study.flink.cdc;
  2. /**
  3.  * @author 只是甲
  4.  * @date   2021-09-30
  5.  * @remark Flink CDC 测试
  6.  */
  7. import com.alibaba.fastjson.JSONObject;
  8. import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
  9. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  10. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  11. import org.apache.flink.streaming.api.functions.source.SourceFunction;
  12.  
  13.  
  14. public class FlinkCDCSQLTest {
  15.     public static void main(String[] args) {
  16.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  17.         env.setParallelism(1);
  18.         SourceFunction<JSONObject> sourceFunction = MySQLSource.<JSONObject>builder()
  19.                 .hostname("10.31.1.122")
  20.                 .port(3306)
  21.                 .databaseList("cdc_test") // monitor all tables under inventory database
  22.  
  23.                 .username("root")
  24.                 //.password("abc123")
  25.                 .password("Abc123456!")
  26.                 .deserializer(new CdcDwdDeserializationSchema()) // converts SourceRecord to String
  27.                 .build();
  28.  
  29.  
  30.         DataStreamSource<JSONObject> stringDataStreamSource = env.addSource(sourceFunction);
  31.  
  32.         stringDataStreamSource.print("===>");
  33.  
  34.  
  35.         try {
  36.             env.execute("测试mysql-cdc");
  37.         } catch (Exception e) {
  38.             e.printStackTrace();
  39.         }
  40.  
  41.     }
  42. }

2.4 测试结果

如下截图所示，可以捕捉到DML语句，但是无法捕捉到DDL语句





 

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

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

  

闽ICP备14008679号