kafka工具_kafka工具中如何删除数据

大数据数据采集工具

Sqoop：这个是用于把Mysql里的数据导入到Hadoop里的。

随着大数据技术体系的发展，越来越多的企业应用大数据技术支撑自己的业务发展。数据采集作为大数据的起点，是企业主动获取数据的一种重要手段。数据采集的多样性、全面性直接影响数据质量。

kafka工具_kafka工具中如何删除数据

kafka工具_kafka工具中如何删除数据

企业获取数据的渠道分为内部和外部两个渠道。内部渠道包含自己建设的业务系统，如电商系统、门户网站、门户等。外部渠道包含爬虫系统爬取的数据、三方合作平台数据、公共社交平台数据等。那么如何从这些渠道获取数据？下面简单地介绍一下常用的数据采集工具。

结构化数据采集工具。

1 Apache Flume

支持离线与实时数据导入，是数据集成的主要工具。

2 Apache Sqoop

半结构化数据采集工具

半结构化的数据多见于日志格式。对于日志采集的工具，比较常见的是

1 Logstash

Logstash与ElasticSearch、Kibana并称为ELK，是采集日志的黄金搭档。

2 Apache Flume也多用于日志文本类数据采集。

非结构化数据采集工具

1 DataX

DataX轻量级中间件，在关系型数据库导入导出性能比较优异。支持多种数据类型的导入导出。

流式数据采日志段大小为broker端参数log.segment.bytes(默认为1073741824，1GB)集工具

1 Kafka

性能优异超高吞吐量。

Binlog日志采集工具

基于MySQL数据库增量日志解析提供增量日志和消费功能。

爬虫采集框架与工具

15.5.1 页缓存 Ja栈，Nutch2、WebMagic等。

2 Python栈，Scrapy、PySpider

3 第三方爬虫工具，八爪鱼、爬山虎、后羿等等。

特种卡夫卡的强度、泛用性如何？上下位替代是谁？

保证端到端的压缩，服务端配置compression.type，默认为"producer",表示保留生产者使用的压缩方式，还可以配置为"gzip","snappy","lz4"

练度不用太高，看你怎么定位卡夫卡，如果是当个工具人的话那技能甚至不需要专精，连精英化都没必要，就用个沉默控制。如果是想当个法伤工具人，那需要吧二技能专三，同时至少也要精二40级才算好用。

总结来说还可以，比较中庸的干员，沉默确实不错，但是恐惧并不是非常的好用，二技能的眩晕也算是对策解的定位，所以综合来说是比较鸡肋的，还要到大规模的投入养成成本，泛用度不行，所以综合实力比较鸡肋也就是说，如果需要通过时间戳查询消息记录，那么其首先会通过时间戳索引文件查询该时间戳对应的位移值，然后通过位移值在位移索引文件中查询消息具体的物理地址。关于位移索引文件，这里有两点需要说明：，手感僵硬还有转火，外加补阻挡性价比不高等问题很明显属于双刃剑的类型吧。

《明日方舟》是由鹰角网络自主开发运营的一款策略向即卡夫卡植物急救包中的剂、修剪工具、标记贴、土壤调节剂等可以帮助植物解决遭受虫害、病害、干旱、寒害等问题，并有效帮助植物应对各种环境和生长过程中的突况。此外，该植物急救包中的杀菌剂效果非常快速，能防治各种疾病，对于种植的作物类型没有限制，可以广泛使用。时战略国产手游，于2019年5月1日公测。该游戏适龄级别为16 。

在游戏中，玩家将作为罗德岛的“博士”，带领罗德岛的一众干员救助受难人群、处理矿石争端以及对抗整合运动。在错综复杂的势力博弈之中，寻找治愈矿石病的方法。

深入理解kafka(五)日志存储

5.2.5 v2版本

5.1文件目录布局

4.ANTICIPATORY

根目录下有以下5个checkpoint文件: cleaner-offset-checkpoint, log-start-offset-checkpoint, meta.properties, recovery-point-offset-checkpoint, replication-offset-checkpoint

分区目录下有以下目录: 0000xxx.index(偏移量为64位长整形，长度固定为20位), 0000xxx.log, 0000xxx.timeindex.

还有可能包含.deleted .cleaned .swap等临时文件, 以及可能的.snapshot .txnindex leader-epoch-checkpoint

5.2日志格式演变

5.2.1 v0版本

kafka0.10.0之前

RECORD_OVERHEAD包括offset(8B)和message size(4B)

RECORD包括:

crc32(4B):crc32校验值

magic(1B):消息版本号0

attributes(1B):消息属性。低3位表示压缩类型：0-NONE 1-GZIP 2-SNAPPY 3-LZ4(0.9.x引入)

key length(4B):表示消息的key的长度。-1代表null

key: 可选

value length(4B):实际消息体的长度。-1代表null

value: 消息体。可以为空，如墓碑消息

比v0多了timestamp(8B)字段，表示消息的时间戳

attributes的第4位也被利用起来，0表示timestamp的类型为CreateTime，1表示timestamp的类型为LogAppendTime

timestamp类型由broker端参数来配置，默认为CreateTime，即采用生产者创建的时间戳

多条消息压缩至value字段，以提高压缩率

5.2.4 变长字段

变长整形(Varints):每一个字节都有一个位于位的m位(most significant bit),除了一个字节为1，其余都为0，字节倒序排列

Record Batch

partition leader epoch:

magic:固定为2

attributes:两个字节。低3位表示压缩格式，第4位表示时间戳类型，第5位表示事务(0-非事务1-事务)，第6位控制消息(0-非控制1控制)

max timestamp:

producer id:

producer epoch:

first sequence:

records count:

v2版本的消息去掉了crc字段，另外增加了length(消息总长度)、timestamp delta(时间戳增量)、offset delta(位移增量)和headers信息，并且弃用了attributes

Record

attributes:弃用，但仍占据1B

timestamp delta:

headers:

5.3日志索引

稀疏索引(sparse index):每当写入一定量(broker端参数log.index.interval.bytes指定，默认为4096B),偏移量索引文件和时间索引文件分别对应一个索引项

日志段切分策略：

2.当前日志段消息的时间戳与当前系统的时间戳值大于log.roll.ms或者log.roll.hours，ms优先级高，默认log.roll.hours=168(7天)

4.偏移量值(offset-baseOffset)>Integer.MAX_VALUE

5.3.1 偏移量索引

每个索引项占用8个字节，分为两个部分：1.relativeOffset相对偏移量(4B) 2.ition物理地址(4B)

使用kafka-dump-log.sh脚本来解析.index文件(包括.timeindex、.snapshot、.txnindex等文件)，如下:

bin/kafka-dump-log.sh --files /tmp/kafka-logs/topicId-0/00……00.index

如果broker端参数log.index.size.max.bytes不是8的倍数，内部会自动转换为8的倍数

每个索引项占用12个字节，分为两个部分：1.timestamp当前日志分段的时间戳(12B) 2.relativeOffset时间戳对应的相对偏移量(4B)

如果broker端参数log.index.size.max.bytes不是12的倍数，内部会自动转换为12的倍数

5.4日志清理

broker端参数log.cleanup.policy设置为delete(默认为delete)

检测周期broker端参数log.retention.check.interval.ms=300000(默认5分钟)

1.基于时间

broker端参数log.retention.hours,log.retention.minutes,log.retention.ms,优先级ms>minutes>hours

2.基于日志大小

3.基于日志起始偏移量

DeleteRecordRequest请求

1.KafkaAdmin的deleteRecord()

2.kafka-delete-record.sh脚本

5.4.2 日志压缩

5.5磁盘存储

相关测试：一个由6块7200r/min的RAID-5阵列组成的磁盘簇的线性写入600MB/s，随机写入100KB/s，随机内存写入400MB/s，线性内存3.6GB/s

Linux作系统的vm.dirty_background_ratio参数用来指定页数量达到系统的百分比之后就触发pdflush/flush/kdmflush，一般小于10，不建议为0

vm.dirty_ratio表示页百分比之后刷盘，但是阻塞新IO请求

kafka同样提供同步刷盘及间断性强制刷盘(fsync)功能，可以通过、log.flush.interval.ms等参数来控制

kafka不建议使用swap分区，vm.swappiness参数上限为100，下限为0，建议设置为1

5.5.2 磁盘I/O流程

一般磁盘IO的场景有以下4种：

1.用户调用标准C库进行IO作，数据流为：应用程序Buffer->C库标准IOBuffer->文件系统也缓存->通过具体文件系统到磁盘

3.用户打开文件时使用O_DIRECT，绕过页缓存直接读写磁盘

4.用户使用类似dd工具，并使用direct参数，绕过系统cache与文件系统直接读写磁盘

Linux系统中IO调度策略有4种：

1.NOOP：no operation

2.CF5.5.3 零拷贝Q

3.DEADLINE

指数据直接从磁盘文件到网卡设备中，不需要经应用程序

对linux而言依赖于底层的sendfile()

对ja而言，FileChannal.transferTo()的底层实现就是sendfile()

网页中常见的内容形式有哪些呢？

1 C

我们常常看见的网页开发中有静态网页和页等两种，静态页面就是没有数据交互，数据是固定的，动态页面就是有、数据库等作为工具，实时显示的数据。下面就以数据流相关知识介绍如下：

首先明确说明Kafka不是数据库，它没有schema，也没有表，更没有索引。

1.它仅仅是生产消息流、消费消息流而已。从这个角度来说1.大小超过broker端参数log.segment.bytes配置的值，默认为1073741824(1GB)Kafka的确不像数据库，至少不像我们熟知的关系型数据库。

1、持久性(durability)

我们先从最容易的持久性开始说起，因为持久性最容易理解。在80年代持久性指的是把数据写入到磁带中，这是一种很古老的存储设备，现在应该已经绝迹了。目前实现持久性更常见的做法是将数据写入到物理磁盘上，而这也只能实现单机的持久性。当演进到分布式系统时代后，持久性指的是将数据通过备份机制拷贝到多台机器的磁盘上。很多数据库厂商都有自己的分布式系统解决方案，如GreenPlum和Oracle RAC。它们都提供了这种多机备份的持久性。和它们类似，Apache Kafka天然也是支持这种持久性的，它提供的副本机制在实现原理上几乎和数据库厂商的方案是一样的。

2、原子性(atomicity)

数据库中的原子性和多线程领域内的原子性不是一回事。我们知道在Ja中有2.用户调用文件IO，数据流为：应用程序Buffer->文件系统也缓存->通过具体文件系统到磁盘AtomicInteger这样的类能够提供线程安全的整数作服务，这里的atomicity关心的是在多个线程并发的情况下如何保证正确性的问题。而在数据库领域，原子性关心的是如何应对错误或异常情况，特别是对于事务的处理。如果服务发生故障，之前提交的事务要保证已经持久化，而当前运行的事务要终止(abort)，它执行的所有作都要回滚，最终的状态就好像该事务从未运行过那样。举个实际的例子。

3、隔离性(isolation)

4、一致性(consistency)

说说一致性。按照Kelppmann大神的原话，这是一个很奇怪的属性：在所有ACID特性中，其他三项特性的确属于数据库层面需要实现或保证的，但只有一致性是由用户来保证的。严格来说，它不属于数据库的特性，而应该属于使用数据库的一种方式。坦率说次听到这句话时我本人还是有点震惊的，因为从没有往这个方面考虑过，但仔细想想还真是这么回事。比如刚才的注册用户名的例子中我们要求每个用户名是的。这种一致性约束是由我们用户做出的，而不是数据库本身。数据库本身并不关心或并不知道用户名是否应该是的。针对Kafka而言，这种一致性又意味着什么呢？Kelppmann没有具体展开，

抓取log的方法

根据查询百度文库显示，抓取LOG的方法如下：

1.使用日志分析工具：使用专业的日志分析工具可以快速抓取出LOG文件或LOlength:G文件记录的信息，可以查看数据的细节，如时间、类别和访问者的IP地址等，来分析用户行为，查找问题等。

2.使用代码抓取：可以将抓取应用程序/脚本写入到存储应用程序，以完成对LOG文件信息的类别分析，分页，排序和排序等。

3.使用外部定时任务：如果要定期从远程抓取日志，则可以设置定时任务及定时脚本来完成。在此过程中，可以使用SS删除时先增加.delete后缀，延迟删除根据file.delete.delay.ms(默认60000)配置H（Secure Shell）安全协议、net或其他可用的协议来连接远程，以查看或取得LOG文件。

4.通过WEB控制台抓取：通过WEB控制台可以实时查看日志文件中的信息，而无offset delta:需登录，因此也是抓取LOG文件信息的方便途径。

5.通过单步调试：可以使用多种语言的单步调试器，如debugger、IDE、ddd等，以检查程序状态，查看LOG文件等信息，从而获取相关的信息。

6.通过shell命令：可以使用相应的命令例如“tail/head”等，查看LOG文件，检查即时状态，调试程序等；而“grep”等可以用来搜索LOG文件中相关信息。

7.通过日志系统：日志系统是分析LOG文件信息的工具，可以让用户快速查看日志中的关键信息，方便查找问题。

8.通过Kafka：Kafka是可以处理实时消息流和日志的工具。可以让用户定期将日志发送到Kafka上，从而便捷的抓取log信息。

kafka低版本的怎么用ja查询给定broker上所有的日志目录信息?

5.2.2 v1版本

1. 日志存储格式

版本的kafka日志是以批为单位进行日志存储的，所谓的批指的是kafka会将多条日志压缩到同一个batch中，然后以batch为单位进行后续的诸如索引的创建和消息的查询等工作。

对于每个批次而言，其默认大小为4KB，并且保存了整个批次的起始位移和时间戳等元数据信息，而对于每条消息而言，其位移和时间戳等元数据存储的则是相对于整个批次的元数据的增量，通过这种方式，kafka能够减少每条消息中数据占用的磁盘空间。

这里我们首先展示一下每个批次的数据格式：

图中消息批次的每个元数据都有固定的长度大小，而只有面的消息个数的是可变的。如下是batch中主要的属性的含义：

起始位移：占用8字节，其存储了当前batch中条消息日志清理策略可以控制到主题级别的位移；

分区leader版本号：记录了当前消息所在分区的leader的版本，主要用于进行一些数据版本的校验和转换工作；

CRC：对当前整个batch的数据的CRC校验码，主要是用于对数据进行错校验的；

属性：占用2个字节，这个字段的3位记录了当前batch中消息的压缩方式，现在主要有GZIP、LZ4和Snappy三种。第4位记录了时间戳的类型，第5和6位记录了新版本引入的事务类型和控制类型；

位移增量：的消息的位移相对于条消息的增量；

起始时间戳：占用8个字节，记录了batch中条消息的时间戳；

时间戳：占用8个字节，记录了batch中的一条消息的时间戳；

PID、producer epoch和起始序列号：这三个参数主要是为了实现事务和幂等性而使用的，其中PID和producer epoch用于确定当前producer是否合法，而起始序列号则主要用于进行消息的幂等校验；

消息个数：占用4个字节，记录当前batch中所有消息的个数；

通过上面的介绍可以看出，每个batch的头部数据中占用的字节数固定为61个字节，可变部分主要是与具体的消息有关，下面我们来看一下batch中每条消息的格式：

这里的消息的头部数据就与batch的大不相同，可以看到，其大部分数据的长度都是可变的。既然是可变的，这里我们需要强调两个问题：

1、对于数字的存储，kafka采用的是Zig-Zag的存储方式，也即负数并不会使用补码的方式进行编码，而是将其转换为对应的正整数，比如-1映射为1希望能帮到你，谢谢！、1映射为2、-2映射为3、2映射为4，关系图如下所示：

通过图可以看出，在对数据反编码的时候，我们只需要将对应的整数转换成其原始值即可；

而这多个字节的表征方式是通过将每个字节的位作为保留位来实现的，如果位为1，则表示需要与后续字节共同表征目标数字，如果位为0，则表示当前位即可表示目标数字。

kafka使用这种编码方式的优点在于，大部分的数据增量都是非常小的数字，因此使用一个字节即可保存，这比直接使用原始类型的数据要节约大概七倍的内存。

对于上面的每条消息的格式，除了消息key和value相关的字段，其还有属性字段和header，属性字段的主要作用是存储当前消息key和value的压缩方式，而header则供给用户进行添加一些动态的属性，从而实现一些定制化的工作。

通过对kafka消息日志的存储格式我们可以看出，其使用batch的方式将一些公共信息进行提取，从而保证其只需要存储一份，虽然看起来每个batch的头部信息比较多，但其平摊到每条消息上之后使用的字节更少了；

在消息层面，kafka使用了数据增量的方式和Zig-Zag编码方式对数据进行的压缩，从而极大地减少其占用的字节数。总体而言，这种存储方式极大的减少了kafka占用的磁盘空间大小。

2. 日志存储方式

在使用kafka时，消息都是推送到某个topic中，然后由producer计算当前消息会发送到哪个partition，在partition中，kafka会为每条消息设置一个偏移量，也就是说，如果要定位一条消息，使用三元组即可。

基于kafka的架构模式，其会将各个分区平均分配到每个broker中，也就是说每个broker会被分配用来提供一个或多个分区的日志存储服务。在broker上，kafka的日志也是按照partition进行存储的，其会在指定的日志存储目录中为每个topic的partition分别创建一个目录，目录中存储的就是这些分区的日志数据，而目录的名称则会以的格式进行创建。如下是kafka日志的存储目录示意图：

在每个分区日志目录中，存在有三种类型的日志文件，即后缀分别为log、index和timeindex的文件。其中log文件就是真正存储消息日志的文件，index文件存储的是消息的位移索引数据，而timeindex文件则存储的是时间索引数据。

如下图所示为一个分区的消息日志数据：

在为日志进行分段时，每个文件的文件名都是以该段中条消息的位移的偏移量来命名的；

kafka会在每个log文件的大小达到1G的时候关闭该文件，而新开一个文件进行数据的写入。可以看到，图中除了的log文件外，其余的log文件的大小都是1G；

对于index文件和timeindex文件，在每个log文件进行分段之后，这两个索引文件也会进行分段，这也就是它们的文件名与log文件一致的原因；

kafka日志的留存时间默认是7天，也就是说，kafka会删除存储时间超过7天的日志，但是对于某些文件，其部分日志存储时间未达到7天，部分达到了7天，此时还是会保留该文件，直至其所有的消息都超过留存时间；

3. 索引文件

1、由于kafka消息都是以batch的形式进行存储，因而索引文件中索引元素的最小单元是batch，也就是说，通过位移索引文件能够定位到消息所在的batch，而没法定位到消息在batch中的具置，查找消息的时候，还需要进一步对batch进行遍历；

2、位移索引文件中记录的位移值并不是消息真正的位移值，而是该位移相对于该位移索引文件的起始位移的偏移量，通过这种方式能够极大的减小位移索引文件的大小。

如下图所示为一个位移索引文件的格式示意图：

关于时间戳索引文件，由于时间戳的变化比位移的变化幅度要大一些，其即使采用了增量的方式存储时间戳索引，但也没法有效地使用Zig-Zag方式对数据进行编码，因而时间戳索引文件是直接存储的消息的时间戳数据，

但是对于时间戳索引文件中存储的位移数据，由于其变化幅度不大，因而其还是使用相对位移的方式进行的存储，并且这种存储方式也可以直接映射到位移索引文件中而无需进行计算。如下图所示为时间戳索引文件的格式图：

如下则是时间戳索引文件的一个存储示例：

可以看到，如果需要通过时间戳来定位消息，就需要首先在时间戳索引文件中定位到具体的位移，然后通过位移在位移索引文件中定位到消息的具体物理地址。

4. 日志压缩

所谓的日志压缩功能，其主要是针对这样的场景的，比如对某个用户的邮箱数据进行修改，其总共修改了三次，修改过程如下：

email=john@gmail

kafka的日志压缩就是为了解决这个问题而存在的，对于使用相同key的消息，其会只保留的一条消息的记录，而中间过程的消息都会被kafka cleaner给清理掉。

图中K1的数据有V1、V3和V4，经过压缩之后只有V4保留了下来，K2的数据则有V2、V6和V10，压缩之后也只有V10保留了下来；同理可推断其他的Key的数据。

另外需要注意的是，kafka开启日志压缩使用的是log.cleanup.policy，其默认值为delete，也即我们正常使用的策略，如果将其设置为compaction，则开启了日志压缩策略，但是需要注意的是，开启了日志压缩策略并不代表kafka会清理历史数据，只有将log.cleaner.enable设置为true才会定时清理历史数据。

另外，consumer在提交位移时，使用的key为groupId+topic+partition，而值则为当前提交的位移，也就是说，对于每一个分组所消费的topic的partition，其都只会保留的位移。如果consumer需要读取位移，那么只需要按照上述格式组装key，然后在该topic中读取的消息数据即可。

大数据都是学哪些软件

如下则是具体的位移索引文件的示例：

Ja：只要了解一些基础即可，做大数据不需要很深的Ja技术，学jaSE就相当于有学习大数据。

在传统的关系型数据库中最强的隔离级别通常是指serializability，国内一般翻译成可串行化或串行化。表达的思想就是连接数据库的每个客户端在执行各自的事务时数据库会给它们一个象：仿佛每个客户端的事务都顺序执行的，即执行完一个事务之后再开始执行下一个事务。其实数据库端同时会处理多个事务，但serializability保证了它们就像单独执行一样。举个例子，在一个系统中，每个新用户都需要注册一个的用户名。

Hadoop：这是现在流行的大数据处理平台几乎已经成为大数据的代名词，所以这个是必学的。

ZoLinux：因为大数据相关软件都是在Linux上运行的，所以Linux要学习的扎实一些，学好Linux对你快速掌握大数据相关技术会有很大的帮助，能让你更好的理解hadoop、hive、hbase、spark等大数据软件的运行环境和网络环境配置，能少踩很多坑，学会shell就能看懂脚本这样能更容易理解和配置大数据集群。okeeper：这是个万金油，安装Hadoop的HA的时候就会用到它，以后的Hbase也会用到它。

Mysql：我们学习完大数据的处理了，接下来学习学习小数据的处理工具mysql数据库，因为一会装hive的时候要用到，mysql需要掌握到什么层度那?你能在Linux上把它安装好，运行起来，会配置简单的权限，修改root的密码，创建数据库。

Hive：这个东西对于会SQL语法的来说就是神器，它能让你处理大数据变的很简单

Oozie：既然学会Hive了，我相信你一定需要这个东西，它可以帮你管理你的Hive或者MapRece、Spark脚本，还能检查你的程序是否执行正确。

Hbase：这是Hadoop生态体系中的NOSQL数据库，他的数据是按照key和value的形式存储的并且key是的，所以它能用来做数据的排重，它与MYSQL相比能存储的数据量大很多。

Kafka：这是个比较好用的队列工具。

Spark：它是用来弥补基于MapRece处理数据速度上的缺点。

卡夫卡植物急救包有用吗

结构化数据在分析型的原始数据占比比较大，大部分经过预处理进入数据仓库进一步分析和数据挖掘。常用的数据采集工具有：

因5.3.2 时间戳索引此，卡夫卡植物急救包是有用输出的。

kafka为什么不适合做业务

日志总大小为broke因而在版本中，kafka将分组消费的位移数据存储在了一个特殊的topic中，即__consumer_offsets，由于每个分组group的位移信息都会提交到该topic，因而kafka默认为其设置了非常多的分区，也即50个分区。r端参数log.retention.bytes(默认为-1，表示无穷大)

卡夫卡本身有什么明显的缺点。如果回到几年前，Kafka、0.8版本，也许可以说他缺少了一些功能，比如认证、SSL端到端加密、相对易用的管理和作工具等等；和现役MQ 1历史上积累的工具相比，很多。不过版本的卡夫卡已经弥补了这些问题，各种功能工具也逐渐完备。

如果说卡夫卡有什么缺点的话，那可能就是作维护略显复杂。您可能需要安装许多组件并调整一些作系统参数来调整Kafka。一些好处(如消息重放)需要额外的开发。对于一些一致性控制，卡夫卡选择在底层暴露一些细节，让集成的开发者来控制。不能拆包。对于小团队来说，这样的运维成本有些高。但是对于一个高性能的系统来说，这些作品是理所当然的。

另外，Kafka并不支持Ja的那些“标准消息传递”协议，所以如果应用场景一定要使用这些标准，你就只能和Kafka说再见了。(但是在这里，我必须吐出来。除了抽象层，那些标准协议根本无法解决消息传递系统中的许多实际问题。但是Kafka确实解决了那些问题，比如HA、自动客户端重新连接、自动重复数据删除等。)

————————————————

版权声明:本文为CSDN博主“赵涵老师”原创文章，遵循CC 4.0 BY-SA版权协议。转载请附上原始来源和本声明的链接。

原文链接first offset::

linux 怎样查看kafka的某 topic数据

broker端参数log.cleanup.policy设置为compact，且log.cleaner.enable设置为true(默认为true)

1、创建一个5.4.1 日志删除需要增加备份因子的topic列表的文件，文件格式是json格式的。

2、使用kafka提供的工具拿到上面topic的partions 分布情况，并重定向到文件中。

3、修改ressgintopic.conf 文件的，手动分配新增加的partion 备份因但是需要注意的是，kafka并不会清理当前处于活跃状态的日志文件中的消息记录。所谓当前处于活跃状态的日志文件，也就是当前正在写入数据的日志文件。如下图所示为一个kafka进行日志压缩的示例图：子。

4、通过下面命令执行备份因子扩容过程，bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json。

5、查看kafka的某 topic数据如图。

注意事项：

Kafka的目的是通过Hadoop的并行加载机制来统一线上和离线的消息处理，也是为了通过集群来提供实时的消息。

kafka查询和修改topic的offset

长度：占用了4个字节，其存储了整个batch所占用的磁盘空间的大小，通过该字段，kafka在进行消息遍历的时候，可以快速的跳跃到下一个batch进行数据读取；

注意如果你的kafka设置了zookeeper root，比如为/kafka，那么命令应该改为：

重启相关的应用程序，就可以从设置的offset开始读数据了。

手动更新Kafka存在Zookeeper中的偏移量。我们有时候需要手动将某个主题的偏移量设置成某个值，这时候我们就需要更新Zookeeper中的数据了。Kafka内置为我们提供了修改偏移量的类：kafka.tools.UpdateOffsetsInZK，我们可以通过它修改Zookeeper中某个主题的偏移量，具体作如下：

在不输入参数的情况下，我们可以在kafka中，其本身也在使用日志压缩策略，主要体现在kafka消息的偏移量存储。在旧版本中，kafka将每个consumer分组当前消费的偏移量信息保存在zookeeper中，但是由于zookeeper是一款分布式协调工具，其对于读作具有非常高的性能，但是对于写作性能比较低，而consumer的位移提交动作是非常频繁的，这势必会导致zookeeper成为kafka消息消费的瓶颈。得知kafka.tools.UpdateOffsetsInZK类需要输入的参数。我们的consumer.properties文件配置内容如下：

这个工具只能把Zookkafka主要有两种类型的索引文件：位移索引文件和时间戳索引文件。位移索引文件中存储的是消息的位移与该位移所对应的消息的物理地址；时间戳索引文件中则存储的是消息的时间戳与该消息的位移值。eeper中偏移量设置成earliest或者latest，如下：