Milvus - Kafka partitioning strategy

[shaiktas]

Hello Team, we are using kafka as a message stream with Milvus. when we logged into broker and watch the log directory. we always observing that only 1 partition is being used by log files though we have overriden broker config (num.partitions = 2) to create 2 partitions. Milvus writing data to only 1 partition the other partition log file is always 0 bytes. In below snapshot only partition 0 is considered by Milvus. Appreciate if you can share which partition strategy is being used by Milvus and also how to obtain Producer and consumer configs? Do we gain any advantage if we increase number of partition in kafka if ingestion rate is too high, the default is 1.

-rw-r--r--. 1 1001 1001     0 Sep 12 01:21 by-dev-rootcoord-dml_8-1/00000000000000000000.log
-rw-r--r--. 1 1001 1001  622K Sep 12 01:31 by-dev-rootcoord-dml_8-0/00000000000000000000.log
-rw-r--r--. 1 1001 1001  947K Sep 12 01:31 by-dev-rootcoord-dml_8-0/00000000000000001963.log

[yhmo]

So far as I know, milvus doesn't specify any partition when creating consumer of kafka. I suppose it always uses the default/first partition.

[elpsycongroo]

I have the same question. I can only find the best practice of kafka in this video and it didn't explain the reason.

In https://milvus.io/docs/data_processing.md it says 'When data ingestion rate reaches bottleneck, consider two things: Whether the log broker node is overloaded and needs to be scaled, and whether there are sufficient shards to ensure load balance for each node.', how can I imporve overall cluster throughput without partitions?

[yhmo]

When milvus is started, it creates 16(default value defined in milvus.yaml) topics in kafka/pulsar.

Let's say we have a collection with shards_num=2. Then the collection will use 2 topics as data channel for the 2 shards.

When you call insert() from the client, the proxy node receives the request. The proxy node splits the data into 2 pieces by hashing the primary key value to different shards. For example, id=[1, 3, 5] are sent to shard_1, id=[2, 4, 6] are sent to shard_2. Once the 2 pieces are successfully stored into kakfa/pulsar, the insert() will return ok to the client.

If you have 2+ data nodes, then the 2 shards are assigned to two data nodes. The 2 data nodes, each create a consumer to consume the data from kafka/pulsar. The 2 data nodes consume the data parallelly. 2 data nodes resources for data processing.
If you have only one data node, the data node creates 2 consumers to consume data from kafka/pulsar. 1 data node resources for data processing.

As you can see, the data process is a two-stage process. In most cases, the first stage is not the bottleneck, data is accumulated in kafka/pulsar and wait to be consumed. Sometimes you have inserted 10 million entities, but only 5 million entities are consumed by datanode/querynode.
Usually the second stage is the bottleneck because the data node needs to handle more tasks such as persisting data into s3, compacting segments, generating bloom filter, etc.

Once the collection is created, the shards_num cannot be changed. Ideally, you should define a proper shards_num for the collection according to the data size of this collection, so that the data is processed by multiple data nodes.

If your collection finally has only 1000 entities, it is a kind of waste to set shards_num=4 for this collection.
But if you want to insert 1 billion entities to this collection, then it is better to set shards_num=4 or 8, and deploy 4 or 8 data nodes to consume data parallelly.

In my opinion, shards_num=1 is fine for millions of entities in one collection. shards_num=2 or 4 is a good value for tens of million entities in one collection. shards_num=8 or 16 is a good value for billions of entities in one collection.

[elpsycongroo]

Thank you for your detailed explanation. Okay, seems that 'shard' is a concept like other distribute systems. The throughput of kafka is mostly not the bottleneck, so milvus not use multi-partition strategy, if we configure num.partitions > 1 in kafka, it's only a waste of storage resource.
Why is it said that setting multiple shards for a small collection is a waste? If our resource is enough, isn't it possible to fully utilize the idle resources of multiple data nodes?

[yhmo]

Milvus uses a time-tick machinery to maintain data consistency across components(datanode/querynode):
https://milvus.io/docs/time_sync.md

The rootcoord sends a time-tick message to pulsar/kafka for each 200ms.
For each shard of each partition, milvus creates two consumers to consume data, one consumer in datanode, the other consumer in querynode(if the collection is loaded). Once a time-tick message is consumed by all consumers, we say "the data at this time point is consistent in all shards".

With more shards, more consumers are created, all the consumers are consuming the time-tick messages, some cpu resource is wasted in processing the time-tick. You will see the entire milvus service getting slowly and slowly, with more collections/partitions and more shards created.

[shaiktas]

Thank you so much @yhmo for this good information. So with this it is clear that even if we configure multiple partitions in kafka topics it do not have any impact as Milvus uses single parition stragely to publish message. Do Milvus have any impact if we change kafka log file size. By default the max log file size in kafka broker's is 1 GB. Assume huge data ingestion (Billion in volume /TB's in size) with this data if we increase kafka log file size to higher value (5GB) will it help Milvus data node/query node to process fast?

[yhmo]

In most cases kafka is not the bottleneck. The bottleneck is in the datanode/querynode to process to data, including data persistence, segments compaction, build bloomfilter, etc. So, I don't think it can affect querynode/datanode performance by increasing kafka log file.

[xiaofan-luan]

kafka is only related to write throughput.

usually we use multiple topics instead of using multiple partitions.

To use multiple topics, you need to specify shard numbers when create colleciton