【flink-cdc】flink-cdc 3版本debug启动pipeline任务，mysql-doris ...

官方文档
github堆栈地址
Flink cdc debug调试动态变动表结构
颠末测试利用，在启动任务设置Modify classpath添加jar的方式，容易出错classNotFoundException等等。

一、build project

flink-cdc版本：3.2.1

1. mvn clean package "-Dmaven.test.skip=true" "-Drat.skip=true" "-Dcheckstyle.skip"
2. # 当然install 也可以
3. mvn clean install "-Dmaven.test.skip=true" "-Drat.skip=true" "-Dcheckstyle.skip"

复制代码

二、create module for test

在flink-cdc项目中新建一个module flink-cdc-test
将必要的flink cdc的connector依靠添加至pom

2.     <properties>        
3.         <flink.version>1.18.1</flink.version>  
4.         <maven.compiler.source>8</maven.compiler.source>  
5.         <maven.compiler.target>8</maven.compiler.target>  
6.         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>  
7.     </properties>  
8.   
9.     <dependencies>        
10.         <dependency>  
11.             <groupId>org.apache.flink</groupId>  
12.             <artifactId>flink-table-common</artifactId>  
13.             <version>${flink.version}</version>  
14.             <scope>test</scope>  
15.         </dependency>  
16.   
17.         <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-table-runtime -->  
18.         <dependency>  
19.             <groupId>org.apache.flink</groupId>  
20.             <artifactId>flink-table-runtime</artifactId>  
21.             <version>${flink.version}</version>  
22.             <scope>test</scope>  
23.         </dependency>  
24.   
25.         <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-table-api-java-bridge -->  
26.         <dependency>  
27.             <groupId>org.apache.flink</groupId>  
28.             <artifactId>flink-table-api-java-bridge</artifactId>  
29.             <version>${flink.version}</version>  
30.             <scope>test</scope>  
31.         </dependency>  
32.   
33.         <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-core -->  
34.         <dependency>  
35.             <groupId>org.apache.flink</groupId>  
36.             <artifactId>flink-core</artifactId>  
37.             <version>${flink.version}</version>  
38.             <scope>test</scope>  
39.         </dependency>  
40.   
41.         <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-streaming-java -->  
42.         <dependency>  
43.             <groupId>org.apache.flink</groupId>  
44.             <artifactId>flink-streaming-java</artifactId>  
45.             <version>1.18.1</version>  
46.             <scope>test</scope>  
47.         </dependency>  
48.   
49.         <dependency>            
50.             <groupId>org.apache.flink</groupId>  
51.             <artifactId>flink-clients</artifactId>  
52.             <version>1.18.1</version>  
53.             <scope>test</scope>  
54.         </dependency>  
55.   
56.         <dependency>            
57.         <groupId>org.apache.flink</groupId>  
58.             <artifactId>flink-cdc-cli</artifactId>  
59.             <version>${project.version}</version>  
60.             <scope>test</scope>  
61.         </dependency>  
62.   
63.         <dependency>            
64.             <groupId>org.apache.flink</groupId>  
65.             <artifactId>flink-cdc-pipeline-connector-mysql</artifactId>  
66.             <version>${project.version}</version>  
67.             <scope>test</scope>  
68.         </dependency>  
69.   
70.         <dependency>            
71.             <groupId>org.apache.flink</groupId>  
72.             <artifactId>flink-cdc-pipeline-connector-kafka</artifactId>  
73.             <version>${project.version}</version>  
74.             <scope>test</scope>  
75.         </dependency>  
76.   
77.         <dependency>            
78.             <groupId>org.apache.flink</groupId>  
79.             <artifactId>flink-cdc-pipeline-connector-doris</artifactId>  
80.             <version>${project.version}</version>  
81.             <scope>test</scope>  
82.         </dependency>  
83.   
84.         <!-- https://mvnrepository.com/artifact/mysql/mysql-connector-java -->  
85.         <dependency>  
86.             <groupId>mysql</groupId>  
87.             <artifactId>mysql-connector-java</artifactId>  
88.             <version>8.0.27</version>  
89.         </dependency>  
90.    
91.     </dependencies>  

复制代码

create pipeline yaml

创建flink cdc pipeline的设置文件mysql-to-doris.yaml：
注意：drois的端口号不是9030，而是8030

1. source:  
2.   type: mysql  
3.   name: MySQL-Source  
4.   hostname: 192.168.100.123
5.   port: 3306  
6.   username: root  
7.   password: 123456  
8.   tables: test.student  
9.   server-id: 5401-5404  
10.   server-time-zone: UTC  
11.   
12. sink:  
13.   type: doris  
14.   name: Doris-Sink  
15.   fenodes: 192.168.101:8030
16.   username: root
17.   password: 123456  
18.   
19.   
20. pipeline:  
21.   name: MySQL to Doris Schema Evolution  
22.   parallelism: 1

复制代码

create pipeline start entrypoint

测试类，需在src/test目次下：

1. import org.apache.flink.cdc.cli.CliFrontend;  
2. import org.junit.jupiter.api.Test;  
3.   
4. import java.util.ArrayList;  
5. import java.util.List;  
6.   
7. public class MysqlPipelineTest {  
8.   
9.     @Test  
10.     public void testMysql() throws Exception {  
11.         List<String> args= new ArrayList<>();  
12.         args.add("D:\\flink-cdc-release-3.2.1\\flink-cdc-test\\pipelines\\mysql-to-doris.yaml");  
13.         args.add("--use-mini-cluster");  
14.         args.add("true");  
15.         CliFrontend.main(args.toArray(new String[0]));  
16.     }  
17. }

复制代码

修改启动设置添加FLINK_HOME情况变量：

启动任务：

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