Kafka partitions estimation

Motivation

Choosing the proper number of partitions for a topic is the key to achieving a high degree of parallelism with respect to writes to and reads and to distribute load. Evenly distributed load over partitions is a key factor to have good throughput. The main target is to increase throughput and decrease latency.

We need to answer the question which is "how many partitions are needed to achieve our goal?"

Estimation

?Making a good decision requires estimation based on the desired throughput of producers and consumers per partition.

For example, if you want to be able to read 1 GB/sec, but your consumer is only able process 50 MB/sec, then you need at least 20 partitions and 20 consumers in the consumer group. Similarly, if you want to achieve the same for producers, and 1 producer can only write at 100 MB/sec, you need 10 partitions. In this case, if you have 20 partitions, you can maintain 1 GB/sec for producing and consuming messages. You should adjust the exact number of partitions to number of consumers or producers, so that each consumer and producer achieve their target throughput.

So a simple formula could be:

#Partitions = max(NP, NC)        

where:

• NP?is the number of required producers determined by calculating:?TT/TP.
• NC?is the number of required consumers determined by calculating:?TT/TC.
• TT?is the total expected throughput for our system.
• TP?is the max throughput of a single producer to a single partition.
• TC?is the max throughput of a single consumer from a single partition.

This calculation gives you a rough indication of the number of partitions.

So again how to calculate the throughput of producer and consumer?

Throughput calculation

Kafka is mostly limited by the disk and network throughput.

There are two ways to calculate throughput based on server specs:

1. Based on mathematical formula:

• A very simple rule could be to size the cluster based on the amount of disk-space required (which can be computed from the estimated rate at which you get data times the required data retention period).
• A slightly more sophisticated estimation can be done based on network and disk throughput requirements. To make this estimation, let's plan for a use case with the following characteristics:

- W?- MB/sec of data that will be written.

- R?- Replication factor.

- C?- Number of consumer groups, that is the number of readers for each write.

• The volume of writing expected is?W * R?(that is, each replica writes each message).
• Data is read by replicas as part of the internal cluster replication and also by consumers.
• Because every replicas but the master read each write, the read volume of replication is?(R-1) * W.
• In addition each of the?C?consumers reads each write, so there will be a read volume of?C * W.
• This gives the following:

- Writes:?W * R

- Reads:?(R+C- 1) * W

2. The most accurate way to model your use case is to simulate the load you expect on your own hardware. You can do this using the load generation tools that ship with Kafka whic are ?kafka-producer-perf-test?and?kafka-consumer-perf-test.

• ?To learn more about them, run:

• For producer the flags of most interest for this command are:

Once the test is completed some stats will be printed on terminal, something like

• For consumer the flags of most interest for this command are:

Once the test is completed some stats will be printed on terminal, something like

Critical configurations

There are several factors which affect throughput dramatically which are:

• batch.size in bytes: Does not take effect unless?linger.ms?is non-zero. This lets the producer package messages and send them together, which reduces the number of requests to the cluster.
• linger.ms in milliseconds: Determines how long a message will be buffered in the current batch until the batch is sent. In other words, the producer will send the current batch either when?batch.size?is reached or the?linger.ms?amount of time has passed since the batch started to be filled.
• compression.type: For applications that produce messages of big sizes, compression can help improve the throughput.
• acks: As a business requirement, you might need to replicate messages across your Kafka cluster. In some cases, you might need to acknowledge all replicas; in others, it might be enough to get acknowledgment only from the original node. I also looked at the impact of this configuration.
• CPU limit: The computational power of the client application considerably impacts the throughput from the producer’s perspective.
• session.timeout.ms: Consumers are considered alive if it can send a heartbeat to a broker within this time. Otherwise, the consumer will be considered dead or failed. This will lead to a consumer re-balance. The lower the consumer?session.timeout.ms?the faster we will be able to detect those failures.If the?session.timeout.ms?is too low, a consumer could experience repeated unnecessary re-balances, due to scenarios such as when a batch of messages takes longer to process or when a JVM GC pause takes too long.
• fetch.min.bytes: This parameter defines the minimum bytes expected from a fetch response of a consumer. Increasing this value will reduce the number of fetch requests made to the broker, therefore reducing extra overhead.
• Producer throughput vs stored data: One of the hidden dangers of many messaging systems is that they work well only as long as the data they retain fits in memory. Their throughput falls when data backs up and isn't consumed (and hence needs to be stored on disk). This means things may be running fine as long as your consumers keep up and the queue is empty, but as soon as they lag, the whole messaging layer backs up with unconsumed data. The backup causes data to go to disk which in turns causes performance to drop to a rate that means messaging system can no longer keep up with incoming data and either backs up or falls over. This is pretty terrible, as in many cases the whole purpose of the queue was to handle such a case gracefully.
• Message size: Smaller messages are the harder problem for a messaging system as they magnify the overhead of the records the system does. We can show this by just graphing throughput in both records/second and MB/second as we vary the record size.

• OS tuning: vm.max_map_count?defines maximum number of mmap a process can have.In Apache Kafka, each log segment requires a pair of index/timeindex files, and each of these files consumes 1 mmap. In other words, each log segment uses 2 mmap. Thus, if each partition hosts a single log segment, it requires minimum 2 mmap. The number of log segments per partition varies depending on the?segment size, load intensity, retention policy, rolling period?and, generally tends to be more than one.?Mmap value = 2*((partition size)/(segment size))*(partitions).