本人使用的StreamingFileSink将数据按照行保存到hdfs中 StreamingFileSink streamingFileSink = StreamingFileSink. forRowFormat(new Path(path), new SimpleStringEncoder ("UTF-8")) .withBucketAssigner(bucketAssigner) .withRollingPolicy( DefaultRollingPolicy.builder()
withRolloverInterval(TimeUnit.HOURS.toMillis(1))
withInactivityInterval(TimeUnit.MINUTES.toMillis(30)) withMaxPartSize(1024 * 1024 * 1024) build()) withOutputFileConfig( OutputFileConfig.builder() withPartSuffix(partSuffix) build() ) build(); 配置如上,checkpoint的配置是10分钟一次,现在有个疑惑想要问下,现在hdfs上文件只是在半个小时都是未完成状态, 如 .part-0-11606723036.inprogress.5b46f31b-8289-44e9-ae26-997f3e479446 这种的处于 inprocress状态,但是我这checkpoint是10分钟一次,如果我的任务在29分钟挂了,那么hdfs上这个文件就肯定不是FINISHED状态,那么那20分钟的数据我这应该怎么处理. 我这现在按照默认的处理中,hive对于inprogress的数据是直接过滤掉的,我这把文件改成正常的名称是能读取到*来自志愿者整理的flink邮件归档
你需要使用oncheckpoint的policy,这样在每次Checkpoint时会滚动文件*来自志愿者整理的flink邮件归档
版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。