hdfs - 为什么重启flink程序后有很多in-progress文件?
问题描述
我使用 flink 消耗 kafka 并将它们以 parquet 格式保存到 hdfs。现在我发现我的目标目录中有很多正在进行的文件,当我重新启动我的 flink 程序时,它们不会作为目标目录中的文件关闭。
我的环境:
val env = StreamExecutionEnvironment.getExecutionEnvironment
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
env.enableCheckpointing(60000)
env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
env.getConfig.registerTypeWithKryoSerializer(classOf[MyMessage],classOf[ProtobufSerializer])
//sinks
val bucketAssigner = new DateTimeBucketAssigner[myCounter]("yyyy-MM-dd", ZoneId.of("Asia/Shanghai"))
val streamingFileSink = StreamingFileSink.
forBulkFormat(path, ParquetAvroWriters.forSpecificRecord(classOf[myCounter]))
.withBucketCheckInterval(60000)
.withBucketAssigner(bucketAssigner).build
-rw-r--r-- 3 Administrator hdfs 1629 2019-08-05 17:06 /user/data/2019-08-05/.part-2-0.inprogress.722265d7-1082-4c84-b70d-da2a08092f5d
-rw-r--r-- 3 Administrator hdfs 1891 2019-08-05 17:07 /user/data/2019-08-05/.part-2-1.inprogress.ac0d8b56-b8f0-4893-9e55-5374b69f16cc
-rw-r--r-- 3 Administrator hdfs 1891 2019-08-05 17:08 /user/data/2019-08-05/.part-2-2.inprogress.a427c2e2-d689-42b8-aa3d-77873c5654f2
-rw-r--r-- 3 Administrator hdfs 1891 2019-08-05 17:09 /user/data/2019-08-05/.part-2-3.inprogress.b5c746e3-354d-4ab3-b1a4-8c6bd88ae430
-rw-r--r-- 3 Administrator hdfs 1891 2019-08-05 16:59 /user/data/2019-08-05/.part-2-3.inprogress.e286d995-3fa7-4696-b51a-27378412a35c
-rw-r--r-- 3 Administrator hdfs 1891 2019-08-05 17:00 /user/data/2019-08-05/.part-2-4.inprogress.bcde4f30-2f78-4f54-92ad-9bc54ac57c5c
-rw-r--r-- 3 Administrator hdfs 1891 2019-08-05 17:10 /user/data/2019-08-05/.part-2-4.inprogress.dbce8a00-6514-43dc-8b31-36c5a8665d37
-rw-r--r-- 3 Administrator hdfs 0 2019-08-05 17:10 /user/data/2019-08-05/.part-2-5.inprogress.34e53418-f5af-4279-87ef-6a27549d90fe
-rw-r--r-- 3 Administrator hdfs 1891 2019-08-05 17:01 /user/data/2019-08-05/.part-2-5.inprogress.936cdb63-4fe2-41bf-b839-2861030c5516
-rw-r--r-- 3 Administrator hdfs 0 2019-08-05 16:55 /user/data/2019-08-05/.part-2-6.inprogress.7a7099a6-9dcd-450b-af2c-8a676276ef0a
-rw-r--r-- 3 Administrator hdfs 0 2019-08-05 17:01 /user/data/2019-08-05/.part-2-6.inprogress.b57f548f-45fc-497c-9807-ef18dba3d11d
-rw-r--r-- 3 Administrator hdfs 1574 2019-08-05 16:56 /user/data/2019-08-05/part-2-0
-rw-r--r-- 3 Administrator hdfs 1868 2019-08-05 16:57 /user/data/2019-08-05/part-2-1
-rw-r--r-- 3 Administrator hdfs 1891 2019-08-05 16:58 /user/data/2019-08-05/part-2-2
-rw-r--r-- 3 Administrator hdfs 1661 2019-08-05 16:53 /user/data/2019-08-05/part-2-3
-rw-r--r-- 3 Administrator hdfs 1891 2019-08-05 16:54 /user/data/2019-08-05/part-2-4
我认为原因是我重新启动程序时正在进行的文件没有关闭,我很困惑为什么重新启动后文件不会关闭,甚至新文件也变成了正在进行中。有人可以解释一下吗?
解决方案
简而言之,对于 Exactly-Once 语义。
请先阅读Flink 官方博客中的这篇文章。
那么让我试着把它解释清楚。
BucketingSink 将所有记录写入临时文件,默认情况下带有后缀 in-progress。
当在这个 sink 上检查点的时间到来时,Flink 会将正在进行的文件的名称保存到检查点;
到了提交时间时,Flink 会将进行中的文件重命名为最终名称,在您的示例中,它们是 part-xx 文件。
并且当你重启 Flink 应用程序时,Flink 作业将从上一个保存点重新启动(如果你设置了参数),并且许多未准备好提交的正在进行的文件将被丢弃,并且永远不会被读取(以点开头的文件不会被HDFS 列表)由用户。
当然,我忽略了很多细节,例如,当文件的体积超过配置时,文件将被重命名为 .pending 文件等。
推荐阅读
- graalvm - 如何设置 GraalVM 原生镜像的堆大小?
- javascript - 需要将对象添加到处于状态的数组中
- c - 是什么原因导致 `c` 中的 `read()` 没有从当前的 `tty` 读取其他进程正在输入的内容?
- react-native - 如何将参数传递给graphql片段?
- java - 在 Oracle Forms 的 jnlp 中创建快捷桌面
- android - 在没有 FileStreams 的 WAL 模式下备份 sqlite db // 通过导出/备份 sql 命令
- android - Android 上的 Unity VideoPlayer 构建:黑屏
- javascript - 用于平移和旋转的 WebGl 位置
- solr - 在 solr 中创建集合时在多个集合之间共享模式
- css - 将所选项目的颜色更改为黑色并在 React Picky 中保持占位符的颜色为灰色