ToB企服应用市场:ToB评测及商务社交产业平台

标题: Flink cdc 实现源表sqlserver到目标表sqlserver或者mysql [打印本页]
作者: 麻花痒    时间: 2024-10-30 18:38
标题: Flink cdc 实现源表sqlserver到目标表sqlserver或者mysql
说明，本文章只提供借鉴，博主也是初次接触flink，请包涵
flink版本1.17.0，flink cdc版本2.3.0，百分百可以及时同步，亲测可用

起首看下我的程序结构吧

准备

1、依赖

  1.     <properties>
  2.         <flink.version>1.17.0</flink.version>
  3.         <flink.cdc.version>2.3.0</flink.cdc.version>
  4.         <logback-classic.version>1.1.11</logback-classic.version>
  5.         <slf4j-api.version>1.7.25</slf4j-api.version>
  6.     </properties>
  8.     <!-- 依赖器 -->
  9.     <dependencyManagement>
  10.         <dependencies>
  12.             <!-- SpringCloud 微服务 -->
  13.             <dependency>
  14.                 <groupId>org.springframework.cloud</groupId>
  15.                 <artifactId>spring-cloud-dependencies</artifactId>
  16.                 <version>2021.0.4</version>
  17.                 <type>pom</type>
  18.                 <scope>import</scope>
  19.             </dependency>
  21.             <!-- SpringCloud Alibaba 微服务 -->
  22.             <dependency>
  23.                 <groupId>com.alibaba.cloud</groupId>
  24.                 <artifactId>spring-cloud-alibaba-dependencies</artifactId>
  25.                 <version>2021.0.4.0</version>
  26.                 <type>pom</type>
  27.                 <scope>import</scope>
  28.             </dependency>
  30.             <!-- SpringBoot 依赖配置 -->
  31.             <dependency>
  32.                 <groupId>org.springframework.boot</groupId>
  33.                 <artifactId>spring-boot-dependencies</artifactId>
  34.                 <version>2.7.3</version>
  35.                 <type>pom</type>
  36.                 <scope>import</scope>
  37.             </dependency>
  38.         </dependencies>
  39.     </dependencyManagement>
  41.     <dependencies>
  42.         <!-- bootstrap 启动器 -->
  43.         <dependency>
  44.             <groupId>org.springframework.cloud</groupId>
  45.             <artifactId>spring-cloud-starter-bootstrap</artifactId>
  46.         </dependency>
  48.         <dependency>
  49.             <groupId>org.springframework.boot</groupId>
  50.             <artifactId>spring-boot-starter-web</artifactId>
  51.             <version>2.7.3</version>
  52.         </dependency>
  54.         <dependency>
  55.             <groupId>org.springframework.boot</groupId>
  56.             <artifactId>spring-boot-starter-freemarker</artifactId>
  57.             <version>2.4.4</version>
  58.         </dependency>
  60.         <dependency>
  61.             <groupId>org.mybatis</groupId>
  62.             <artifactId>mybatis</artifactId>
  63.             <version>3.5.9</version>
  64.         </dependency>
  66.         <dependency>
  67.             <groupId>com.microsoft.sqlserver</groupId>
  68.             <artifactId>mssql-jdbc</artifactId>
  69.             <version>9.4.1.jre8</version>
  70.         </dependency>
  72.         <dependency>
  73.             <groupId>com.google.code.gson</groupId>
  74.             <artifactId>gson</artifactId>
  75.             <version>2.8.2</version>
  76.         </dependency>
  77.         <dependency>
  78.             <groupId>org.apache.flink</groupId>
  79.             <artifactId>flink-java</artifactId>
  80.             <version>${flink.version}</version>
  81.         </dependency>
  82.         <dependency>
  83.             <groupId>org.apache.flink</groupId>
  84.             <artifactId>flink-streaming-java</artifactId>
  85.             <version>${flink.version}</version>
  86.         </dependency>
  87.         
  88.         <dependency>
  89.             <groupId>com.ververica</groupId>
  90.             <artifactId>flink-connector-sqlserver-cdc</artifactId>
  91.             <version>${flink.cdc.version}</version>
  92.         </dependency>
  94.         <dependency>
  95.             <groupId>com.ververica</groupId>
  96.             <artifactId>flink-connector-oracle-cdc</artifactId>
  97.             <version>${flink.cdc.version}</version>
  98.         </dependency>
  100.         <dependency>
  101.             <groupId>mysql</groupId>
  102.             <artifactId>mysql-connector-java</artifactId>
  103.             <version>8.0.33</version>
  104.         </dependency>
  106.         <dependency>
  107.             <groupId>org.springframework</groupId>
  108.             <artifactId>spring-context-support</artifactId>
  109.             <version>5.3.22</version>
  110.         </dependency>
  112.         <dependency>
  113.             <groupId>org.apache.flink</groupId>
  114.             <artifactId>flink-runtime-web</artifactId>
  115.             <version>${flink.version}</version>
  116.         </dependency>
  118.         <dependency>
  119.             <groupId>com.oracle.database.jdbc</groupId>
  120.             <artifactId>ojdbc8</artifactId>
  121.             <version>12.2.0.1</version>
  122.         </dependency>
  124.         <dependency>
  125.             <groupId>com.ibm.db2</groupId>
  126.             <artifactId>jcc</artifactId>
  127.             <version>11.5.0.0</version>
  128.         </dependency>
  130.         <dependency>
  131.             <groupId>org.projectlombok</groupId>
  132.             <artifactId>lombok</artifactId>
  133.             <version>1.18.26</version>
  134.             <scope>compile</scope>
  135.         </dependency>
  137.         <dependency>
  138.             <groupId>org.springframework.boot</groupId>
  139.             <artifactId>spring-boot-configuration-processor</artifactId>
  140.             <optional>true</optional>
  141.         </dependency>
  143.     </dependencies>
复制代码
2、打包插件

  1.     <build>
  2.             <plugins>
  3.                 <plugin>
  4.                     <groupId>org.apache.maven.plugins</groupId>
  5.                     <artifactId>maven-compiler-plugin</artifactId>
  6.                     <version>3.1</version>
  7.                     <configuration>
  8.                         <source>1.8</source>
  9.                         <target>1.8</target>
  10.                     </configuration>
  11.                 </plugin>
  12.                 <plugin>
  13.                     <groupId>org.apache.maven.plugins</groupId>
  14.                     <artifactId>maven-shade-plugin</artifactId>
  15.                     <version>3.1.1</version>
  16.                     <executions>
  17.                         <!-- Run shade goal on package phase -->
  18.                         <execution>
  19.                             <phase>package</phase>
  20.                             <goals>
  21.                                 <goal>shade</goal>
  22.                             </goals>
  23.                             <configuration>
  24.                                 <artifactSet>
  25.                                     <excludes>
  26.                                         <exclude>org.apache.flink:force-shading</exclude>
  27.                                         <exclude>com.google.code.findbugs:jsr305</exclude>
  28. <!--                                        <exclude>org.slf4j:*</exclude>-->
  29. <!--                                        <exclude>org.apache.logging.log4j:*</exclude>-->
  30.                                     </excludes>
  31.                                 </artifactSet>
  32.                                 <filters>
  33.                                     <filter>
  34.                                         <artifact>*:*</artifact>
  35.                                         <excludes>
  36.                                             <exclude>META-INF/*.SF</exclude>
  37.                                             <exclude>META-INF/*.DSA</exclude>
  38.                                             <exclude>META-INF/*.RSA</exclude>
  39.                                         </excludes>
  40.                                     </filter>
  41.                                 </filters>
  42.                                 <transformers combine.children="append">
  43.                                     <!-- TODO：这个防止多个connector的相同类名覆盖-->
  44.                                     <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
  45.                                     <!--指定 主类-->
  46.                                     <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
  47.                                         <mainClass>com.hbos.FlinkCDCApplication</mainClass>
  48.                                     </transformer>
  49.                                 </transformers>
  50.                             </configuration>
  51.                         </execution>
  52.                     </executions>
  53.                 </plugin>
  54.             </plugins>
  55.             <resources>
  56.                 <resource>
  57.                     <directory>src/main/resources</directory>
  58.                     <includes>
  59.                         <include>application.yml</include>
  60.                         <include>**/*.ftl</include>
  61.                         <include>**/*.xml</include>
  62.                         <include>**/*.properties</include>
  63.                     </includes>
  64.                     <filtering>true</filtering>
  65.                 </resource>
  66.             </resources>
  67.         </build>
复制代码
编写主类

  1. @SpringBootApplication(excludeName = {"org.springframework.security"})
  2. @Slf4j
  3. public class FlinkCDCApplication {
  5.     public static void main(String[] args){
  7.         SpringApplication.run(FlinkCDCApplication.class,args);
  8.         log.info("CDC服务启动成功");
  9.         System.out.println("=========================="+"\n"+
  10.                 "(♥◠‿◠)ﾉﾞCDC服务启动成功"+"\n"
  11.                 +"==========================");
  12.     }
  14. }
复制代码
编写源表SourceConfig 类设置

  1. package com.hbos.cdc.config;
  3. import com.hbos.cdc.entity.SourceDataProperties;
  4. import com.hbos.cdc.listener.DataChangeSink;
  5. import com.ververica.cdc.connectors.sqlserver.SqlServerSource;
  6. import com.ververica.cdc.connectors.sqlserver.table.StartupOptions;
  7. import com.ververica.cdc.debezium.DebeziumSourceFunction;
  8. import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
  9. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  10. import org.apache.kafka.connect.json.DecimalFormat;
  11. import org.apache.kafka.connect.json.JsonConverterConfig;
  12. import org.springframework.beans.factory.annotation.Qualifier;
  13. import org.springframework.boot.context.properties.ConfigurationProperties;
  14. import org.springframework.context.annotation.Bean;
  15. import org.springframework.context.annotation.Configuration;
  17. import java.util.HashMap;
  18. import java.util.Map;
  19. import java.util.Properties;
  21. @Configuration
  22. public class SourceConfig {
  24.     @Bean(name = "sourceInfo")
  25.     @ConfigurationProperties(prefix = "source")
  26.     public SourceDataProperties sourceInfo() {
  27.         return new SourceDataProperties();
  28.     }
  30.     @Bean
  31.     public StreamExecutionEnvironment source(@Qualifier("sourceInfo") SourceDataProperties sourceInfo) throws Exception {
  34.         Map config = new HashMap();
  35.         config.put(JsonConverterConfig.DECIMAL_FORMAT_CONFIG, DecimalFormat.NUMERIC.name());
  36.         JsonDebeziumDeserializationSchema jdd = new JsonDebeziumDeserializationSchema(false, config);
  38.         //通过FlinkCDC构建SourceFunction
  39.         DebeziumSourceFunction<String> sourceFunction  = SqlServerSource.<String>builder()
  40.                 .hostname(sourceInfo.getHost())
  41.                 .port(sourceInfo.getPort())
  42.                 .username(sourceInfo.getUsername())
  43.                 .password(sourceInfo.getPassword())
  44.                 .database(sourceInfo.getDatabase()) //设置捕获的库名
  45.                 .tableList(sourceInfo.getTableList()) //设置捕获的表,也可以去掉表，那就证明监控某库下所有的表
  46.                 .deserializer(jdd) //转换成JSON
  47.                 /*
  48.                  * initial初始化快照,即全量导入后增量导入(检测更新数据写入)
  49.                  * latest:只进行增量导入(不读取历史变化)
  50.                  */
  51.                 .startupOptions(StartupOptions.initial())//启动模式
  52.                 .build();
  54.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  55.         env.setParallelism(5); //全局并行度为5
  57.         env.addSource(sourceFunction).addSink(new DataChangeSink()).setParallelism(2);
  58.         //4、启动任务
  59.         env.execute("JHYGCDC1");
  60.         return env;
  61.     }
  62. }
复制代码
由于flink的结构和springboot的结构不一样，以是要想把bean交给spring管理那就必要手动去创建
创建bean工具，也就是SpringBeanUtil类

  1. @Component
  2. public class SpringBeanUtil implements ApplicationContextAware {
  4.     private static ApplicationContext applicationContext;
  6.     public void setApplicationContext(ApplicationContext applicationContext)
  7.       throws BeansException {
  8.         SpringBeanUtil.applicationContext = applicationContext;
  9.     }
  11.     public static <T> T getBean(Class<T> clazz) {
  12.         return applicationContext.getBean(clazz);
  13.     }
  15.     public static Object getBean(String name) throws BeansException {
  17.         return applicationContext.getBean(name);
  18.     }
  19. }
复制代码
创建源表实体和目标表实体

源实体SourceDataProperties类
  1. //@Component
  2. @Data
  3. //@ConfigurationProperties(prefix = "source")
  4. public class SourceDataProperties {
  6.     private String driverClassName;
  8.     private String url;
  10.     private String host;
  12.     private int port;
  14.     private String username;
  16.     private String password;
  18.     private String database;
  20.     private String tableList;
  22. }
复制代码
目标实体TargetDataProperties 类
  1. @Component
  2. @Data
  3. @ConfigurationProperties(prefix = "target")
  4. public class TargetDataProperties {
  6.     private String driverClassName;
  8.     private String url;
  10.     private String host;
  12.     private String port;
  14.     private String username;
  16.     private String password;
  18.     private String database;
  20. //    @Value("${CDC.DataSource.target.schema}")
  21. //    private String schema;
  22. //
  23. //    @Value("${CDC.DataSource.target.tableList}")
  24. //    private String tableList;
  26. }
复制代码
sink类

也就是处置处罚业务逻辑的代码，有点多我就不粘贴出来了，你们可以自己处置处罚你们自己的业务逻辑
也可以借鉴我的，地址如下
flink cdc业务逻辑层处置处罚-CSDN博客
https://blog.csdn.net/weixin_50448162/article/details/140635326?spm=1001.2014.3001.5502


免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。



欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/) Powered by Discuz! X3.4