Kafka過期數據清理


Kafka將數據持久化到了硬盤上,允許你配置一定的策略對數據清理,清理的策略有兩個,刪除和壓縮。

數據清理的方式

刪除

log.cleanup.policy=delete啟用刪除策略
直接刪除,刪除后的消息不可恢復。可配置以下兩個策略:
清理超過指定時間清理:  
log.retention.hours=16
超過指定大小后,刪除舊的消息:
log.retention.bytes=1073741824
為了避免在刪除時阻塞讀操作,采用了copy-on-write形式的實現,刪除操作進行時,讀取操作的二分查找功能實際是在一個靜態的快照副本上進行的,這類似於Java的CopyOnWriteArrayList。

壓縮

將數據壓縮,只保留每個key最后一個版本的數據。
首先在broker的配置中設置log.cleaner.enable=true啟用cleaner,這個默認是關閉的。
在topic的配置中設置log.cleanup.policy=compact啟用壓縮策略。

壓縮策略的細節

如上圖,在整個數據流中,每個Key都有可能出現多次,壓縮時將根據Key將消息聚合,只保留最后一次出現時的數據。這樣,無論什么時候消費消息,都能拿到每個Key的最新版本的數據。
壓縮后的offset可能是不連續的,比如上圖中沒有5和7,因為這些offset的消息被merge了,當從這些offset消費消息時,將會拿到比這個offset大的offset對應的消息,比如,當試圖獲取offset為5的消息時,實際上會拿到offset為6的消息,並從這個位置開始消費。
這種策略只適合特俗場景,比如消息的key是用戶ID,消息體是用戶的資料,通過這種壓縮策略,整個消息集里就保存了所有用戶最新的資料。
壓縮策略支持刪除,當某個Key的最新版本的消息沒有內容時,這個Key將被刪除,這也符合以上邏輯。


免責聲明!

本站轉載的文章為個人學習借鑒使用,本站對版權不負任何法律責任。如果侵犯了您的隱私權益,請聯系本站郵箱yoyou2525@163.com刪除。



 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM