一文说清flink从编码到摆设上线

弁言：如今flink的文章比力多，但一样寻常都关注某一特定方面，很少有一个文章，从一个简朴的例子入手，说清楚从编码、构建、摆设全流程是怎么样的。以是编写本文，本身做个纪录备查同时跟各人分享一下。本文以简朴的mysql cdc为例睁开阐明。
    环境阐明：MySQL：5.7；flink：1.14.0；hadoop：3.0.0；操纵体系：CentOS 7.6；JDK：1.8.0_401。
  1.MySQL

1.1 创建数据库和测试数据

数据库脚本：

1. CREATE DATABASE `flinktest`;
2. USE `flinktest`;
3. CREATE TABLE `products` (
4.   `id` int(11) NOT NULL AUTO_INCREMENT,
5.   `name` varchar(255) NOT NULL,
6.   `description` varchar(512) DEFAULT NULL,
7.   PRIMARY KEY (`id`)
8. ) ENGINE=InnoDB AUTO_INCREMENT=9 DEFAULT CHARSET=utf8mb4;
10. insert  into `products`(`id`,`name`,`description`) values
11. (1,'aaa','aaaa'),
12. (2,'ccc','ccc'),
13. (3,'dd','ddd'),
14. (4,'eeee','eee'),
15. (5,'ffff','ffff'),
16. (6,'hhhh','hhhh'),
17. (7,'iiii','iiii'),
18. (8,'jjjj','jjjj');

复制代码

账号利用root就行。
1.2 开启binlog

参考：
踩坑：测试过程中发现mysql 9.0不停无法获取更新的数据，终极利用的5.7。
2.编码

2.1 EnvUtil实现

EnvUtil用于创建flink的运行环境。

1. package com.zl.utils;
3. import org.apache.flink.api.common.restartstrategy.RestartStrategies;
4. import org.apache.flink.api.common.time.Time;
5. import org.apache.flink.configuration.Configuration;
6. import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
7. import org.apache.flink.streaming.api.CheckpointingMode;
8. import org.apache.flink.streaming.api.environment.CheckpointConfig;
9. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
10. import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
12. import java.time.Duration;
13. import java.time.ZoneOffset;
14. import java.util.concurrent.TimeUnit;
16. /**
17. * EnvUtil
18. * @description:
19. */
20. public class EnvUtil {
21.     /**
22.      * 设置flink执行环境
23.      * @param parallelism 并行度
24.      */
25.     public static StreamExecutionEnvironment setFlinkEnv(int parallelism) {
26.         // System.setProperty("HADOOP_USER_NAME", "用户名") 对应的是 hdfs文件系统目录下的路径：/user/用户名的文件夹名，本文为root
27.         System.setProperty("HADOOP_USER_NAME", "root");
28.         Configuration conf = new Configuration();
29.         conf.setInteger("rest.port", 1000);
30.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
32.         if (parallelism >0 ){
33.             //设置并行度
34.             env.setParallelism(parallelism);
35.         } else {
36.             env.setParallelism(1);// 默认1
37.         }
39.         // 添加重启机制
40.         env.setRestartStrategy(RestartStrategies.fixedDelayRestart(50, Time.minutes(6)));
41.         // 启动checkpoint,设置模式为精确一次 (这是默认值)，10*60*1000=60000
42.         env.enableCheckpointing(600000, CheckpointingMode.EXACTLY_ONCE);
43.         //rocksdb状态后端,启用增量checkpoint
44.         env.setStateBackend(new EmbeddedRocksDBStateBackend(true));
45.         //设置checkpoint路径
46.         CheckpointConfig checkpointConfig = env.getCheckpointConfig();
48.         // 同一时间只允许一个 checkpoint 进行(默认)
49.         checkpointConfig.setMaxConcurrentCheckpoints(1);
50.         //最小间隔，10*60*1000=60000
51.         checkpointConfig.setMinPauseBetweenCheckpoints(60000);
52.         // 取消任务后,checkpoint仍然保存
53.         checkpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
54.         //checkpoint容忍失败的次数
55.         checkpointConfig.setTolerableCheckpointFailureNumber(5);
56.         //checkpoint超时时间 默认10分钟
57.         checkpointConfig.setCheckpointTimeout(TimeUnit.MINUTES.toMillis(10));
58.         //禁用operator chain(方便排查反压)
59.         env.disableOperatorChaining();
60.         return env;
61.     }
63.     public static StreamTableEnvironment getFlinkTenv(StreamExecutionEnvironment env) {
64.         StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
65.         //设置时区 东八
66.         tenv.getConfig().setLocalTimeZone(ZoneOffset.ofHours(8));
67.         Configuration configuration = tenv.getConfig().getConfiguration();
68.         // 开启miniBatch
69.         configuration.setString("table.exec.mini-batch.enabled", "true");
70.         // 批量输出的间隔时间
71.         configuration.setString("table.exec.mini-batch.allow-latency", "5 s");
72.         // 防止OOM设置每个批次最多缓存数据的条数，可以设为2万条
73.         configuration.setString("table.exec.mini-batch.size", "20000");
74.         // 开启LocalGlobal
75.         configuration.setString("table.optimizer.agg-phase-strategy", "TWO_PHASE");
76.         //设置TTL API指定
77.         tenv.getConfig().setIdleStateRetention(Duration.ofHours(25));
79.         return tenv;
80.     }
82. }

复制代码

2.2 核心代码

1. package com.zl;
3. import com.ververica.cdc.connectors.mysql.source.MySqlSource;
4. import com.ververica.cdc.connectors.mysql.table.StartupOptions;
5. import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
6. import com.zl.utils.EnvUtil;
7. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
8. import org.apache.flink.api.common.restartstrategy.RestartStrategies;
9. import org.apache.flink.api.common.time.Time;
10. import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
11. import org.apache.flink.streaming.api.CheckpointingMode;
12. import org.apache.flink.streaming.api.environment.CheckpointConfig;
13. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
15. import java.util.Arrays;
16. import java.util.List;
17. import java.util.concurrent.TimeUnit;
19. public class MysqlExampleDatastream {
20.     public static void main(String[] args) throws Exception {
22.         List<String> SYNC_TABLES = Arrays.asList("flinktest1.products");
23.         MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
24.                 .hostname("10.86.37.169")
25.                 .port(3306)
26.                 .databaseList("flinktest1")
27.                 .tableList(String.join(",", SYNC_TABLES))
28.                 .username("root")
29.                 .password("pwd")// 记得修改为实际密码
30.                 .startupOptions(StartupOptions.initial())
31.                 .deserializer(new JsonDebeziumDeserializationSchema())
32.                 .build();
34.         // 配置运行环境，并行度1
35.         StreamExecutionEnvironment env = EnvUtil.setFlinkEnv(1);
36.         // 程序间隔离，每个程序单独设置
37.         env.getCheckpointConfig().setCheckpointStorage("hdfs://10.86.97.191:9000/flinktest/MysqlExampleDatastream");
39.         // 如果不能正常读取mysql的binlog：①可能是mysql没有打开binlog或者mysql版本不支持（当前在mysql5.7.20环境下，功能正常）；
40.         // ②可能是数据库ip、port、账号、密码错误。
41.         env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source")
42.                 .setParallelism(1).print();
44.         env.execute("Print MySQL Snapshot + Binlog");
46.     }
48. }

复制代码

2.3 依靠

1. <project xmlns="http://maven.apache.org/POM/4.0.0"
2.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
3.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
4.          http://maven.apache.org/maven-v4_0_0.xsd">
5.     <modelVersion>4.0.0</modelVersion>
6.     <groupId>com.zl.flinkcdc</groupId>
7.     <artifactId>FlickCDC</artifactId>
8.     <packaging>jar</packaging>
9.     <version>1.0-SNAPSHOT</version>
10.     <name>FlickCDC</name>
11.     <properties>
12.         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
13.         <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
14.         <maven.compiler.source>8</maven.compiler.source>
15.         <maven.compiler.target>8</maven.compiler.target>
16.         <flink-version>1.14.0</flink-version>
17.         <flink-cdc-version>2.4.0</flink-cdc-version>
18.         <hadoop.version>3.0.0</hadoop.version>
19.         <slf4j.version>1.7.25</slf4j.version>
20.         <log4j.version>2.16.0</log4j.version>
21.     </properties>
22.     <dependencies>
23.         <dependency>
24.             <groupId>junit</groupId>
25.             <artifactId>junit</artifactId>
26.             <version>3.8.1</version>
27.             <scope>test</scope>
28.         </dependency>
29.         <dependency>
30.             <groupId>org.apache.flink</groupId>
31.             <artifactId>flink-java</artifactId>
32.             <version>${flink-version}</version>
33.         </dependency>
35.         <dependency>
36.             <groupId>org.apache.flink</groupId>
37.             <artifactId>flink-streaming-java_2.11</artifactId>
38.             <version>${flink-version}</version>
39.         </dependency>
41.         <dependency>
42.             <groupId>com.ververica</groupId>
43.             <artifactId>flink-connector-mysql-cdc</artifactId>
44.             <version>${flink-cdc-version}</version>
45.         </dependency>
47.         <dependency>
48.             <groupId>org.apache.flink</groupId>
49.             <artifactId>flink-shaded-guava</artifactId>
50.             <version>30.1.1-jre-15.0</version>
51.         </dependency>
52.         <!--<dependency>
53.             <groupId>org.apache.flink</groupId>
54.             <artifactId>flink-shaded-guava</artifactId>
55.             <version>18.0-13.0</version>
56.         </dependency>-->
58.         <dependency>
59.             <groupId>org.apache.flink</groupId>
60.             <artifactId>flink-connector-base</artifactId>
61.             <version>${flink-version}</version>
62.         </dependency>
64.         <dependency>
65.             <groupId>org.apache.flink</groupId>
66.             <artifactId>flink-clients_2.11</artifactId>
67.             <version>${flink-version}</version>
68.         </dependency>
70.         <dependency>
71.             <groupId>org.apache.flink</groupId>
72.             <artifactId>flink-table-common</artifactId>
73.             <version>${flink-version}</version>
74.         </dependency>
76.         <dependency>
77.             <groupId>org.apache.flink</groupId>
78.             <artifactId>flink-runtime-web_2.11</artifactId>
79.             <version>${flink-version}</version>
80.         </dependency>
82.         <!--        hadoop相关依赖-->
83.         <dependency>
84.             <groupId>org.apache.hadoop</groupId>
85.             <artifactId>hadoop-client</artifactId>
86.             <version>${hadoop.version}</version>
87.             <scope>provided</scope>
88.             <exclusions>
89.                 <exclusion>
90.                     <artifactId>commons-cli</artifactId>
91.                     <groupId>commons-cli</groupId>
92.                 </exclusion>
93.                 <exclusion>
94.                     <artifactId>commons-compress</artifactId>
95.                     <groupId>org.apache.commons</groupId>
96.                 </exclusion>
97.                 <exclusion>
98.                     <artifactId>guava</artifactId>
99.                     <groupId>com.google.guava</groupId>
100.                 </exclusion>
101.                 <exclusion>
102.                     <artifactId>jackson-annotations</artifactId>
103.                     <groupId>com.fasterxml.jackson.core</groupId>
104.                 </exclusion>
105.                 <exclusion>
106.                     <artifactId>jackson-core</artifactId>
107.                     <groupId>com.fasterxml.jackson.core</groupId>
108.                 </exclusion>
109.                 <exclusion>
110.                     <artifactId>jackson-databind</artifactId>
111.                     <groupId>com.fasterxml.jackson.core</groupId>
112.                 </exclusion>
113.                 <exclusion>
114.                     <artifactId>slf4j-api</artifactId>
115.                     <groupId>org.slf4j</groupId>
116.                 </exclusion>
117.             </exclusions>
118.         </dependency>
119.         <dependency>
120.             <groupId>org.apache.hadoop</groupId>
121.             <artifactId>hadoop-common</artifactId>
122.             <version>${hadoop.version}</version>
123.             <scope>provided</scope>
124.             <exclusions>
125.                 <exclusion>
126.                     <artifactId>asm</artifactId>
127.                     <groupId>org.ow2.asm</groupId>
128.                 </exclusion>
129.                 <exclusion>
130.                     <artifactId>avro</artifactId>
131.                     <groupId>org.apache.avro</groupId>
132.                 </exclusion>
133.                 <exclusion>
134.                     <artifactId>commons-cli</artifactId>
135.                     <groupId>commons-cli</groupId>
136.                 </exclusion>
137.                 <exclusion>
138.                     <artifactId>commons-codec</artifactId>
139.                     <groupId>commons-codec</groupId>
140.                 </exclusion>
141.                 <exclusion>
142.                     <artifactId>commons-compress</artifactId>
143.                     <groupId>org.apache.commons</groupId>
144.                 </exclusion>
145.                 <exclusion>
146.                     <artifactId>commons-io</artifactId>
147.                     <groupId>commons-io</groupId>
148.                 </exclusion>
149.                 <exclusion>
150.                     <artifactId>commons-lang3</artifactId>
151.                     <groupId>org.apache.commons</groupId>
152.                 </exclusion>
153.                 <exclusion>
154.                     <artifactId>commons-logging</artifactId>
155.                     <groupId>commons-logging</groupId>
156.                 </exclusion>
157.                 <exclusion>
158.                     <artifactId>commons-math3</artifactId>
159.                     <groupId>org.apache.commons</groupId>
160.                 </exclusion>
161.                 <exclusion>
162.                     <artifactId>guava</artifactId>
163.                     <groupId>com.google.guava</groupId>
164.                 </exclusion>
165.                 <exclusion>
166.                     <artifactId>jackson-databind</artifactId>
167.                     <groupId>com.fasterxml.jackson.core</groupId>
168.                 </exclusion>
169.                 <exclusion>
170.                     <artifactId>jaxb-api</artifactId>
171.                     <groupId>javax.xml.bind</groupId>
172.                 </exclusion>
173.                 <exclusion>
174.                     <artifactId>log4j</artifactId>
175.                     <groupId>log4j</groupId>
176.                 </exclusion>
177.                 <exclusion>
178.                     <artifactId>nimbus-jose-jwt</artifactId>
179.                     <groupId>com.nimbusds</groupId>
180.                 </exclusion>
181.                 <exclusion>
182.                     <artifactId>slf4j-api</artifactId>
183.                     <groupId>org.slf4j</groupId>
184.                 </exclusion>
185.                 <exclusion>
186.                     <artifactId>slf4j-log4j12</artifactId>
187.                     <groupId>org.slf4j</groupId>
188.                 </exclusion>
189.                 <exclusion>
190.                     <artifactId>zookeeper</artifactId>
191.                     <groupId>org.apache.zookeeper</groupId>
192.                 </exclusion>
193.                 <exclusion>
194.                     <artifactId>jsr305</artifactId>
195.                     <groupId>com.google.code.findbugs</groupId>
196.                 </exclusion>
197.                 <exclusion>
198.                     <artifactId>gson</artifactId>
199.                     <groupId>com.google.code.gson</groupId>
200.                 </exclusion>
201.             </exclusions>
202.         </dependency>
203.         <dependency>
204.             <groupId>org.apache.hadoop</groupId>
205.             <artifactId>hadoop-hdfs</artifactId>
206.             <version>${hadoop.version}</version>
207.             <scope>provided</scope>
208.             <exclusions>
209.                 <exclusion>
210.                     <artifactId>commons-cli</artifactId>
211.                     <groupId>commons-cli</groupId>
212.                 </exclusion>
213.                 <exclusion>
214.                     <artifactId>guava</artifactId>
215.                     <groupId>com.google.guava</groupId>
216.                 </exclusion>
217.                 <exclusion>
218.                     <artifactId>jackson-databind</artifactId>
219.                     <groupId>com.fasterxml.jackson.core</groupId>
220.                 </exclusion>
221.             </exclusions>
222.         </dependency>
223.         <dependency>
224.             <groupId>commons-cli</groupId>
225.             <artifactId>commons-cli</artifactId>
226.             <version>1.5.0</version>
227.         </dependency>
229.         <!--mvn install:install-file -Dfile
230.         =D:/maven/flink-shaded-hadoop-3-uber-3.1.1.7.2.9.0-173-9.0.jar
231.        -DgroupId=org.apache.flink -DartifactId
232.        =flink-shaded-hadoop-3 -Dversion=3.1.1.7.2.9.0-173-9.0 -Dpackaging=jar-->
233.         <dependency>
234.             <groupId>org.apache.flink</groupId>
235.             <artifactId>flink-shaded-hadoop-3</artifactId>
236.             <version>3.1.1.7.2.9.0-173-9.0</version>
237.         </dependency>
238.     </dependencies>
240.     <build>
241.         <plugins>
242.             <plugin>
243.                 <groupId>org.apache.maven.plugins</groupId>
244.                 <artifactId>maven-assembly-plugin</artifactId>
245.                 <version>3.0.0</version>
246.                 <configuration>
247.                     <descriptorRefs>
248.                         <descriptorRef>jar-with-dependencies</descriptorRef>
249.                     </descriptorRefs>
250.                     <archive>
251.                         <manifest>
252.                             <addClasspath>true</addClasspath>
253.                             <mainClass>com.zl.MysqlExampleDatastream</mainClass>
254.                         </manifest>
255.                     </archive>
256.                 </configuration>
257.                 <executions>
258.                     <execution>
259.                         <id>make-assembly</id>
260.                         <phase>package</phase>
261.                         <goals>
262.                             <goal>single</goal>
263.                         </goals>
264.                     </execution>
265.                 </executions>
266.             </plugin>
267.         </plugins>
268.     </build>
269. </project>

复制代码

完备代码见：https://gitee.com/core815/flink-cdc-mysql
3.打包

mvn版本：3.5.4。
到pom.xml所在路径，实行“mvn package”

打包效果：

4.jar直接运行

1. java -jar FlickCDC-1.0-SNAPSHOT-jar-with-dependencies.jar

复制代码

5.flink yarn运行

hadoop、flink、yarn环境见：
把FlickCDC-1.0-SNAPSHOT-jar-with-dependencies.jar放到“/home”路径下。
实行下面下令：

1. flink run-application -t yarn-application -Dparallelism.default=1 -Denv.java.opts=" -Dfile.encoding=UTF-8 -Dsun.jnu.encoding=UTF-8" -Dtaskmanager.memory.process.size=1g -Dyarn.application.name="FlinkCdcMysql"  -Dtaskmanager.numberOfTaskSlots=1 -c com.zl.MysqlExampleDatastream /home/FlickCDC-1.0-SNAPSHOT-jar-with-dependencies.jar

复制代码

控制台看到如下打印：

yarn管理页面：

运行日记检察步调：

下面即可看到完备日记：

6.常见题目

6.1 题目1

日记错误：
The MySQL server has a timezone offset (0 seconds ahead of UTC) which does not match the configured timezone Asia/Shanghai. Specify the right server-time-zone to avoid inconsistencies for time-related fields.
办理：
修改my.cnf文件。
[mysqld]
default-time-zone=‘Asia/Shanghai’
重启MySQL服务。
6.2 题目2：hdfs

日记错误：
Permission denied: user=PC2023, access=WRITE, inode=“/”:root:supergroup:drwxr-xr-x
办理：

1. 临时解决
2. hadoop fs -chmod -R 777 /

复制代码

6.3 题目3：guava30 guava18辩论

分析：
flink 1.13 cdc2.3的组合轻易出这个题目。
办理：
参考：https://developer.aliyun.com/ask/574901
flink 利用1.14.0版本；cdc利用2.4.0版本。
6.4 题目4

日记错误：
/user/root/.flink/application_1733492706887_0002/log4j.properties could only be written to 0 of the 1 minReplication nodes
办理：
https://www.pianshen.com/article/1554707968/