Flink的滚动策略

约定不等于承诺〃 2022-11-07 05:57 181阅读 0赞

滚动策略是用于checkpoint的

**RollingPolicy**用于决定数据如何滚动保存，比如文件(保存checkpoint的文件)到达多大或者经过多久就关闭当前文件，开启下一个新文件保存后续内容。\[2\]

根据\[3\]

1).In-progress ：

当前文件正在写入中

2).Pending ：

当处于 In-progress 状态的文件关闭（closed）了，就变为 Pending 状态

3).Finished ：

在成功的 Checkpoint 后，Pending 状态将变为 Finished 状态,处于 Finished 状态的文件不会再被修改，可以被下游系统安全地读取。

<table style="width:500px;"> 
 <tbody> 
 <tr> 
 <td>策略</td> 
 <td>用途</td> 
 </tr> 
 <tr> 
 <td>DefaultRollingPolicy</td> 
 <td> 参考[3] DefaultRollingPolicy rollingPolicy = DefaultRollingPolicy &nbsp; &nbsp; &nbsp; &nbsp; .create() &nbsp; &nbsp; &nbsp; &nbsp; .withMaxPartSize(1024*1024*120) // 设置每个文件的最大大小 ,默认是128M。这里设置为120M &nbsp; &nbsp; &nbsp; &nbsp; .withRolloverInterval(Long.MAX_VALUE) // 滚动写入新文件的时间，默认60s。这里设置为无限大 &nbsp; &nbsp; &nbsp; &nbsp; .withInactivityInterval(60*1000) // 60s空闲，就滚动写入新的文件 &nbsp; &nbsp; &nbsp; &nbsp; .build(); </td> 
 </tr> 
 <tr> 
 <td>OnCheckpointRollingPolicy</td> 
 <td> 参考[4]: "使用BulkEncoding时只能用<code>OnCheckpointRollingPolicy</code>" 代码案例[5] &nbsp; StreamingFileSink&lt;Tuple3&lt;String,String, String&gt;&gt; sink = StreamingFileSink &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .forRowFormat(new Path(s3://BUCKET), &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; (Tuple3&lt;String,String, String&gt; element, OutputStream stream) -&gt; { &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; PrintStream out = new PrintStream(stream); &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; out.println(element.f2); &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; }) &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; // Determine component type for each record &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .withBucketAssigner(new CustomBucketAssigner()) &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .withRollingPolicy(OnCheckpointRollingPolicy.build()) &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .withBucketCheckInterval((TimeUnit.MINUTES.toMillis(1))) &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .build();&nbsp; </td> 
 </tr> 
 </tbody> 
</table>

\[6\]中的描述也可以学习下。

**Reference:**

\[1\][Flink源码分析之深度解读流式数据写入hive][Flink_hive]

\[2\][Flink入门：StreamingFileSink相关特性及代码实战][Flink_StreamingFileSink]

\[3\][Flink消费Kafka数据，写入HDFS - 使用 StreamingFileSink][Flink_Kafka_HDFS - _ StreamingFileSink]

\[4\][Flink-StreaimingFileSink-自定义序列化-Parquet批量压缩][Flink-StreaimingFileSink-_-Parquet]

\[5\][Flink s3 write performance optimization][]

\[6\][学习 Flink（十六）：Streaming Parquet File][Flink_Streaming Parquet File]

[Flink_hive]: https://cloud.tencent.com/developer/article/1697918
[Flink_StreamingFileSink]: https://my.oschina.net/u/2380815/blog/4444535
[Flink_Kafka_HDFS - _ StreamingFileSink]: https://blog.csdn.net/magic_kid_2010/article/details/96116321
[Flink-StreaimingFileSink-_-Parquet]: https://blog.csdn.net/baichoufei90/article/details/104749504/
[Flink s3 write performance optimization]: https://stackoverflow.com/questions/62068787/flink-s3-write-performance-optimization
[Flink_Streaming Parquet File]: https://www.codercto.com/a/88093.html