一起体验 Milvus 2.0 新功能

Zilliz RDS

发布于 2022-02-21 15:39:17

1.2K0

发布于 2022-02-21 15:39:17

✏️ 编者按：

在历经 9 个 RC 版本的迭代与全球 1000 家用户的实战验证后，Milvus 2.0 正式 GA！Zilliz 质量保障团队负责人乔燕良撰文解析新功能、新亮点。 Milvus 2.0 背后，有哪些开发秘辛？文末预约直播，与开发者们在线畅聊！

经过半年多的等待，Milvus 2.0 正式版终于和大家见面了。现在就让我们一起来体验一下 Milvus 2.0 的一些新功能吧！

删除功能（Entity Deletion)

Milvus 2.0 新增了向量删除功能，允许用户通过向量 id 将某些向量从 Collection 中删除。用户再也不用为一些过期或无效的向量数据犯愁了。赶紧试一下：

1. 准备数据：插入 300 条 128 维的向量数据

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

# connect to milvus
host = 'x.x.x.x'
connections.add_connection(default={"host": host, "port": 19530})
connections.connect(alias='default')

# create a collection with customized primary field: id_field
dim = 128
id_field = FieldSchema(name="cus_id", dtype=DataType.INT64, is_primary=True)
age_field = FieldSchema(name="age", dtype=DataType.INT64, description="age")
embedding_field = FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=dim)
schema = CollectionSchema(fields=[id_field, age_field, embedding_field],
                          auto_id=False, description="hello MilMil")
collection_name = "hello_milmil"
collection = Collection(name=collection_name, schema=schema)

import random
# insert data with customized ids
nb = 300
ids = [i for i in range(nb)]
ages = [random.randint(20, 40) for i in range(nb)]
embeddings = [[random.random() for _ in range(dim)] for _ in range(nb)]

entities = [ids, ages, embeddings]
ins_res = collection.insert(entities)
print(f"insert entities primary keys: {ins_res.primary_keys}")

insert entities primary keys: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49, 50, 51, 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65, 66, 67, 68, 69, 70, 71, 72, 73, 74, 75, 76, 77, 78, 79, 80, 81, 82, 83, 84, 85, 86, 87, 88, 89, 90, 91, 92, 93, 94, 95, 96, 97, 98, 99, 100, 101, 102, 103, 104, 105, 106, 107, 108, 109, 110, 111, 112, 113, 114, 115, 116, 117, 118, 119, 120, 121, 122, 123, 124, 125, 126, 127, 128, 129, 130, 131, 132, 133, 134, 135, 136, 137, 138, 139, 140, 141, 142, 143, 144, 145, 146, 147, 148, 149, 150, 151, 152, 153, 154, 155, 156, 157, 158, 159, 160, 161, 162, 163, 164, 165, 166, 167, 168, 169, 170, 171, 172, 173, 174, 175, 176, 177, 178, 179, 180, 181, 182, 183, 184, 185, 186, 187, 188, 189, 190, 191, 192, 193, 194, 195, 196, 197, 198, 199, 200, 201, 202, 203, 204, 205, 206, 207, 208, 209, 210, 211, 212, 213, 214, 215, 216, 217, 218, 219, 220, 221, 222, 223, 224, 225, 226, 227, 228, 229, 230, 231, 232, 233, 234, 235, 236, 237, 238, 239, 240, 241, 242, 243, 244, 245, 246, 247, 248, 249, 250, 251, 252, 253, 254, 255, 256, 257, 258, 259, 260, 261, 262, 263, 264, 265, 266, 267, 268, 269, 270, 271, 272, 273, 274, 275, 276, 277, 278, 279, 280, 281, 282, 283, 284, 285, 286, 287, 288, 289, 290, 291, 292, 293, 294, 295, 296, 297, 298, 299]

2. 在删除向量之前，先通过 search 和 query 来验证待删除的向量是存在的，并且做二次查询来验证这个结果是可靠的。

# search
nq = 10
search_vec = [[random.random() for _ in range(dim)] for _ in range(nq)]
search_params = {"metric_type": "L2", "params": {"nprobe": 16}}
limit = 3
# search 2 times to verify the vector persists
for i in range(2):
    results = collection.search(search_vec, embedding_field.name, search_params, limit)
    ids = results[0].ids
    print(f"search result ids: {ids}")
    expr = f"cus_id in {ids}"
    # query to verify the ids exist
    query_res = collection.query(expr)
    print(f"query results: {query_res}")

search result ids: [76, 2, 246]
query results: [{'cus_id': 246}, {'cus_id': 2}, {'cus_id': 76}]
search result ids: [76, 2, 246]
query results: [{'cus_id': 246}, {'cus_id': 2}, {'cus_id': 76}]

3. 删除查询结果中的第一个向量（id=76)

print(f"trying to delete one vector: id={ids[0]}")
collection.delete(expr=f"cus_id in {[ids[0]]}")

results = collection.search(search_vec, embedding_field.name, search_params, limit)
ids = results[0].ids
print(f"after deleted: search result ids: {ids}")
expr = f"cus_id in {ids}"
# query to verify the id exists
query_res = collection.query(expr)
print(f"after deleted: query res: {query_res}")
print("completed")

trying to delete one vector: id=76
after deleted: search result ids: [76, 2, 246]
after deleted: query res: [{'cus_id': 246}, {'cus_id': 2}, {'cus_id': 76}]
completed

4. 居然发现被删除向量（id=76）在删除后还能被 search 和 query 到，这是为什么呢？查看了下代码，发现删除操作是异步的，同时为了提高效率，并没有在数据层真正的做“删除”操作，而是做了删除标记。等待数秒之后再去 query 一下，发现 id=76 的向量已经不可见了。成功！

expr = f"cus_id in {[76, 2, 246]}"
# query to verify the id exists
query_res = collection.query(expr)
print(f"after deleted: query res: {query_res}")
print("completed")

after deleted: query res: [{'cus_id': 246}, {'cus_id': 2}]
completed

那么有没有办法在删除后立刻查询不可见呢？当然——设置一致性等级。

一致性等级 (Consistency_level)

上面删除的实验中，我们看到：某个向量被删除后，如果立刻去查询该被删除向量还是可能会被查询到的，其实这是由于一致性等级（consistency_level）所决定的。

Milvus 用户可以根据自己的业务场景的不同，灵活配置不同的一致性等级。目前 2.0 支持四种不同的一致性等级：(目前一致性等级设置都是 Collection 级别的，而且只能在新建 Collection 的时候设置，之后不能修改，这显然还不够灵活，后期需要继续优化。)

强一致性(CONSISTENCY_STRONG)：GuaranteeTs 设为系统最新时间戳，QueryNodes 需要等待 ServiceTime 推进到当前最新时间戳才能执行该 Search 请求；
最终一致性(CONSISTENCY_EVENTUALLY)：GuaranteeTs 设为一个特别小的值（比如说设为 1），跳过一致性检查，立刻在当前已有数据上执行 Search 查询；
有界一致性(CONSISTENCY_BOUNDED)：GuaranteeTs 是一个比系统最新时间稍旧的时间，在可容忍范围内可以立刻执行查询；
客户端一致性(CONSISTENCY_SESSION)：客户端使用上一次写入的时间戳作为 GuaranteeTs，那么每个客户端至少能看到自己插入的全部数据。

在 2.0 正式版之前，Milvus 默认为强一致性；从 2.0 正式版开始，Milvus 默认为有界一致性。这主要是考虑到绝大多数场景下，相比于响应速度，用户对数据一致性的要求是比较低的。因此使用有界一致性可以更大限度地平衡用户对响应速度和数据一致性的要求（具体设计：https://github.com/milvus-io/milvus/blob/master/docs/developer_guides/how-guarantee-ts-works-cn.md）。

那么，有界一致性的查询速度真的会更快么？老规矩，实验一下：

1. 简单修改一下上面例子中 search 的代码，查询 5 次分别记录一下查询延迟

for i in range(5):
    start = time.time()
    results = collection.search(search_vec, embedding_field.name, search_params, limit)
    end = time.time()
    print(f"search latency: {round(end-start, 4)}")
    ids = results[0].ids
    print(f"search result ids: {ids}")

2. 相同的数据规模和查询参数，设置为 CONSISTENCY_STRONG 的 Collection：

collection_name = "hello_milmil_consist_strong"
collection = Collection(name=collection_name, schema=schema,
                        consistency_level=CONSISTENCY_STRONG)

search latency: 0.3293
search latency: 0.1949
search latency: 0.1998
search latency: 0.2016
search latency: 0.198
completed

3. 设置为 CONSISTENCY_BOUNDED 的 Collection（2.0 正式版后的默认值）：

collection_name = "hello_milmil_consist_bounded"
collection = Collection(name=collection_name, schema=schema,
                        consistency_level=CONSISTENCY_BOUNDED)

search latency: 0.0144
search latency: 0.0104
search latency: 0.0107
search latency: 0.0104
search latency: 0.0102
completed

4. 非常明显，设置了有界一致性的查询响应速度都变快了 200ms 左右。

那么上面删除实验中，如果设置了强一致性，是不是被删除的向量就立刻会查询不可见呢？答案是肯定的。不妨自己实验一下吧。

动态加载索引（Handoff）

在 Streaming 场景下，很多用户会先对 collection 做 build index 和 load 操作，然后再插入数据，但在Milvus 2.0 之前这并不能保证 QueryNode 加载的一定是索引。因此需要在建完索引后，手动重新加载一下 collection，才能将内存中加载的原始向量替换成索引，麻烦而且耗时。2.0 推出 Handoff 功能：当一个 segment 从 growing 变为 sealed 的时候，如果该 segment 的 Collection 已经建立了索引，Milvus 会等待这个 segment 建完索引后再加载带索引的 sealed segment 来替换原来的 growing segment，这样能大大提高查询性能。

1. 再改动一下代码，在插入数据之前，先建立索引并启动加载向量：

# index
index_params = {"index_type": "IVF_SQ8", "metric_type": "L2", "params": {"nlist": 64}}
collection.create_index(field_name=embedding_field.name, index_params=index_params)
# load
collection.load()

2. 每次插入 5 万条向量，循环插入 200 次（为了方便，我插入了相同的向量，但不影响此实验结果）

import random
# insert data with customized ids
nb = 50000
ids = [i for i in range(nb)]
ages = [random.randint(20, 40) for i in range(nb)]
embeddings = [[random.random() for _ in range(dim)] for _ in range(nb)]
entities = [ids, ages, embeddings]
for i in range(200):
    ins_res = collection.insert(entities)
    print(f"insert entities primary keys: {ins_res.primary_keys}")

3. 在插入过程中和结束后查询 QueryNode 加载向量的情况：

# I did this in another python console
>>> utility.get_query_segment_info("hello_milmil_handoff")
[segmentID: 430640405514551298
collectionID: 430640403705757697
partitionID: 430640403705757698
mem_size: 394463520
num_rows: 747090
index_name: "_default_idx"
indexID: 430640403745079297
nodeID: 7
state: Sealed
, segmentID: 430640405514551297
collectionID: 430640403705757697
partitionID: 430640403705757698
mem_size: 397536480
num_rows: 752910
index_name: "_default_idx"
indexID: 430640403745079297
nodeID: 7
state: Sealed
...

4. 所有加载到 QueryNode 的 Sealed segments 都是带索引的（有 index_name 的）：）