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企服之家，中国第一个企服评测及商务社交产业平台。