devbox
小丑西瓜9
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

实战Java springboot 采用Flink CDC操作SQL Server数据库获取增量变更数据_springboot flinkcdc

作者：小丑西瓜9 | 2024-03-06 16:51:34

赞

踩

springboot flinkcdc

目录

前言：

1、springboot引入依赖：

2、yml配置文件

3、创建SQL server CDC变更数据监听器

4、反序列化数据,转为变更JSON对象

5、CDC 数据实体类

6、自定义ApplicationContextUtil

7、自定义sink 交由spring管理，处理变更数据

前言：

        我的场景是从SQL Server数据库获取指定表的增量数据，查询了很多获取增量数据的方案，最终选择了Flink的 flink-connector-sqlserver-cdc ，这个需要用到SQL Server 的CDC（变更数据捕获），通过CDC来获取增量数据，处理数据前需要对数据库进行配置，如果不清楚如何配置可以看看我这篇文章：《SQL Server数据库开启CDC变更数据捕获操作指引》

废话不多说，直接上干货，如有不足还请指正

1、springboot引入依赖：

  1.     <properties>
  2.         <flink.version>1.16.0</flink.version>
  3.     </properties>
  4.     <dependencies>
  5.         <dependency>
  6.             <groupId>com.microsoft.sqlserver</groupId>
  7.             <artifactId>mssql-jdbc</artifactId>
  8.             <version>9.4.0.jre8</version>
  9.         </dependency>        
  10.         <dependency>
  11.             <groupId>org.projectlombok</groupId>
  12.             <artifactId>lombok</artifactId>
  13.             <version>1.18.26</version>
  14.         </dependency>
  15.         <dependency>
  16.             <groupId>org.apache.flink</groupId>
  17.             <artifactId>flink-java</artifactId>
  18.             <version>${flink.version}</version>
  19.         </dependency>
  20.         <dependency>
  21.             <groupId>org.apache.flink</groupId>
  22.             <artifactId>flink-streaming-java</artifactId>
  23.             <version>${flink.version}</version>
  24.         </dependency>
  25.         <dependency>
  26.             <groupId>org.apache.flink</groupId>
  27.             <artifactId>flink-clients</artifactId>
  28.             <version>${flink.version}</version>
  29.         </dependency>
  30.         <dependency>
  31.             <groupId>com.ververica</groupId>
  32.             <artifactId>flink-connector-sqlserver-cdc</artifactId>
  33.             <version>2.3.0</version>
  34.         </dependency>
  35.         <dependency>
  36.             <groupId>org.apache.flink</groupId>
  37.             <artifactId>flink-connector-kafka</artifactId>
  38.             <version>${flink.version}</version>
  39.         </dependency>
  40.  
  41.  
  42.         <dependency>
  43.             <groupId>org.apache.flink</groupId>
  44.             <artifactId>flink-table-planner-blink_2.11</artifactId>
  45.             <version>1.13.6</version>
  46.         </dependency>
  47.     </dependencies>

2、yml配置文件

  1. spring:
  2.     datasource:
  3.     url: jdbc:sqlserver://127.0.0.1:1433;DatabaseName=HM_5001
  4.     username: sa
  5.     password: root
  6.     driver-class-name: com.microsoft.sqlserver.jdbc.SQLServerDriver
  7. # 实时同步SQL Server数据库配置
  8. CDC:
  9.   DataSource:
  10.     host: 127.0.0.1
  11.     port: 1433
  12.     database: HM_5001
  13.     tableList: dbo.t1,dbo.Tt2,dbo.t3,dbo.t4
  14.     username: sa
  15.     password: sa

3、创建SQL server CDC变更数据监听器

  1. import com.ververica.cdc.connectors.sqlserver.SqlServerSource;
  2. import com.ververica.cdc.connectors.sqlserver.table.StartupOptions;
  3. import com.ververica.cdc.debezium.DebeziumSourceFunction;
  4. import lombok.extern.slf4j.Slf4j;
  5. import org.apache.flink.streaming.api.datastream.DataStream;
  6. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  7. import org.springframework.beans.factory.annotation.Value;
  8. import org.springframework.boot.ApplicationArguments;
  9. import org.springframework.boot.ApplicationRunner;
  10. import org.springframework.stereotype.Component;
  11.  
  12. import java.io.Serializable;
  13.  
  14. /**
  15.  * SQL server CDC变更监听器
  16.  **/
  17. @Component
  18. @Slf4j
  19. public class SQLServerCDCListener implements ApplicationRunner, Serializable {
  20.  
  21.     /**
  22.      * CDC数据源配置
  23.      */
  24.     @Value("${CDC.DataSource.host}")
  25.     private String host;
  26.     @Value("${CDC.DataSource.port}")
  27.     private String port;
  28.     @Value("${CDC.DataSource.database}")
  29.     private String database;
  30.     @Value("${CDC.DataSource.tableList}")
  31.     private String tableList;
  32.     @Value("${CDC.DataSource.username}")
  33.     private String username;
  34.     @Value("${CDC.DataSource.password}")
  35.     private String password;
  36.  
  37.     private final DataChangeSink dataChangeSink;
  38.  
  39.     public SQLServerCDCListener(DataChangeSink dataChangeSink) {
  40.         this.dataChangeSink = dataChangeSink;
  41.     }
  42.  
  43.     @Override
  44.     public void run(ApplicationArguments args) throws Exception {
  45.         log.info("开始启动Flink CDC获取ERP变更数据......");
  46.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  47.         env.setParallelism(1);
  48.         DebeziumSourceFunction<DataChangeInfo> dataChangeInfoMySqlSource = buildDataChangeSource();
  49.         DataStream<DataChangeInfo> streamSource = env
  50.                 .addSource(dataChangeInfoMySqlSource, "SQLServer-source")
  51.                 .setParallelism(1);
  52.         streamSource.addSink(dataChangeSink);
  53.         env.execute("SQLServer-stream-cdc");
  54.     }
  55.  
  56.     /**
  57.      * 构造CDC数据源
  58.      */
  59.     private DebeziumSourceFunction<DataChangeInfo> buildDataChangeSource() {
  60.         String[] tables = tableList.replace(" ", "").split(",");
  61.         return SqlServerSource.<DataChangeInfo>builder()
  62.                 .hostname(host)
  63.                 .port(Integer.parseInt(port))
  64.                 .database(database) // monitor sqlserver database
  65.                 .tableList(tables) // monitor products table
  66.                 .username(username)
  67.                 .password(password)
  68.                 /*
  69.                  *initial初始化快照,即全量导入后增量导入(检测更新数据写入)
  70.                  * latest:只进行增量导入(不读取历史变化) 
  71.                  */
  72.                 .startupOptions(StartupOptions.latest())
  73.                 .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
  74.                 .build();
  75.     }
  76. }

4、反序列化数据,转为变更JSON对象

  1. import com.alibaba.fastjson.JSONObject;
  2. import com.ververica.cdc.debezium.DebeziumDeserializationSchema;
  3. import io.debezium.data.Envelope;
  4. import lombok.extern.slf4j.Slf4j;
  5. import org.apache.flink.api.common.typeinfo.TypeInformation;
  6. import org.apache.flink.util.Collector;
  7. import org.apache.kafka.connect.data.Field;
  8. import org.apache.kafka.connect.data.Schema;
  9. import org.apache.kafka.connect.data.Struct;
  10. import org.apache.kafka.connect.source.SourceRecord;
  11.  
  12. import java.time.Instant;
  13. import java.time.LocalDateTime;
  14. import java.time.ZoneId;
  15. import java.util.List;
  16. import java.util.Optional;
  17.  
  18. /**
  19.  * SQLServer消息读取自定义序列化
  20.  **/
  21. @Slf4j
  22. public class JsonDebeziumDeserializationSchema implements DebeziumDeserializationSchema<DataChangeInfo> {
  23.  
  24.     public static final String TS_MS = "ts_ms";
  25.     public static final String BEFORE = "before";
  26.     public static final String AFTER = "after";
  27.     public static final String SOURCE = "source";
  28.     public static final String CREATE = "CREATE";
  29.     public static final String UPDATE = "UPDATE";
  30.  
  31.     /**
  32.      *
  33.      * 反序列化数据,转为变更JSON对象
  34.      */
  35.     @Override
  36.     public void deserialize(SourceRecord sourceRecord, Collector<DataChangeInfo> collector) {
  37.         try {
  38.             String topic = sourceRecord.topic();
  39.             String[] fields = topic.split("\\.");
  40.             String database = fields[1];
  41.             String tableName = fields[2];
  42.             Struct struct = (Struct) sourceRecord.value();
  43.             final Struct source = struct.getStruct(SOURCE);
  44.             DataChangeInfo dataChangeInfo = new DataChangeInfo();
  45.             dataChangeInfo.setBeforeData(getJsonObject(struct, BEFORE).toJSONString());
  46.             dataChangeInfo.setAfterData(getJsonObject(struct, AFTER).toJSONString());
  47.             // 获取操作类型  CREATE UPDATE DELETE  1新增 2修改 3删除
  48.             Envelope.Operation operation = Envelope.operationFor(sourceRecord);
  49.             String type = operation.toString().toUpperCase();
  50.             int eventType = type.equals(CREATE) ? 1 : UPDATE.equals(type) ? 2 : 3;
  51.             dataChangeInfo.setEventType(eventType);
  52.             dataChangeInfo.setDatabase(database);
  53.             dataChangeInfo.setTableName(tableName);
  54.             ZoneId zone = ZoneId.systemDefault();
  55.             Long timestamp = Optional.ofNullable(struct.get(TS_MS)).map(x -> Long.parseLong(x.toString())).orElseGet(System::currentTimeMillis);
  56.             dataChangeInfo.setChangeTime(LocalDateTime.ofInstant(Instant.ofEpochMilli(timestamp), zone));
  57.             //7.输出数据
  58.             collector.collect(dataChangeInfo);
  59.         } catch (Exception e) {
  60.             log.error("SQLServer消息读取自定义序列化报错：{}", e.getMessage());
  61.             e.printStackTrace();
  62.         }
  63.     }
  64.  
  65.     /**
  66.      *
  67.      * 从源数据获取出变更之前或之后的数据
  68.      */
  69.     private JSONObject getJsonObject(Struct value, String fieldElement) {
  70.         Struct element = value.getStruct(fieldElement);
  71.         JSONObject jsonObject = new JSONObject();
  72.         if (element != null) {
  73.             Schema afterSchema = element.schema();
  74.             List<Field> fieldList = afterSchema.fields();
  75.             for (Field field : fieldList) {
  76.                 Object afterValue = element.get(field);
  77.                 jsonObject.put(field.name(), afterValue);
  78.             }
  79.         }
  80.         return jsonObject;
  81.     }
  82.  
  83.  
  84.  
  85.     @Override
  86.     public TypeInformation<DataChangeInfo> getProducedType() {
  87.         return TypeInformation.of(DataChangeInfo.class);
  88.     }
  89. }

5、CDC 数据实体类

  1. import lombok.Data;
  2. import java.io.Serializable;
  3. import java.time.LocalDateTime;
  4.  
  5. /**
  6.  * CDC 数据实体类
  7.  */
  8. @Data
  9. public class DataChangeInfo implements Serializable {
  10.  
  11.     /**
  12.      * 数据库名
  13.      */
  14.     private String database;
  15.     /**
  16.      * 表名
  17.      */
  18.     private String tableName;
  19.     /**
  20.      * 变更时间
  21.      */
  22.     private LocalDateTime changeTime;
  23.     /**
  24.      * 变更类型 1新增 2修改 3删除
  25.      */
  26.     private Integer eventType;
  27.     /**
  28.      * 变更前数据
  29.      */
  30.     private String beforeData;
  31.     /**
  32.      * 变更后数据
  33.      */
  34.     private String afterData;
  35. }

6、自定义ApplicationContextUtil

  1. import org.springframework.beans.BeansException;
  2. import org.springframework.context.ApplicationContext;
  3. import org.springframework.context.ApplicationContextAware;
  4. import org.springframework.stereotype.Component;
  5.  
  6. import java.io.Serializable;
  7.  
  8. @Component
  9. public class ApplicationContextUtil implements ApplicationContextAware, Serializable {
  10.     /**
  11.      * 上下文
  12.      */
  13.     private static ApplicationContext context;
  14.     @Override
  15.     public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
  16.         this.context = applicationContext;
  17.     }
  18.     public static ApplicationContext getApplicationContext() {
  19.         return context;
  20.     }
  21.     public static <T> T getBean(Class<T> beanClass) {
  22.         return context.getBean(beanClass);
  23.     }
  24. }

7、自定义sink 交由spring管理，处理变更数据

  1. import org.apache.flink.configuration.Configuration;
  2. import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
  3. import org.springframework.stereotype.Component;
  4. import lombok.extern.slf4j.Slf4j;
  5.  
  6. /**
  7.  * 自定义sink 交由spring管理
  8.  * 处理变更数据
  9.  **/
  10. @Component
  11. @Slf4j
  12. public class DataChangeSink extends RichSinkFunction<DataChangeInfo> {
  13.  
  14.     private static final long serialVersionUID = -74375380912179188L;
  15.  
  16.     private UserMapper userMapper;
  17.  
  18.     /**
  19.      * 在open()方法中动态注入Spring容器的类
  20.      * 在启动SpringBoot项目是加载了Spring容器，其他地方可以使用@Autowired获取Spring容器中的类；
  21.      * 但是Flink启动的项目中，默认启动了多线程执行相关代码，导致在其他线程无法获取Spring容器，
  22.      * 只有在Spring所在的线程才能使用@Autowired，故在Flink自定义的Sink的open()方法中初始化Spring容器
  23.      */
  24.     @Override
  25.     public void open(Configuration parameters) throws Exception {
  26.         super.open(parameters);
  27.         userMapper = ApplicationContextUtil.getBean(UserMapper.class);
  28.     }
  29.  
  30.     @Override
  31.     public void invoke(DataChangeInfo dataChangeInfo, Context context) {
  32.         log.info("收到变更原始数据：{}", dataChangeInfo);
  33.         // TODO 开始处理你的数据吧
  34.     }

以上是我亲自验证测试的结果，已发布生产环境，如有不足还请指正。

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

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

  

闽ICP备14008679号