Kafka中怎么實(shí)現(xiàn)日志存儲(chǔ)

Kafka中怎么實(shí)現(xiàn)日志存儲(chǔ)，針對這個(gè)問題，這篇文章詳細(xì)介紹了相對應(yīng)的分析和解答，希望可以幫助更多想解決這個(gè)問題的小伙伴找到更簡單易行的方法。

1. 日志存儲(chǔ)格式

        最新版本的kafka日志是以批為單位進(jìn)行日志存儲(chǔ)的，所謂的批指的是kafka會(huì)將多條日志壓縮到同一個(gè)batch中，然后以batch為單位進(jìn)行后續(xù)的諸如索引的創(chuàng)建和消息的查詢等工作。對于每個(gè)批次而言，其默認(rèn)大小為4KB，并且保存了整個(gè)批次的起始位移和時(shí)間戳等元數(shù)據(jù)信息，而對于每條消息而言，其位移和時(shí)間戳等元數(shù)據(jù)存儲(chǔ)的則是相對于整個(gè)批次的元數(shù)據(jù)的增量，通過這種方式，kafka能夠減少每條消息中數(shù)據(jù)占用的磁盤空間。這里我們首先展示一下每個(gè)批次的數(shù)據(jù)格式：

        圖中消息批次的每個(gè)元數(shù)據(jù)都有固定的長度大小，而只有最后面的消息個(gè)數(shù)的是可變的。如下是batch中主要的屬性的含義：

• 起始位移：占用8字節(jié)，其存儲(chǔ)了當(dāng)前batch中第一條消息的位移；

• 長度：占用了4個(gè)字節(jié)，其存儲(chǔ)了整個(gè)batch所占用的磁盤空間的大小，通過該字段，kafka在進(jìn)行消息遍歷的時(shí)候，可以快速的跳躍到下一個(gè)batch進(jìn)行數(shù)據(jù)讀??；

• 分區(qū)leader版本號：記錄了當(dāng)前消息所在分區(qū)的leader的服務(wù)器版本，主要用于進(jìn)行一些數(shù)據(jù)版本的校驗(yàn)和轉(zhuǎn)換工作；

• CRC：對當(dāng)前整個(gè)batch的數(shù)據(jù)的CRC校驗(yàn)碼，主要是用于對數(shù)據(jù)進(jìn)行差錯(cuò)校驗(yàn)的；

• 屬性：占用2個(gè)字節(jié)，這個(gè)字段的最低3位記錄了當(dāng)前batch中消息的壓縮方式，現(xiàn)在主要有GZIP、LZ4和Snappy三種。第4位記錄了時(shí)間戳的類型，第5和6位記錄了新版本引入的事務(wù)類型和控制類型；

• 最大位移增量：最新的消息的位移相對于第一條消息的唯一增量；

• 起始時(shí)間戳：占用8個(gè)字節(jié)，記錄了batch中第一條消息的時(shí)間戳；

• 最大時(shí)間戳：占用8個(gè)字節(jié)，記錄了batch中最新的一條消息的時(shí)間戳；

• PID、producer epoch和起始序列號：這三個(gè)參數(shù)主要是為了實(shí)現(xiàn)事務(wù)和冪等性而使用的，其中PID和producer epoch用于確定當(dāng)前producer是否合法，而起始序列號則主要用于進(jìn)行消息的冪等校驗(yàn)；

• 消息個(gè)數(shù)：占用4個(gè)字節(jié)，記錄當(dāng)前batch中所有消息的個(gè)數(shù)；

        通過上面的介紹可以看出，每個(gè)batch的頭部數(shù)據(jù)中占用的字節(jié)數(shù)固定為61個(gè)字節(jié)，可變部分主要是與具體的消息有關(guān)，下面我們來看一下batch中每條消息的格式：

        這里的消息的頭部數(shù)據(jù)就與batch的大不相同，可以看到，其大部分?jǐn)?shù)據(jù)的長度都是可變的。既然是可變的，這里我們需要強(qiáng)調(diào)兩個(gè)問題：

• 對于數(shù)字的存儲(chǔ)，kafka采用的是Zig-Zag的存儲(chǔ)方式，也即負(fù)數(shù)并不會(huì)使用補(bǔ)碼的方式進(jìn)行編碼，而是將其轉(zhuǎn)換為對應(yīng)的正整數(shù)，比如-1映射為1、1映射為2、-2映射為3、2映射為4，關(guān)系圖如下所示：

  

  通過圖可以看出，在對數(shù)據(jù)反編碼的時(shí)候，我們只需要將對應(yīng)的整數(shù)轉(zhuǎn)換成其原始值即可；

• 在使用Zig-Zag編碼方式的時(shí)候，每個(gè)字節(jié)最大為128，而其中一半要存儲(chǔ)正數(shù)，一半要存儲(chǔ)負(fù)數(shù)，還有一個(gè)0，也就是說每個(gè)字節(jié)能夠表示的最大整數(shù)為64，此時(shí)如果有大于64的數(shù)字，kafka就會(huì)使用多個(gè)字節(jié)進(jìn)行存儲(chǔ)，而這多個(gè)字節(jié)的表征方式是通過將每個(gè)字節(jié)的最大位作為保留位來實(shí)現(xiàn)的，如果最高位為1，則表示需要與后續(xù)字節(jié)共同表征目標(biāo)數(shù)字，如果最高位為0，則表示當(dāng)前位即可表示目標(biāo)數(shù)字。

        kafka使用這種編碼方式的優(yōu)點(diǎn)在于，大部分的數(shù)據(jù)增量都是非常小的數(shù)字，因此使用一個(gè)字節(jié)即可保存，這比直接使用原始類型的數(shù)據(jù)要節(jié)約大概七倍的內(nèi)存。

        對于上面的每條消息的格式，除了消息key和value相關(guān)的字段，其還有屬性字段和header，屬性字段的主要作用是存儲(chǔ)當(dāng)前消息key和value的壓縮方式，而header則供給用戶進(jìn)行添加一些動(dòng)態(tài)的屬性，從而實(shí)現(xiàn)一些定制化的工作。        通過對kafka消息日志的存儲(chǔ)格式我們可以看出，其使用batch的方式將一些公共信息進(jìn)行提取，從而保證其只需要存儲(chǔ)一份，雖然看起來每個(gè)batch的頭部信息比較多，但其平攤到每條消息上之后使用的字節(jié)更少了；在消息層面，kafka使用了數(shù)據(jù)增量的方式和Zig-Zag編碼方式對數(shù)據(jù)進(jìn)行的壓縮，從而極大地減少其占用的字節(jié)數(shù)。總體而言，這種存儲(chǔ)方式極大的減少了kafka占用的磁盤空間大小。

2. 日志存儲(chǔ)方式

        在使用kafka時(shí)，消息都是推送到某個(gè)topic中，然后由producer計(jì)算當(dāng)前消息會(huì)發(fā)送到哪個(gè)partition，在partition中，kafka會(huì)為每條消息設(shè)置一個(gè)偏移量，也就是說，如果要唯一定位一條消息，使用<topic, partition, offset>三元組即可。基于kafka的架構(gòu)模式，其會(huì)將各個(gè)分區(qū)平均分配到每個(gè)broker中，也就是說每個(gè)broker會(huì)被分配用來提供一個(gè)或多個(gè)分區(qū)的日志存儲(chǔ)服務(wù)。在broker服務(wù)器上，kafka的日志也是按照partition進(jìn)行存儲(chǔ)的，其會(huì)在指定的日志存儲(chǔ)目錄中為每個(gè)topic的partition分別創(chuàng)建一個(gè)目錄，目錄中存儲(chǔ)的就是這些分區(qū)的日志數(shù)據(jù)，而目錄的名稱則會(huì)以<topic-patition>的格式進(jìn)行創(chuàng)建。如下是kafka日志的存儲(chǔ)目錄示意圖：

        這里我們需要注意的是，圖中對于分區(qū)日志的存儲(chǔ)，當(dāng)前broker只會(huì)存儲(chǔ)分配給其的分區(qū)的日志，比如圖中的connect-status就只有分區(qū)1和分區(qū)4的目錄，而沒有分區(qū)2和分區(qū)3的目錄，這是因?yàn)檫@些分區(qū)被分配在了集群的其他節(jié)點(diǎn)上。在每個(gè)分區(qū)日志目錄中，存在有三種類型的日志文件，即后綴分別為log、index和timeindex的文件。其中l(wèi)og文件就是真正存儲(chǔ)消息日志的文件，index文件存儲(chǔ)的是消息的位移索引數(shù)據(jù)，而timeindex文件則存儲(chǔ)的是時(shí)間索引數(shù)據(jù)。如下圖所示為一個(gè)分區(qū)的消息日志數(shù)據(jù)：

        從圖中可以看出，每種類型的日志文件都是分段的，這里關(guān)于分段的規(guī)則主要有如下幾點(diǎn)需要說明：

• 在為日志進(jìn)行分段時(shí)，每個(gè)文件的文件名都是以該段中第一條消息的位移的偏移量來命名的；

• kafka會(huì)在每個(gè)log文件的大小達(dá)到1G的時(shí)候關(guān)閉該文件，而新開一個(gè)文件進(jìn)行數(shù)據(jù)的寫入?？梢钥吹剑瑘D中除了最新的log文件外，其余的log文件的大小都是1G；

• 對于index文件和timeindex文件，在每個(gè)log文件進(jìn)行分段之后，這兩個(gè)索引文件也會(huì)進(jìn)行分段，這也就是它們的文件名與log文件一致的原因；

• kafka日志的留存時(shí)間默認(rèn)是7天，也就是說，kafka會(huì)刪除存儲(chǔ)時(shí)間超過7天的日志，但是對于某些文件，其部分日志存儲(chǔ)時(shí)間未達(dá)到7天，部分達(dá)到了7天，此時(shí)還是會(huì)保留該文件，直至其所有的消息都超過留存時(shí)間；

3. 索引文件

        kafka主要有兩種類型的索引文件：位移索引文件和時(shí)間戳索引文件。位移索引文件中存儲(chǔ)的是消息的位移與該位移所對應(yīng)的消息的物理地址；時(shí)間戳索引文件中則存儲(chǔ)的是消息的時(shí)間戳與該消息的位移值。也就是說，如果需要通過時(shí)間戳查詢消息記錄，那么其首先會(huì)通過時(shí)間戳索引文件查詢該時(shí)間戳對應(yīng)的位移值，然后通過位移值在位移索引文件中查詢消息具體的物理地址。關(guān)于位移索引文件，這里有兩點(diǎn)需要說明：

• 由于kafka消息都是以batch的形式進(jìn)行存儲(chǔ)，因而索引文件中索引元素的最小單元是batch，也就是說，通過位移索引文件能夠定位到消息所在的batch，而沒法定位到消息在batch中的具體位置，查找消息的時(shí)候，還需要進(jìn)一步對batch進(jìn)行遍歷；

• 位移索引文件中記錄的位移值并不是消息真正的位移值，而是該位移相對于該位移索引文件的起始位移的偏移量，通過這種方式能夠極大的減小位移索引文件的大小。如下圖所示為一個(gè)位移索引文件的格式示意圖：

  

  如下則是具體的位移索引文件的示例：

  

        關(guān)于時(shí)間戳索引文件，由于時(shí)間戳的變化比位移的變化幅度要大一些，其即使采用了增量的方式存儲(chǔ)時(shí)間戳索引，但也沒法有效地使用Zig-Zag方式對數(shù)據(jù)進(jìn)行編碼，因而時(shí)間戳索引文件是直接存儲(chǔ)的消息的時(shí)間戳數(shù)據(jù)，但是對于時(shí)間戳索引文件中存儲(chǔ)的位移數(shù)據(jù)，由于其變化幅度不大，因而其還是使用相對位移的方式進(jìn)行的存儲(chǔ)，并且這種存儲(chǔ)方式也可以直接映射到位移索引文件中而無需進(jìn)行計(jì)算。如下圖所示為時(shí)間戳索引文件的格式圖：

        如下則是時(shí)間戳索引文件的一個(gè)存儲(chǔ)示例：

        可以看到，如果需要通過時(shí)間戳來定位消息，就需要首先在時(shí)間戳索引文件中定位到具體的位移，然后通過位移在位移索引文件中定位到消息的具體物理地址。

4. 日志壓縮

        所謂的日志壓縮功能，其主要是針對這樣的場景的，比如對某個(gè)用戶的郵箱數(shù)據(jù)進(jìn)行修改，其總共修改了三次，修改過程如下：

email=john@gmail.com
email=john@yahoo.com.cn
email=john@163.com

        在這么進(jìn)行修改之后，很明顯，我們主要需要關(guān)心的是最后一次修改，因?yàn)槠涫亲罱K數(shù)據(jù)記錄，但是如果我們按順序處理上述消息，則需要處理三次消息。kafka的日志壓縮就是為了解決這個(gè)問題而存在的，對于使用相同key的消息，其會(huì)只保留最新的一條消息的記錄，而中間過程的消息都會(huì)被kafka cleaner給清理掉。但是需要注意的是，kafka并不會(huì)清理當(dāng)前處于活躍狀態(tài)的日志文件中的消息記錄。所謂當(dāng)前處于活躍狀態(tài)的日志文件，也就是當(dāng)前正在寫入數(shù)據(jù)的日志文件。如下圖所示為一個(gè)kafka進(jìn)行日志壓縮的示例圖：

        圖中K1的數(shù)據(jù)有V1、V3和V4，經(jīng)過壓縮之后只有V4保留了下來，K2的數(shù)據(jù)則有V2、V6和V10，壓縮之后也只有V10保留了下來；同理可推斷其他的Key的數(shù)據(jù)。另外需要注意的是，kafka開啟日志壓縮使用的是log.cleanup.policy，其默認(rèn)值為delete，也即我們正常使用的策略，如果將其設(shè)置為compaction，則開啟了日志壓縮策略，但是需要注意的是，開啟了日志壓縮策略并不代表kafka會(huì)清理歷史數(shù)據(jù)，只有將log.cleaner.enable設(shè)置為true才會(huì)定時(shí)清理歷史數(shù)據(jù)。

        在kafka中，其本身也在使用日志壓縮策略，主要體現(xiàn)在kafka消息的偏移量存儲(chǔ)。在舊版本中，kafka將每個(gè)consumer分組當(dāng)前消費(fèi)的偏移量信息保存在zookeeper中，但是由于zookeeper是一款分布式協(xié)調(diào)工具，其對于讀操作具有非常高的性能，但是對于寫操作性能比較低，而consumer的位移提交動(dòng)作是非常頻繁的，這勢必會(huì)導(dǎo)致zookeeper稱為kafka消息消費(fèi)的瓶頸。因而在最新版本中，kafka將分組消費(fèi)的位移數(shù)據(jù)存儲(chǔ)在了一個(gè)特殊的topic中，即__consumer_offsets，由于每個(gè)分組group的位移信息都會(huì)提交到該topic，因而kafka默認(rèn)為其設(shè)置了非常多的分區(qū)，也即50個(gè)分區(qū)。另外，consumer在提交位移時(shí)，使用的key為groupId+topic+partition，而值則為當(dāng)前提交的位移，也就是說，對于每一個(gè)分組所消費(fèi)的topic的partition，其都只會(huì)保留最新的位移。如果consumer需要讀取位移，那么只需要按照上述格式組裝key，然后在該topic中讀取最新的消息數(shù)據(jù)即可。

關(guān)于Kafka中怎么實(shí)現(xiàn)日志存儲(chǔ)問題的解答就分享到這里了，希望以上內(nèi)容可以對大家有一定的幫助，如果你還有很多疑惑沒有解開，可以關(guān)注億速云行業(yè)資訊頻道了解更多相關(guān)知識(shí)。