如何刪除kafka積壓數(shù)據(jù)？卡夫卡可以通過(guò)兩種方式刪除數(shù)據(jù)根據(jù)時(shí)間，刪除一段時(shí)間后過(guò)期的消息根據(jù)消息大小，消息數(shù)超過(guò)一定大小后刪除最早的數(shù)據(jù)卡夫卡刪除數(shù)據(jù)的最小單位：segment卡夫卡刪除數(shù)據(jù)的主要

如何刪除kafka積壓數(shù)據(jù)？

卡夫卡可以通過(guò)兩種方式刪除數(shù)據(jù)

根據(jù)時(shí)間，刪除一段時(shí)間后過(guò)期的消息

根據(jù)消息大小，消息數(shù)超過(guò)一定大小后刪除最早的數(shù)據(jù)

卡夫卡刪除數(shù)據(jù)的最小單位：segment

卡夫卡刪除數(shù)據(jù)的主要邏輯：卡夫卡源代碼

def cleanuplogs（）{debug（”beging log cleanup。。。）var總計(jì)=0 val開(kāi)始時(shí)間=時(shí)間.毫秒For（log

Kafka在一段時(shí)間內(nèi)（配置文件設(shè)置）調(diào)用cleanuplogs一次，刪除所有需要?jiǎng)h除的日志數(shù)據(jù)。

Cleanupexpiredsegments負(fù)責(zé)清除超時(shí)數(shù)據(jù)

private def Cleanupexpiredsegments（log:log）：int={val startms=時(shí)間.毫秒log.deleteOldSegments文件（開(kāi)始時(shí)間->上次修改時(shí)間>log.config.retentions保留)}

cleanupsegmenttomaintainsize負(fù)責(zé)清理大于大小的數(shù)據(jù)私有def cleanupsegmentstomaintainsize（log:log）：int={if（log.config.retentionSize文件=0）{差異-=段.尺寸真}否則{假}log.deleteOldSegments文件（shouldDelete）}

我沒(méi)事，來(lái)這里玩，開(kāi)始在各種網(wǎng)絡(luò)上尋找技術(shù)信息，之后以“標(biāo)題”為主。從尋找信息到交朋友。因?yàn)槲矣X(jué)得事情落后于時(shí)代，有人認(rèn)為，是因?yàn)樽约核讲桓?。只是在心里想，無(wú)法實(shí)現(xiàn)現(xiàn)實(shí)

JAVA面試如何保證消息不被重復(fù)消費(fèi)？如何保證消息消費(fèi)的冪等性？

基于接收器的實(shí)現(xiàn)將使用kakfa的高級(jí)消費(fèi)API。與所有其他接收器一樣，接收到的數(shù)據(jù)將保存到執(zhí)行器，然后sparkstreaming將啟動(dòng)作業(yè)來(lái)處理數(shù)據(jù)。

在默認(rèn)配置中，如果出現(xiàn)故障，此方法將丟失數(shù)據(jù)。為了確保零數(shù)據(jù)丟失，我們需要啟用wal（writeaheadlogs）。它將接收到的數(shù)據(jù)同步保存到分布式文件系統(tǒng)，如HDFS。因此，在發(fā)生錯(cuò)誤時(shí)可以恢復(fù)數(shù)據(jù)。

使用兩個(gè)步驟：1。添加依賴項(xiàng)：Spark streaming Kafka 2.10-1.3.0

2導(dǎo)入器g.apache.spark. 卡夫卡._