前言

前两周在技术讨论群里有同学问对于kafka的topic，如何设置partition的数量？

乍一看，好像挺简单的，甚至很多同学在实践中都没有把这个当成一个问题，随便给个数就完了。

但是老实说，这个问题想回答好，其实并不太容易，因为这里没有一个统一的标准答案，你可以根据当前这个topic所属业务的某个维度来，也可以按照你kafka的broker数量来决定，还可以根据kafka系统默认的分区数量来。

但是有一点必须要搞清楚的是，在建立topic时必须要确定partition的数量(虽然后期可以动态增加)，其根本目的在于：决定后续计算引擎对数据处理的并发度。

因为对于任何一个大数据生态下的存储组件，你如何存数据一定是由你怎么使用数据来决定的，也就是你上游的数据形态必须由下游依赖的处理模式来定。

01

—

kafka的技术架构

虽然是分布式的存储系统，但是kafka却是一个[去中心化的]技术组件，它没有一般分布式系统的master和slaver的角色之分，因此不存在说哪个节点挂掉了，导致整个集群不可用。

因为没有master这个老大，所有的broker都是独立而平等的，他们之间属于一种合作共赢的关系，任何一个kafka集群可以由任意个broker实例组成。

虽然不是一般的master和slaver这种有中心化的架构，但是kafka也并非完全没有中心，只不过它的中心是以partition为单位，每个partition由一个leader和0或多个follower组成，其中这个leader就是中心，是数据写入的入口，这个一旦挂掉，那么这个partition的数据就暂时不可用，需要等待新的leader被选举出来。

每个topic通过若干个partition来分治，核心目的除了分摊单个broker的压力(IO压力、存储压力、连接压力等)外，还有一个很重要的原因是给后续的计算引擎提供一个很好的[并行度]，让数据处理的效率更高。

我们都知道在用分布式计算引擎消费kafka的数据时，都会指定一个[Consumer Group]，那为什么不是单个Consumer，而是Group呢？原因就是在消费Topic数据时(一个或者多个)，一般都有多个partition组成，而我们需要一个partition对应一个consumer，因此消费多个partition的consumer就构成了一个Group。

02

—

根据业务特点确定partition数量

怎么理解呢？我们都知道一个topic肯定是跟某个业务强相关，比如电商业务中的用户登录数据，会用一个专门的topic，而购买数据而会用另外一个topic。

以用户登录数据为例，定多少个partition，可以根据你后续要如何对这个数据进行处理来定。

我们知道partition之间的数据是没有办法保证有序，且没有业务逻辑关系的，默认情况下，当用客户端向某个topic灌数据时，如果没有指定消息的key和要写入的partition，那么数据会以round-robin的方式均匀写到topic的每个partition中。

msg轮询写入

假如这些数据来自全国各地，而你后续要对这个数据根据各个省份的登录情况进行汇总统计，按照这种大家最常见的数据写入方式，那么后续计算引擎对其处理的流程大概是这样的：

分组时因为宽依赖而导致shuffle

这样做最大的坏处就是数据读到计算引擎后，会因为根据省份分组而导致宽依赖的产生，而宽依赖则会导致计算引擎的shuffle操作，而shuffle是计算引擎性能的坟墓，需要尽可能的避免。

但是如果根据省份分组这个业务特点，将你的partition数量设置为省份的个数，且每条数据在写入partition前以当前省份作为消息的key，例如：

key: hubei  value: msg

通过kafka的api就是这样来写：

/**如果添加key，相同key的数据只会推送到同一个partition*/
ProducerRecord record = new ProducerRecord<>(topic, "hubei", msg);

这样，数据就会按照这个key进行hash然后跟partition的个数取模，最终进入到特定的partition中。

通过key路由写入

这样，同一个省份的数据就必然处在同一个partition中，而又因为partition的数量跟省份的数量保持一致。因此，每个partition中的数据都来自某一个省份，后续计算引擎对其分组处理就是这个样子：

分组时依然是窄依赖不会产生shuffle

kafka的分区数据跟计算引擎的分区数据完全一一对应，这样最大的好处在于，当计算引擎读取数据后，对数据进行聚合分组时，因为数据在写入kafka时就已经按业务分好了组，因此当你用同样的的并行度来取数据时，此时的数据是天然按照省份分组的，因此避免了宽依赖的产生，而没有宽依赖就不会有shuffle，数据的处理性能自然大大提升。

因此，通过key路由的方式写入partition，可以大大提高后续数据的处理效率。这个也是我们在实际项目中优化计算速度很重要的一个优化点。

03

—

根据并行度确定partition数量

如果说你的业务处理并没有像上面那样对数据有分组统计的要求，单纯的只需要对数据进行分布式处理就好了。

那么就可以根据你后续数据要处理的并行度来定partition个数，为了防止单个节点负载过高，可以定为跟broker数量一样，或者跟后续计算引擎要并行的数量一致。

这个时候有同学会说，何必那么麻烦，我的partition随便定一个比较小的值就可以了，不用考虑后续数据处理的并行度，因为这个并行度我可以随时改的。

确实可以改，比如拿spark举例，你可以通过reparation和coalesce这两个函数来修改你要想的并行度。但是，你知道这样带来了什么代价吗？

来看下源码的解释：

将原本小的partition数改大，必定会导致宽依赖的产生，而宽依赖则一定会产生shuffle，既然产生shuffle，那数据处理效率也就必然下降。

partition由小变多，必定产生宽依赖

因此，并不是像一些同学所说的没有关系，只是看这个关系会对你产生多大的影响罢了。

04

—

说说分区数与并行度

发现有些同学搞不清楚这两种的区别，这里说说我的理解。

分区数(partition数量)：由数据源决定，是个相对静态的概念，一旦这个数据源决定了，那么你用计算引擎读取数据时候的初始分区数量也就定了，反映的是数据源的划分多少；

并行度：由消费数据源的计算引擎决定，是通过任务提交参数来定义的，是一个相对可调整的数值，是相对动态的，反映的是可以并行同时处理多少数据源的数据；

两者关系：并行度 <= 分区数。

举个栗子，如果你kafka的某个topic有10个partition，那么你的计算引擎并行度就只能小于等于10个，以spark为例，可以这么设置：

//方式1：
--executor-cores 1 --num-executors 10
//方式2：
--executor-cores 2 --num-executors 5

这样设置就可以一次性将你有10个partition的分布式任务处理完成，但是如果你当前机器资源不够，一次性不足以处理完所有的分区，那么也可以这样设置：

//方式3：一次处理不完，分2次的情况
--executor-cores 1 --num-executors 5
//一次处理不完，分5次的情况
--executor-cores 1 --num-executors 2

注意，即便一次处理不完，你对并行度设置的数值最好能被总分区数整除，即10个分区，并行度最好设置为：10、5、2。

避免因为其他数值可能导致在最后一次并行处理时，某些计算节点没有数据造成资源浪费。怎么理解？

假如你将并行度设置为3，那么这个任务就得分4批执行，而到第4批的时候，因为你的并行度还是3，会占用3个计算节点的资源，但是此时你只有1个节点有数据，其他2个是空的。

在群中讨论时，还有同学说并行度是由计算节点的核数来定，这个说法你稍微想一下就会觉得不可能。

试想，如果并行度跟计算节点核数保持一致，那一个分布式计算集群，如果同时运行了几百上千个任务，每个任务都占满了CPU的核心，那CPU的抢占得多么的激烈，上下文切换得多么频繁，效率得多低。

因此，并行度多少，一定是根据数据源分布情况，以及当前集群资源可用情况，外加业务需求情况3者来综合考虑确定。

结语

——

虽然只是一个小小的partition分区数，但是里面其实蕴含了很多跟存储、计算、业务、性能优化等息息相关的技术点。

也可以从这么一件[小事]，来考验设计者的经验和水平。

ITPub博客

如何确定Kafka的partition数

最新文章