漫游Kafka之过期数据清理
2019獨(dú)角獸企業(yè)重金招聘Python工程師標(biāo)準(zhǔn)>>>
Kafka將數(shù)據(jù)持久化到了硬盤上,允許你配置一定的策略對(duì)數(shù)據(jù)清理,清理的策略有兩個(gè),刪除和壓縮。
數(shù)據(jù)清理的方式
刪除
log.cleanup.policy=delete啟用刪除策略
直接刪除,刪除后的消息不可恢復(fù)。可配置以下兩個(gè)策略:
清理超過(guò)指定時(shí)間清理: ?
log.retention.hours=16
超過(guò)指定大小后,刪除舊的消息:
log.retention.bytes=1073741824
為了避免在刪除時(shí)阻塞讀操作,采用了copy-on-write形式的實(shí)現(xiàn),刪除操作進(jìn)行時(shí),讀取操作的二分查找功能實(shí)際是在一個(gè)靜態(tài)的快照副本上進(jìn)行的,這類似于Java的CopyOnWriteArrayList。
壓縮
將數(shù)據(jù)壓縮,只保留每個(gè)key最后一個(gè)版本的數(shù)據(jù)。
首先在broker的配置中設(shè)置log.cleaner.enable=true啟用cleaner,這個(gè)默認(rèn)是關(guān)閉的。
在topic的配置中設(shè)置log.cleanup.policy=compact啟用壓縮策略。
壓縮策略的細(xì)節(jié)
如上圖,在整個(gè)數(shù)據(jù)流中,每個(gè)Key都有可能出現(xiàn)多次,壓縮時(shí)將根據(jù)Key將消息聚合,只保留最后一次出現(xiàn)時(shí)的數(shù)據(jù)。這樣,無(wú)論什么時(shí)候消費(fèi)消息,都能拿到每個(gè)Key的最新版本的數(shù)據(jù)。
壓縮后的offset可能是不連續(xù)的,比如上圖中沒有5和7,因?yàn)檫@些offset的消息被merge了,當(dāng)從這些offset消費(fèi)消息時(shí),將會(huì)拿到比這個(gè)offset大的offset對(duì)應(yīng)的消息,比如,當(dāng)試圖獲取offset為5的消息時(shí),實(shí)際上會(huì)拿到offset為6的消息,并從這個(gè)位置開始消費(fèi)。
這種策略只適合特俗場(chǎng)景,比如消息的key是用戶ID,消息體是用戶的資料,通過(guò)這種壓縮策略,整個(gè)消息集里就保存了所有用戶最新的資料。
壓縮策略支持刪除,當(dāng)某個(gè)Key的最新版本的消息沒有內(nèi)容時(shí),這個(gè)Key將被刪除,這也符合以上邏輯。
轉(zhuǎn)載于:https://my.oschina.net/sniperLi/blog/807130
總結(jié)
以上是生活随笔為你收集整理的漫游Kafka之过期数据清理的全部?jī)?nèi)容,希望文章能夠幫你解決所遇到的問題。
- 上一篇: C++ TypeId简介与使用
- 下一篇: TextMate 通用快捷键