Jmeter压测Milvus单条数据插入性能时，发现QPS稳定在162左右

[yangrz7]

1. 数据插入代码

func (c *MilvusClient) Insert(dim int, collectionName string, data *VectorData) (bool, error) {
	exist, err := c.HasCollection(collectionName)
	if err != nil {
		return false, err
	}

	if exist {
		ctx := context.Background()
		
		imgIds := []string{data.ImageId}
		fields := []string{data.Field}
		stamps := []int64{data.TimeStamp}
		vectors := [][]float32{data.Vector}

		//idColumn := entity.NewColumnInt64("ID", ids)
		imgIdColumn := entity.NewColumnVarChar("ImgId", imgIds)
		fieldColumn := entity.NewColumnVarChar("Field", fields)
		stampColumn := entity.NewColumnInt64("TimeStamp", stamps)
		vectorColumn := entity.NewColumnFloatVector("Vector", dim, vectors)
		_, err = c.MilvusCli.Insert(ctx, collectionName, "", imgIdColumn, stampColumn, fieldColumn, vectorColumn)
		c.Log.Info("insert success")
		if err != nil {
			c.Log.Error("insert collection: ", err.Error())
			return false, err
		}
	} 
	return true, nil
}

2. 使用Jmeter压测发现QPS稳定在162左右, 尝试扩容服务节点和my-release-milvus-datanode节点发现都无法提升。

3. Milvus配置：

[yhmo]

Milvus的insert是按如下流程执行：

1. client发送insert请求
2. proxy接收insert请求，将数据写入pulsar/kafka，写入后立刻返回，告知client该请求已完成。此时，数据并未写入milvus的存储系统中。
3. querynode和datanode从pulsa/kafka订阅数据，querynode拿到数据后存放于内存中。datanode拿到数据后，先存放于内存，如果数据积累足够大，则将数据写入milvus的存储系统

由于在第2步，insert请求返回结果给客户端，所以客户端的插入耗时只包含从1到2的耗时。
而1和2的耗时仅跟这几个因素相关：网络传输耗时，proxy处理耗时，pulsar/kafka处理耗时。所以，客户端的插入性能也取决于这几个因素。
网络传输耗时就不说了。
proxy如果是瓶颈，则通过增加proxy节点可以少许提升性能。如果增加proxy的数量无法再提高插入性能，则pulsar/kafka就是瓶颈。基本上，插入性能达到pulsar/kafka的瓶颈之后，就很难再提升。

[yangrz7]

有日志可以看到插入的耗时吗？我没用过pulsar,但是kafka仅插入应该有挺高的qps

[yhmo]

proxy日志里看insert请求的耗时可以这么看：

• 先是这条表示proxy接收到insert请求
  [2023/04/08 03:22:20.236 +00:00] [DEBUG] [proxy/impl.go:2525] ["Start processing insert request in Proxy"] [traceID=312d07449c9e02aa]

• 然后把请求放进任务队列（排队执行，量大的时候未必会立即执行）。我这里是插入10000行512维的数据
  [2023/04/08 03:22:20.236 +00:00] [DEBUG] [proxy/impl.go:2588] ["Enqueue insert request in Proxy"] [role=proxy] [db=] [collection=demo] [partition=_default] [len(FieldsData)=1] [len(HashKeys)=0] [NumRows=10000] [traceID=312d07449c9e02aa]

• 当任务开始执行后，先是做一些准备工作。我这里是第一次insert，所以准备时间比较长。下面这几个耗时都是在task_insert.go打出的以"tr/“打头的日志
  [2023/04/08 03:22:20.246 +00:00] [DEBUG] [proxy/task_insert.go:436] ["tr/proxy execute insert 440644177816453122"] [msg="get collection id & partition id from cache"] [duration=2.2439ms]
  [2023/04/08 03:22:20.274 +00:00] [DEBUG] [proxy/task_insert.go:442] ["tr/proxy execute insert 440644177816453122"] [msg="get used message stream"] [duration=28.149ms]
  [2023/04/08 03:22:20.351 +00:00] [DEBUG] [proxy/task_insert.go:469] ["tr/proxy execute insert 440644177816453122"] [msg="assign segment id"] [duration=76.4009ms]

• 然后是把数据送进pulsar/kafka中
  [2023/04/08 03:22:20.459 +00:00] [DEBUG] [proxy/task_insert.go:476] ["tr/proxy execute insert 440644177816453122"] [msg="send insert request to dml channel"] [duration=108.3331ms]

• 整个insert任务执行完成的总耗时
  [2023/04/08 03:22:20.461 +00:00] [DEBUG] [proxy/task_insert.go:481] ["tr/proxy execute insert 440644177816453122"] [msg="insert execute done"] [duration=217.0929ms]

• insert请求返回结果给客户端
  [2023/04/08 03:22:20.464 +00:00] [DEBUG] [proxy/impl.go:2645] ["Finish processing insert request in Proxy"] [traceID=312d07449c9e02aa]

你可以查看一下你的proxy日志，先看一下每个insert请求的总耗时

[yangrz7]

我先把proxy节点扩容了，但是发现QPS并没有上升。
查看proxy日志，整个insert任务执行完成的总耗时基本也是在20ms左右。
[2023/04/09 06:43:44.248 +00:00] [DEBUG] [proxy/task_insert.go:465] ["tr/proxy execute insert 440669994789109825"] [msg="insert execute done"] [duration=35.083477ms]
[2023/04/09 06:43:44.248 +00:00] [DEBUG] [proxy/task_insert.go:420] ["tr/proxy execute insert 440669994803003440"] [msg="get collection id & partition id from cache"] [duration=2.685µs]
[2023/04/09 06:43:44.248 +00:00] [DEBUG] [proxy/task_insert.go:426] ["tr/proxy execute insert 440669994803003440"] [msg="get used message stream"] [duration=18.757µs]
[2023/04/09 06:43:44.248 +00:00] [DEBUG] [proxy/task_insert.go:436] ["send insert request to virtual channels"] [traceID=1c69b9c9b8ba2ce4] [collection=testVectors] [partition=_default] [collection_id=440530021492337066] [partition_id=440530021492337067] [virtual_channels="[by-dev-rootcoord-dml_4_440530021492337066v0]"] [task_id=440669994803003440]
[2023/04/09 06:43:44.250 +00:00] [INFO] [proxy/impl.go:356] ["HasCollection received"] [traceID=d98cc341803d81a] [role=proxy] [db=] [collection=testVectors]
[2023/04/09 06:43:44.251 +00:00] [INFO] [proxy/impl.go:387] ["HasCollection enqueued"] [traceID=d98cc341803d81a] [role=proxy] [MsgID=440669994803003442] [BeginTS=440669994803003442] [EndTS=440669994803003442] [db=] [collection=testVectors]
[2023/04/09 06:43:44.251 +00:00] [INFO] [proxy/impl.go:417] ["HasCollection done"] [traceID=d98cc341803d81a] [role=proxy] [MsgID=440669994803003442] [BeginTS=440669994803003442] [EndTS=440669994803003442] [db=] [collection=testVectors]
[2023/04/09 06:43:44.252 +00:00] [INFO] [proxy/impl.go:356] ["HasCollection received"] [traceID=cad0f8b2409729] [role=proxy] [db=] [collection=testVectors]
[2023/04/09 06:43:44.252 +00:00] [INFO] [proxy/impl.go:387] ["HasCollection enqueued"] [traceID=cad0f8b2409729] [role=proxy] [MsgID=440669994803003444] [BeginTS=440669994803003444] [EndTS=440669994803003444] [db=] [collection=testVectors]
[2023/04/09 06:43:44.252 +00:00] [INFO] [proxy/impl.go:417] ["HasCollection done"] [traceID=cad0f8b2409729] [role=proxy] [MsgID=440669994803003444] [BeginTS=440669994803003444] [EndTS=440669994803003444] [db=] [collection=testVectors]
[2023/04/09 06:43:44.252 +00:00] [DEBUG] [proxy/task_insert.go:453] ["tr/proxy execute insert 440669994803003440"] [msg="assign segment id"] [duration=4.514666ms]
[2023/04/09 06:43:44.252 +00:00] [DEBUG] [proxy/task_insert.go:460] ["tr/proxy execute insert 440669994803003405"] [msg="send insert request to dml channel"] [duration=17.988378ms]
[2023/04/09 06:43:44.252 +00:00] [DEBUG] [proxy/task_insert.go:465] ["tr/proxy execute insert 440669994803003405"] [msg="insert execute done"] [duration=22.321635ms]

[xiaofan-luan]

有尝试过增加proxy的数目或者提高客户端的并发么？

[yangrz7]

我等会尝试下扩容proxy后在压测一轮

[xiaofan-luan]

我等会尝试下扩容proxy后在压测一轮

现在是增加并发不影响吞吐么？有可能批量插入么？

[yangrz7]

目前业务可以批量插入，但是有些还是还是要单条插入

[xiaofan-luan]

多大并发，看到你的描述里，单个写入只要20ms，一个线程理论上就可以实现50tps了。
我认为这个可能是python客户端层面的性能问题，你现在测试是一个python sdk多线程？

[yangrz7]

服务端使用了Go SDK, jmeter设置了10线程。

[xiaofan-luan]

另外，你不应该每次写入的时候都调用hasColleciton，这个会造成额外的RPC开销，影响性能

[yangrz7]

我移除了hasColleciton()，发现QPS还是上不去。

[yangrz7]

你们有这样压测过吗？我尝试直接在腾讯云环境下压也是这样。

[xiaofan-luan]

我移除了hasColleciton()，发现QPS还是上不去。

你试过jmeter开为10或者更高么，我们应该是压到过比这个更高的性能的。不过延迟是相对确定的

[yangrz7]

试过100线程，tps达到300左右，我们预计至少需要1500+ qps

[xiaofan-luan]

试过100线程，tps达到300左右，我们预计至少需要1500+ qps

那我估计你要用更大并发或者更大batch了，milvus那边应该还没到瓶颈，kafka produce的延迟瓶颈

[yangrz7]

我增加了线程数，看到压测的时候写入耗时800ms～1s左右了，想问下写入pulsar是同步写入还是异步写入。增加pulsar配置或者扩容pod数目会有提升吗？如果切换到kafka会不会好一点？ @xiaofan-luan

[yangrz7]

那你直接用我这个脚本对你那边的k8s集群测试，显示的qps是多少？

好嘞，我试试

[yangrz7]

那你直接用我这个脚本对你那边的k8s集群测试，显示的qps是多少？

用你的脚本在k8s集群压测：
50 threads totally performed 4091 insert requests, 202.3 req/second
100 threads totally performed 4327 insert requests, 211.4 req/second
200 threads totally performed 4220 insert requests, 201.5 req/second
到200左右就上不去了

我在我自己笔记本也用docker-compose搭建了服务，稍微高一点，但是也基本稳定在了270左右：
50 threads totally performed 5045 insert requests, 247.7 req/second
100 threads totally performed 5498 insert requests, 270.0 req/second
200 threads totally performed 5805 insert requests, 280.9 req/second
300 threads totally performed 5763 insert requests, 273.3 req/second

是什么配置导致了性能上不去吗？

[yhmo]

我没什么特别的配置，是用官网上的2.2.5 docker-compose.yml来起的服务。可能是我的机器综合性能好一点
50 threads totally performed 6979 insert requests, 347.5 req/second
100 threads totally performed 10329 insert requests, 511.9 req/second
200 threads totally performed 11486 insert requests, 565.0 req/second
300 threads totally performed 11274 insert requests, 548.8 req/second

[yhmo]

可以考虑在插入环节做一层缓存，所有的客户端插入的数据先进入这个缓存，然后设置两个条件调用milvus的insert()

1. 当缓存的大小超过某个值，比如4MB时，调用milvus insert()，并清空缓存
2. 间隔一段时间(比如间隔1秒)检查，如果缓存里有数据，调用milvus insert()，并清空缓存

这样做的好处是把大量零碎的写入，合并成一块块写入，减少后续环节的压力。不过这个缓存层可能要考虑宕机时数据会丢的问题。

[yangrz7]

我没什么特别的配置，是用官网上的2.2.5 docker-compose.yml来起的服务。可能是我的机器综合性能好一点 50 threads totally performed 6979 insert requests, 347.5 req/second 100 threads totally performed 10329 insert requests, 511.9 req/second 200 threads totally performed 11486 insert requests, 565.0 req/second 300 threads totally performed 11274 insert requests, 548.8 req/second

好吧。我是M1，机器性能应该也不差才对。

[yhmo]

我用python写一个简易的压测，代码如下。
做法：

1. 建一个新表，只有id和vector两列，向量维度768
2. 准备好一条向量，所有的insert调用都插入这条向量
3. 启动若干个线程，每个线程反复调用insert，持续时间20秒
4. 统计插入数，检查表里的行数是否一致

import datetime
import random
import time
from datetime import timedelta
import threading

from pymilvus import (
    connections,
    FieldSchema, CollectionSchema, DataType,
    Collection,
    utility,
)

_HOST = '127.0.0.1'
_PORT = '19530'

_DIM = 768
_COLLECTION_NAME = "demo"
_AUTO_ID = False
_SHARD_NUM = 2

def create_collection():
    connections.connect(host=_HOST, port=_PORT)
    if utility.has_collection(_COLLECTION_NAME):
        utility.drop_collection(_COLLECTION_NAME)

    field1 = FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=_AUTO_ID)
    field2 = FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=_DIM)
    schema = CollectionSchema(fields=[field1, field2])
    collection = Collection(name=_COLLECTION_NAME, schema=schema, shards_num=_SHARD_NUM)
    print("\ncollection created:", _COLLECTION_NAME)


def check_rowcount():
    connections.connect(host=_HOST, port=_PORT)
    collection = Collection(name=_COLLECTION_NAME)
    collection.flush()
    print("Collection row count:", collection.num_entities)


if __name__ == '__main__':
    create_collection()

    # prepare the data
    one_vector = [random.random() for _ in range(_DIM)]

    all_done = {"done": 0}
    done_lock = threading.Lock()

    def do_insert(end_time: datetime.datetime):
        pname = threading.current_thread().name
        print("Thread '{}' start".format(pname))
        try:
            # use process name as connection name
            connections.connect(host=_HOST, port=_PORT, alias=pname)
        except Exception as e:
            print("Thread '{}' failed to create connection, error: {}"
                  .format(pname, e))
            return

        collection = Collection(name=_COLLECTION_NAME, using=pname)

        if _AUTO_ID:
            data = [
                [one_vector],
            ]
        else:
            data = [
                [random.randint(0, 100)],
                [one_vector],
            ]

        start = time.time()
        done = 0
        while True:
            dt = datetime.datetime.now()
            if dt >= end_time:
                break

            try:
                collection.insert(data)
                done = done + 1
            except Exception as e:
                continue

        end = time.time()
        done_lock.acquire()
        all_done["done"] = all_done["done"] + done
        done_lock.release()
        print("Thread '{:s}' finished, insert requests done: {:d}, {:.1f} req/second"
              .format(pname, done, done/(end-start)))

    # duration: seconds
    def run_threads(thread_count: int, duration: int):
        end_time = datetime.datetime.now() + timedelta(seconds=duration)
        thread_list = []
        for i in range(thread_count):
            t = threading.Thread(target=do_insert, name="t-"+str(i), args=(end_time,))
            t.start()
            thread_list.append(t)
        for t in thread_list:
            t.join()

    thread_num = 50
    duration = 20
    start = time.time()
    run_threads(thread_num, duration)
    end = time.time()

    print("{:d} threads totally performed {:d} insert requests, {:.1f} req/second"
          .format(thread_num, all_done["done"], all_done["done"]/(end-start)))

    # check collection's row count
    check_rowcount()

我在本地用docker-compose起了一个milvus cluster，直接用官网上的docker-compose.yml
使用10线程，100+ tps
使用30线程，200+ tps
使用50线程，400+ tps
使用100线程，500+ tps
使用200线程，500+ tps
后面再增加线程数也没法增加tps了

[UFOXD]

你这个压测结果的图片看不到

[xiaofan-luan]

看起来，写入是有一些瓶颈，接下来我们会花时间优化写入的瓶颈