當前位置：首頁 > 编程资源 > 编程问答 >内容正文

编程问答

Flink的滚动策略

發布時間：2023/12/31 编程问答 28 豆豆

生活随笔收集整理的這篇文章主要介紹了 Flink的滚动策略小編覺得挺不錯的,現在分享給大家,幫大家做個參考.

滾動策略是用于checkpoint的

RollingPolicy用于決定數據如何滾動保存，比如文件(保存checkpoint的文件)到達多大或者經過多久就關閉當前文件，開啟下一個新文件保存后續內容。[2]

根據[3]

1).In-progress ：

????當前文件正在寫入中

2).Pending ：

????當處于 In-progress 狀態的文件關閉（closed）了，就變為 Pending 狀態

3).Finished ：

????在成功的 Checkpoint 后，Pending 狀態將變為 Finished 狀態,處于 Finished 狀態的文件不會再被修改，可以被下游系統安全地讀取。

策略

用途

DefaultRollingPolicy

參考[3]
DefaultRollingPolicy rollingPolicy = DefaultRollingPolicy
? ? ? ? .create()
? ? ? ? .withMaxPartSize(1024*1024*120) // 設置每個文件的最大大小 ,默認是128M。這里設置為120M
? ? ? ? .withRolloverInterval(Long.MAX_VALUE) // 滾動寫入新文件的時間，默認60s。這里設置為無限大
? ? ? ? .withInactivityInterval(60*1000) // 60s空閑，就滾動寫入新的文件
? ? ? ? .build();

OnCheckpointRollingPolicy

參考[4]:

"使用BulkEncoding時只能用OnCheckpointRollingPolicy"

代碼案例[5]

? StreamingFileSink<Tuple3<String,String, String>> sink = StreamingFileSink
? ? ? ? ? ? .forRowFormat(new Path(s3://BUCKET),
? ? ? ? ? ? ? ? ? ? (Tuple3<String,String, String> element, OutputStream stream) -> {
? ? ? ? ? ? ? ? ? ? ? ? PrintStream out = new PrintStream(stream);
? ? ? ? ? ? ? ? ? ? ? ? out.println(element.f2);
? ? ? ? ? ? ? ? ? ? })
? ? ? ? ? ? // Determine component type for each record
? ? ? ? ? ? .withBucketAssigner(new CustomBucketAssigner())
? ? ? ? ? ? .withRollingPolicy(OnCheckpointRollingPolicy.build())
? ? ? ? ? ? .withBucketCheckInterval((TimeUnit.MINUTES.toMillis(1)))
? ? ? ? ? ? .build();?

[6]中的描述也可以學習下。

Reference:

[1]Flink源碼分析之深度解讀流式數據寫入hive

[2]Flink入門：StreamingFileSink相關特性及代碼實戰

[3]Flink消費Kafka數據，寫入HDFS - 使用 StreamingFileSink

[4]Flink-StreaimingFileSink-自定義序列化-Parquet批量壓縮

[5]Flink s3 write performance optimization

[6]學習 Flink（十六）：Streaming Parquet File

總結

以上是生活随笔為你收集整理的Flink的滚动策略的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。

策略
Flink

上一篇： kafka的消费隔离级别(持续更新中)
下一篇：一个口罩引发的老黄牛