java Flink（十八）flink 1.10从程序打包到flink安装、服务部署包含kafka、检查点使用、状态保存、故障状态恢复checkpoint等

本是古典何须时尚 2023-01-20 04:57 22阅读 0赞

### **一、检查服务器运行环境\*（重要）** ###

运行 vim /etc/profile

![20210511141422683.png][]

查看当前服务器的运行环境 jdk需要大于1.8以上版本（如果没有需要安装jdk）

hadoop版本2.7.5

scala版本2.11

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70][]

### **二、flink下载**  ###

下载我们需要的flink版本 [https://archive.apache.org/dist/flink/][https_archive.apache.org_dist_flink]

我们scala是2.11

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70 1][]

下载之后 tar zxvf   flink-1.10.0-bin-scala\_2.11.tgz进行解压

修改conf中的部分配置 vim conf/flink-conf.yaml

（详细介绍看之前的文章 有个配置介绍）

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70 2][]

下载hadoop依赖jar包 我们hadoop是2.7.5版本   [https://flink.apache.org/zh/downloads.html][https_flink.apache.org_zh_downloads.html]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70 3][]

下载完成 将jar包移动到flink的lib目录下（如果已启动  建议重启）

![20210511143238585.png][]

运行flink  ./bin/start-cluster.sh

![20210511142921427.png][]

查看本机的8081 访问到flink的ui界面即启动成功

### 三、flink代码配置checkpoint ###

我们使用rocksdb 保存到hdfs下边

### **这里需要注意几个点** ###

1、与hdfs网络、端口通畅

2、配置hadoop集群hosts：

vim /etc/hosts

![2021051114190962.png][]

![20210511141942578.png][]

3、配置hdfs目录的权限是可写的、保证文件的写入

//每60s执行一次checkpoint
            env.enableCheckpointing(5000);
            // 设置模式为exactly-once （这是默认值）表示数据只被处理一次
            env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
            // 确保检查点之间有至少5min的间隔【checkpoint最小间隔】
            env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
            // 检查点必须在一分钟内完成，或者被丢弃【checkpoint的超时时间】
            env.getCheckpointConfig().setCheckpointTimeout(60000);
            // 同一时间只允许进行一个检查点
            env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
            //env.setStateBackend(new FsStateBackend("hdfs://192.168.40.150:9000/flink_lgy_test"));
            env.setStateBackend(new RocksDBStateBackend("hdfs://192.168.40.150:9000/flink"));
            // 表示一旦Flink处理程序被cancel后，会保留Checkpoint数据，以便根据实际需要恢复到指定的Checkpoint【详细解释见备注】
            //ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION:表示一旦Flink处理程序被cancel后，会保留Checkpoint数据，以便根据实际需要恢复到指定的Checkpoint
            //ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION: 表示一旦Flink处理程序被cancel后，会删除Checkpoint数据，只有job执行失败的时候才会保存checkpoint
            env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

### **四、flink代码打包** ###

首先修改pom文件

添加：

注意：

修改其中的主类，就是你的程序入口。

在主类复制引用修改即可

![20210511142325252.png][]

<build>
            <plugins>
                <plugin>
                    <artifactId>maven-compiler-plugin</artifactId>
                    <version>3.6.0</version>
                    <configuration>
                        <source>1.8</source>
                        <target>1.8</target>
                    </configuration>
                </plugin>
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-shade-plugin</artifactId>
                    <version>3.1.0</version>
                    <executions>
                        <execution>
                            <phase>package</phase>
                            <goals>
                                <goal>shade</goal>
                            </goals>
                            <configuration>
                                <transformers>
                                    <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
    
                                        <mainClass>Project.CalcHealthWarn</mainClass>
                                    </transformer>
                                    <transformer implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
                                        <resource>reference.conf</resource>
                                    </transformer>
                                </transformers>
                                <relocations>
                                    <relocation>
                                        <pattern>org.codehaus.plexus.util</pattern>
                                        <shadedPattern>org.shaded.plexus.util</shadedPattern>
                                        <excludes>
                                            <exclude>org.codehaus.plexus.util.xml.Xpp3Dom</exclude>
                                            <exclude>org.codehaus.plexus.util.xml.pull.*</exclude>
                                        </excludes>
                                    </relocation>
                                </relocations>
                            </configuration>
                        </execution>
                    </executions>
                </plugin>
            </plugins>
        </build>

这段配置的大体意思就是把需要的依赖jar包打进。

然后

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70 4][]

使用maven进行打包

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70 5][]

这个就是我们需要的最终jar包。

### **五、部署服务** ###

将jar包上传服务器

运行 ： ./bin/flink run \*.jar

![20210511143444694.png][]

运行成功查看检查点状态

hdfs dfs -ls /flink （这里的/flink就是我们的检查点保存的目录）

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70 6][]

根据更新时间发现我们的保存目录

![20210511143735658.png][]

已经进行了790次保存？

### **六、故障恢复** ###

关闭服务运行，测试故障恢复

./bin/flink run -s hdfs://192.168.40.150:9000/flink/a5bcea529bbfff34b9f7af942dcf232f/chk-790 ../flink-healthy-alarm-1.0-SNAPSHOT.jar

\-s 指定路径 到/chk-下即可

[20210511141422683.png]: /images/20221021/4b127153dafc4ccfa9e35897f01bf9c9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70]: /images/20221021/c5187c5f2a6e46f0ae2c7fd4acfd6b03.png
[https_archive.apache.org_dist_flink]: https://archive.apache.org/dist/flink/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70 1]: /images/20221021/3acc63c688c14211a45deec8a97d724c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70 2]: /images/20221021/c2bf9b5041804c58aa396185972b34ae.png
[https_flink.apache.org_zh_downloads.html]: https://flink.apache.org/zh/downloads.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70 3]: /images/20221021/cc7f44155efd4984b5dd7532f4ba3f79.png
[20210511143238585.png]: /images/20221021/f5b3e5c8ec04458d9c7388c172b6e155.png
[20210511142921427.png]: /images/20221021/b00cfe21e26e45c2808f7fc5bea9ed12.png
[2021051114190962.png]: /images/20221021/c10441fab2214e79be0e57622215e2b5.png
[20210511141942578.png]: /images/20221021/9087e2f5e2814a50bd2102b5d2bb639a.png
[20210511142325252.png]: /images/20221021/9269d08258644706a77bba05e9d22fe9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70 4]: /images/20221021/bb3e50eabd9148a4a6577929c9880ae4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70 5]: /images/20221021/e99af54203b24e59a378c25cdb0114d2.png
[20210511143444694.png]: /images/20221021/ec4fd280b5f94390a5610d9971f13807.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70 6]: /images/20221021/e8cae0f08640475b81b222c01f871d82.png
[20210511143735658.png]: /images/20221021/3da30a4f424e4b2fa3ce8946d66edcbe.png