|NO.Z.00061|——————————|BigDataEnd|——|Hadoop&kafka.V46|-------------------------------------------|ka

一、日志存储

### --- [kafka高级特性解析]

~~~     [日志存储索引]
~~~     [偏移量索引]
~~~     [时间戳索引]

### --- 索引

~~~     偏移量索引文件用于记录消息偏移量与物理地址之间的映射关系。
~~~     时间戳索引文件则根据时间戳查找对应的偏移量。

~~~     # 文件：
~~~     # 查看一个topic分区目录下的内容，发现有log、index和timeindex三个文件：

~~~     log文件名是以文件中第一条message的offset来命名的，
~~~     实际offset长度是64位，但是这里只使用了20位，应付生产是足够的。
~~~     一组index+log+timeindex文件的名字是一样的，并且log文件默认写满1G后，
~~~     会进行logrolling形成一个新的组合来记录消息，
~~~     这个是通过broker端log.segment.bytes =1073741824指定的。
~~~     index和timeindex在刚使用时会分配10M的大小，当进行log rolling 后，它会修剪为实际的大小。

[root@hadoop01 ~]# ll /opt/yanqi/servers/kafka/kafka-logs/topic_x-1/
total 0
-rw-r--r-- 1 root root 10485760 Sep 24 02:42 00000000000000000000.index
-rw-r--r-- 1 root root        0 Sep 24 02:42 00000000000000000000.log
-rw-r--r-- 1 root root 10485756 Sep 24 02:42 00000000000000000000.timeindex
-rw-r--r-- 1 root root        0 Sep 24 02:42 leader-epoch-checkpoint

二、日志存储索引

### --- 创建主题：

[root@hadoop01 ~]# kafka-topics.sh --zookeeper localhost:2181/myKafka \
--create --topic tp_demo_05 --partitions 1 --replication-factor 1 \
--config segment.bytes=104857600

### --- 创建消息文件：

~~~     # 创建文件
[root@hadoop01 ~]# for i in `seq 10000000`; do echo "hello yanqi $i" >> nmm.txt;
done

=~~~        # 查看创建的文件内容
[root@hadoop01 ~]# cat nmm.txt
hello yanqi 1
~~~省略部分输出
hello yanqi 10000000

### --- 将文本消息生产到主题中：

[root@hadoop01 ~]# kafka-console-producer.sh --broker-list localhost:9092 \
--topic  tp_demo_05 <nmm.txt

### --- 查看存储文件：

[root@hadoop01 ~]# ll /opt/yanqi/servers/kafka/kafka-logs/tp_demo_05-0/
-rw-r--r-- 1 root root     51256 Sep 24 02:58 00000000000000000000.index
-rw-r--r-- 1 root root 104856091 Sep 24 02:58 00000000000000000000.log
-rw-r--r-- 1 root root     76884 Sep 24 02:58 00000000000000000000.timeindex
-rw-r--r-- 1 root root     51264 Sep 24 02:58 00000000000003925358.index
-rw-r--r-- 1 root root 104844831 Sep 24 02:58 00000000000003925358.log
-rw-r--r-- 1 root root        10 Sep 24 02:58 00000000000003925358.snapshot
-rw-r--r-- 1 root root     76896 Sep 24 02:58 00000000000003925358.timeindex
-rw-r--r-- 1 root root  10485760 Sep 24 02:59 00000000000007809212.index
-rw-r--r-- 1 root root  59140429 Sep 24 02:59 00000000000007809212.log
-rw-r--r-- 1 root root        10 Sep 24 02:58 00000000000007809212.snapshot
-rw-r--r-- 1 root root  10485756 Sep 24 02:59 00000000000007809212.timeindex
-rw-r--r-- 1 root root         8 Sep 24 02:57 leader-epoch-checkpoint

### --- 如果想查看这些文件，可以使用kafka提供的shell来完成，几个关键信息如下：

~~~     offset是逐渐增加的整数，每个offset对应一个消息的偏移量。
~~~     position：消息批字节数，用于计算物理地址。
~~~     CreateTime：时间戳。
~~~     magic：2代表这个消息类型是V2，如果是0则代表是V0类型，1代表V1类型。
~~~     compresscodec：None说明没有指定压缩类型，
~~~     kafka目前提供了4种可选择，0-None、1-GZIP、2-snappy、3-lz4。
~~~     crc：对所有字段进行校验后的crc值。

[root@hadoop01 ~]# cd /opt/yanqi/servers/kafka/kafka-logs/tp_demo_05-0/
[root@hadoop01 tp_demo_05-0]# kafka-run-class.sh kafka.tools.DumpLogSegments \
--files 00000000000000000000.log --print-data-log | head
~~~输出参数
Dumping 00000000000000000000.log
Starting offset: 0
baSEOffset: 0 lastOffset: 704 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionleaderEpoch: 0 isTransactional: false position: 0 CreateTime: 1632423425826 isvalid: true size: 16374 magic: 2 compresscodec: NONE crc: 1772470647
baSEOffset: 705 lastOffset: 1384 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionleaderEpoch: 0 isTransactional: false position: 16374 CreateTime: 1632423425916 isvalid: true size: 16366 magic: 2 compresscodec: NONE crc: 3253031656
baSEOffset: 1385 lastOffset: 2054 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionleaderEpoch: 0 isTransactional: false position: 32740 CreateTime: 1632423426019 isvalid: true size: 16371 magic: 2 compresscodec: NONE crc: 1829057350
baSEOffset: 2055 lastOffset: 2736 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionleaderEpoch: 0 isTransactional: false position: 49111 CreateTime: 1632423426059 isvalid: true size: 16365 magic: 2 compresscodec: NONE crc: 3938174312
baSEOffset: 2737 lastOffset: 3418 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionleaderEpoch: 0 isTransactional: false position: 65476 CreateTime: 1632423426092 isvalid: true size: 16365 magic: 2 compresscodec: NONE crc: 3649012870
baSEOffset: 3419 lastOffset: 4100 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionleaderEpoch: 0 isTransactional: false position: 81841 CreateTime: 1632423426132 isvalid: true size: 16365 magic: 2 compresscodec: NONE crc: 2502224461
baSEOffset: 4101 lastOffset: 4782 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionleaderEpoch: 0 isTransactional: false position: 98206 CreateTime: 1632423426157 isvalid: true size: 16365 magic: 2 compresscodec: NONE crc: 3342412428
baSEOffset: 4783 lastOffset: 5464 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionleaderEpoch: 0 isTransactional: false position: 114571 CreateTime: 1632423426202 isvalid: true size: 16365 magic: 2 compresscodec: NONE crc: 2744927020

三、关于消息偏移量索引：

### --- 消息存储

~~~     消息内容保存在log日志文件中。
~~~     消息封装为Record，追加到log日志文件末尾，采用的是顺序写模式。
~~~     一个topic的不同分区，可认为是queue，顺序写入接收到的消息。
~~~     消费者有offset。下图中，消费者A消费的offset是9，消费者B消费的offset是11，
~~~     不同的消费者offset是交给一个内部公共topic来记录的。

~~~     时间戳索引文件，它的作用是可以让用户查询某个时间段内的消息，
~~~     它一条数据的结构是时间戳（8byte）+相对offset（4byte），
~~~     如果要使用这个索引文件，首先需要通过时间范围，找到对应的相对offset，
~~~     然后再去对应的index文件找到position信息，
~~~     然后才能遍历log文件，它也是需要使用上面说的index文件的。
~~~     但是由于producer生产消息可以指定消息的时间戳，
~~~     这可能将导致消息的时间戳不一定有先后顺序，因此尽量不要生产消息时指定时间戳。

===============================END===============================

Walter Savage Landor:strove with none,for none was worth my strife.Nature I loved and, next to Nature, Art:I warm'd both hands before the fire of life.It sinks, and I am ready to depart ——W.S.Landor

来自为知笔记(Wiz)

|NO.Z.00061|——————————|BigDataEnd|——|Hadoop&kafka.V46|-------------------------------------------|ka

相关推荐