向量检索库Milvus架构及数据处理流程

背景

搜索或推荐场景，需要将非结构化的物料（媒资）结构化，也即提取特征，然后将特征存储向量数据库，从而实现海量数据快速检索功能。

当前，开源市场比较火的搜索引擎有Faiss，但Faiss更类似于es的lucene，需要上层解决分布式水平扩容、数据一致性、高可用等问题。所以对于数据量大，要求高可用等架构场景，使用milvus。

milvus想做的事

Lucene——Faiss
Milvus——Elasticsearch
专注向量检索框架，解决数据一致性，分布式水平扩容等问题

设计思想：

1. CAP中选择去牺牲一定的一致性，来实现可用性和 Latency
2. 日志即数据，流批一体

做一个数据库，而不是引擎。如何做管理、计费、可视化，数据迁移。数据库不仅要提供传统的增删改查能力，还提供数据转换、迁移、多租户加密管理、计费、限流、可视化、备份快找等更加多样的服务

• 做数据分片
• 如何保证数据的高可靠性
• 如何保证分布式系统有节点出现异常时如何恢复
• 如何在一个大规模集群中实现负载均衡
• 如何查询语句
• 如何做 Parse 和 Optimize
• 系统做持久化存储，需要考量不同的数据存储格式

milvus之前——向量检索的一些基础

近似算法

欧式距离

各个点的具体坐标数值对结果会有比较大的影响。在推荐系统场景下，欧式距离一般用于需要从维度的数值大小中体现差异的相关度分析
例如以登陆次数和平均观看时长作为特征时，余弦相似度会认为（1，10）、（10，100）两个用户距离很近，但显然这两个用户的活跃度是有着很大差异的，（10，100）这个用户的价值更高，此时我们更关注数值绝对差异，应当使用欧氏距离

余弦距离

跟欧式距离的差别主要在于它对具体数值的差异并不敏感。一句话总结就是，虽然数值上确实有差异，但是两者的x，y轴相对应的数值的分值之差保持相近，所以两者的相似度还是很高。余弦相似度更倾向于衡量两者在方向趋势上的差异，余弦相似度更多的适用于使用用户对内容评分来区分兴趣的相似度和差异

常见向量索引

1） FLAT

也就是大家常说的暴力搜索，这种方式是典型的牺牲性能和成本换取准确性，是唯一可以实现 100% 召回率的方式，同时可以较好地使用显卡等异构硬件加速。

2） Hash based

基于 locality sensitive hashing 将数据分到不同的哈希桶中。这种方式实现简单，性能较高，但是召回率不够理想。

3） Tree based

代表是 KDTree 或者 BallTree，通过将高维空间进行分割，并在检索时通过剪枝来减少搜索的数据量，这种方式性能不高，尤其是在维度较高时性能不理想。

4） 基于聚类的倒排

通过 k-means 算法找到数据的一组中心点，并在查询时利用查询向量和中心点距离选择部分桶进行查询。倒排这一类又拥有很多的变种，比如可以通过 PCA 将数据进行降维，进行标量量化，或者通过乘积量化 PQ 将数据降精度，这些都有助于减少系统的内存使用和单次数据计算量。

5） NSW（Navigable Small World）图

是一种基于图存储的数据结构，这种索引基于一种朴素的假设，通过在构建图连接相邻的友点，然后在查询时不断寻找距离更近的节点实现局部最优。在 NSW 的基础上，HNSW（Navigable Small World）图借鉴了跳表的机制，通过层状结构构建了快速通道，提升了查询效率。
hnsw参考：https://www.pinecone.io/learn/series/faiss/hnsw/

k-means动态算法：
https://www.naftaliharris.com/blog/visualizing-k-means-clustering/
dbscan动态算法：
https://www.naftaliharris.com/blog/visualizing-dbscan-clustering/

向量数据库对比

相比较其他向量数据库，Milvus：

• 支持的索引类型较多
• 代码开源，社区比较活跃，生态良好（工具）
• GO语言实现，性能高
• 流批一体的设计模式，很好的解决了数据一致性、高可用等问题
  
  
  

https://zhuanlan.zhihu.com/p/364923722
https://www.jianshu.com/p/43cc19426113

milvus架构

milvus的四大角色和十一组件

四大角色

• Access layer：主要功能验证请求参数和合并返回结果
• Coordinator service： 如系统大脑，分配任务；包括集群拓扑管理、负载均衡、时间戳生成、数据声明和数据管理等
• Worker nodes： 执行具体工作节点
• Storage：数据存储和持久化

十一组件

• proxy：验证请求参数和合并返回结果
• Root coordinator：处理DDL和DCL请求，如创建（删除）collection、partition、index，以及TSO (timestamp Oracle)管理
• Query coordinator ：管理查询节点的拓扑结构和负载均衡，以及将growing的segmend切换到sealed
• Data coordinator：管理数据节点的拓扑结构，维护元数据，并触发刷新、压缩和其他后台数据操作；如1）分配 segment 数据2）记录分配空间及其过期时间3）Segment flush 逻辑 4）哪些 channel 被哪些 Data Node 消费则需要 data coord 来做一个整体的分配
• Index coordinator：管理索引结点的拓扑结构，建立索引，并维护索引元数据。
• Data node：订阅日志代理获取增量日志数据，处理变更请求，将日志数据打包成日志快照，并存储在对象存储中。
• Index node：建立索引文件，存储对象存储中
• Query node： 订阅日志代理检索增量日志数据，将它们转化为growing segments，从对象存储加载历史数据，并在向量数据和标量数据之间运行混合搜索。
• Meta storage（etcd）：存储了诸如collection schema、节点状态、消息消费检查点等元数据的快照。此外，Milvus还使用etcd进行服务注册和健康检查
• Object storage：存储日志的快照文件、标量数据和矢量数据的索引文件以及中间查询结果。
• Log broker：负责数据流的持久化、可靠异步查询的执行、事件通知以及查询结果的返回，还在Worker节点从系统故障中恢复时，确保增量数据的完整性。

proxy和其他系统所有主要组件的交互

milvus的数据模型

milvus属性和关系数据库类比

database：类比关系数据库database， 2.2.9之后支持；为多租户，一个租户一个database设计
collection：类比关系数据库表
Entity： 是传统数据库里面“一行”的概念
Field：字段

创建一个collection

# We're going to create a collection with 3 fields.
# +-+------------+------------+------------------+------------------------------+
# | | field name | field type | other attributes |       field description      |
# +-+------------+------------+------------------+------------------------------+
# |1|    "pk"    |   VarChar  |  is_primary=True |      "primary field"         |
# | |            |            |   auto_id=False  |                              |
# +-+------------+------------+------------------+------------------------------+
# |2|  "random"  |    Double  |                  |      "a double field"        |
# +-+------------+------------+------------------+------------------------------+
# |3|"embeddings"| FloatVector|     dim=8        |  "float vector with dim 8"   |
# +-+------------+------------+------------------+------------------------------+
fields = [FieldSchema(name="pk", dtype=DataType.VARCHAR, is_primary=True, auto_id=False, max_length=100),FieldSchema(name="random", dtype=DataType.DOUBLE),FieldSchema(name="embeddings", dtype=DataType.FLOAT_VECTOR, dim=dim)
]schema = CollectionSchema(fields, "hello_milvus is the simplest demo to introduce the APIs")print(fmt.format("Create collection `hello_milvus`"))
hello_milvus = Collection("hello_milvus", schema, consistency_level="Strong")参考：
https://raw.githubusercontent.com/milvus-io/pymilvus/master/examples/hello_milvus.py

shard、partition和segment

• shard：提升写能力。有的文档也称channel，类似 Kafka 中的 topic。Shard 是指将数据写入操作分散到不同节点上，使 Milvus 能充分利用集群的并行计算能力进行写入。
• partition：提升读能力。MMS通过partition key区分libId
• segment ：整个系统调度的最小单元，分为 Growing Segment 和 Sealed Segment

DML：任何传入的插入/删除请求都根据主键的哈希值被路由到shard，默认情况下是两个 Shard，推荐 Shard 的规模做到 Data Node 的两到三倍。
DDL：仅分享一个shard。

virtual channel VS physical channel

• collection 在创建时可以指定 shard 的数目，一个 shard 代表一个 virtual channel
• 将消息存储系统中的 channel 称之为 physical channel

一个 proxy 都会对应所有的 VChannel
多个 V channel 可以对应到同一个 PChannel
一个data node/query node对应多个PChannel

collection 级别的 VChannel可以很多，而且不同 collection 之间也可以共用 PChannel；从而利用消息系统高并发特性提高吞吐量。

https://zhuanlan.zhihu.com/p/517553501?utm_id=0

segment

Segment 在内存中的状态有 3 种，分别是 growing、sealed 和 flushed。 Growing：当新建了一个 segment 时就是 growing 的状态，它在一个可分配的状态。 Sealed：Segment 已经被关闭了，它的空间不可以再往外分配。 Flushed：Segment 已经被写入磁盘
Growing segment 内部的空间可以分为三部份：

• Used （已经使用的空间）：已经被 data node 消费掉。
• Allocated：Proxy 向 Data coord deletor 去请求 segment 分配出的空间。
• Free：还没有用到的空间。

Sealed segment 表示这个 segment 的空间不可以再进行分配。有几种条件可以 seal 一个 segment：

1. 空间使用了达到上限（75%）。
2. 收到 flush collection 要把这个 collection 里面所有的数据都持久化，这个 segment 就不能再分配空间了。
3. Segment 存活时间太长。
4. 太多 growing segment 会导致 data node 内存使用较多，进而强制关闭存活时间最久的那一部分 segment。

数据存储

minio中数据存储

1. insert_log
   bucketName/file/insert_log/ collectionId/ partitionId/ segmentId/ field_ids
   featureId: 100
   libId: 101
   feature: 102
   

2. index_files
   bucketName/file/index_files/ index build id/IndexTaskVersion/ partitionId/ segmentId/index file
   

3. delta_log
   bucketName/file/delta_log/ collectionId/ partitionId/ segmentId/unique ID
   

4. stats_log
   bucketName/file/stats_log/ collectionId/ partitionId/ segmentId/field_id
   

文件内部内容

@TODO

Binlog 里面分成了很多 event，每个 event 都会有两部分，一个是 event header 和 event data。Event header 存的就是一些元信息，比如说创建时间、写入节点 ID、event length 和 NextPosition（下个 event 的偏移量）

INSERT_EVENT 的 event data 固定的部分主要有三个，StartTimestamp、EndTimestamp 和 reserved。Reserved 也就是保留了一部分空间来扩展这个 fixed part。 Variable part 存的就是实际的插入数据。我们把这个数据序列化成一个 parquet 的形式存到这个文件里

https://zhuanlan.zhihu.com/p/486971488

milvus一些限制

https://milvus.io/docs/limitations.md

数据流向

Create Collection

1. 会请求RootCoood，组织好格式，将数据存储etcd
2. 会组织成Msg格式，发送消息队列
   

Flush Collection

主要内容：1）将segment 由growing改为sealed状态，数据不可再写入 2）将数据持久化到Object storage

两个问题：

1. sealed segments可能还在内存，没有持久化
   解决：通过定期调用GetSegmentInfo请求DataCoord，直到所有sealed segments flushed
2. DataCoord 对sealed segments不再分配，但如何确认所有分配的都被DataNode消费了
   解决：1）DataCoord收到冻结后应该会记录当前的ts位点
   2）DataNode从MsgStream消费package时会向DataCoord 发送DataNodeTtMsg报告timestamp位点
   3）DataCoord后台线程解析该请求，判断是否已经消费到冻结的位点

https://github.com/milvus-io/milvus/blob/master/docs/design_docs/20211109-milvus_flush_collections.md

Insert Data

1. 请求proxy，进行参数检验
2. Proxy向RootCoord请求Timestamp（全局时钟）
3. Proxy向DataCoord批量请求entities的segments以及primary keys
4. 按照primary keys列进行一致性哈希映射到shard X，确定其pchannel(c1,…c6)
5. 构造MsgStream对象<collection, partition, channel,…>并插入pchannel中
6. DataNode(QueryNode)根据DataCoord配置从固定pchannel取出数据，并按照collection聚类（flowgraph）形成log snapshot，并写入s3等；并向DataCoord汇报binlog paths；
7. DataCoord将写入路径记录在etcd
   

参考：https://zhuanlan.zhihu.com/p/517553501?utm_id=0

Create Index

索引按照segment进行构建（索引异步删除逻辑类似）

• RootCoord首先获取出该collection所有sealed segments；
• 对每个segments，RootCoord复杂索引构建任务管理：
  • 向DataCoord获取其Binlog paths(GetInsertBinlogPathsRequest)
  • 向IndexCoord发送创建segment index请求(BuildIndexRequest)
• IndexCoord收到请求，对该segment任务进行如下调度：
  • 生成segment索引构建任务(初始状态位unissued)存入etcd，
  • 根据负载均衡选择IndexNode并发送请求
  • IndexCoord监控segment索引构建任务状态
• IndexNode segment索引构建过程
  • segment的binlogpaths中load log snapshots到memory中
  • 反序列化log snapshot为data blocks
  • 内存中构建segment index
  • index构建完毕后序列化为data blocks，写入index files(indexBuildID对应一个segment)：（indexBuildID/IndexTaskVersion/partitionID/segmentID/key）
  • IndexNode修改etcd中index meta状态

参考：
https://milvus.io/docs/data_processing.md
https://github.com/milvus-io/milvus/blob/master/docs/design_docs/20211227-milvus_create_index.md

Search

1. 从Object Storage获取Index Files中的flushed segment建立索引
2. 也会从Growing Segments中建立索引，每个索引单位是一个segment
3. Segments从Growing 到flushed 状态转换，也会有索引转换
   

具体流程：

1. query coord 会询问 data coord。Data coord 因为一直在负责持续的插入数据，它可以反馈给 query coord 两种信息：一种是已经持久化存储了哪些 segment，另一种是这些已经持久化的 segment 所对应 checkpoint 信息，根据 checkpoint 可以知道从 log broker 中获得这些 segment 所消费到的最后位置
2. query coord 会输出一定的分配策略。这些策略也分成两部分：按照 segment 进行分配（如图示 segment allocator），或按照 channel 进行分配（如图示 channel allocator）
3. 分配给不同的 query node 进行处理
4. query node 就会按照策略进行相应的 load 和 watch 操作。如图示 query node 1 中，historical （批数据）部分会将分配给它的 S1、S3 数据从持久化存储中加载进来，而 streaming 部分会订阅 log broker 中的 Ch1，将这部分流数据接入
   
   

knowhere

对于 Knowhere，不区分训练数据和查询数据。对于每一个 segment，Knowhere 都是用该 segment 的全量数据做训练，再基于该训练结果插入全量数据构建索引

Milvus如何解决单机架构的一些问题

水平扩容

milvus的索引内存数据，存储在query node中，当query扩容（或缩容）时，由于索引文件持久化在对象存储中，query coord会进行重新分配，从而拥有水平扩（缩）容的能力

数据丢失

插入的数据，只要写入消息系统，就不会丢失；索引数据、插入日志也持久化到了对象存储中

数据一致性

Milvus每一条 insert message 中都有分配了一个时间戳，如果 service time 大于 query message 中的 guarantee timestamp，那么就会执行这个查询；从而通过配置，达到不同级别的数据一致性
如何使用 Milvus 向量数据库实现实时查询

效果

Milvus针对一个segment构建一个索引，最后proxy合并检索结果，默认一个segment 1g，从而避免单个索引过大导致效果问题

helm安装部署及升级

开源chart

# Add Milvus Helm repository.
$ helm repo add milvus https://milvus-io.github.io/milvus-helm/# Update charts locally.
$ helm repo update# show chart
helm show chart milvus/milvus# pull chart
helm pull milvus/milvus

prometheus+grafana监控

https://milvus.io/docs/monitor.md

参考
https://zhuanlan.zhihu.com/p/473617910
https://zhuanlan.zhihu.com/p/491030589
https://zhuanlan.zhihu.com/p/500551056
https://zhuanlan.zhihu.com/p/486703915
https://zhuanlan.zhihu.com/p/486971488
https://zhuanlan.zhihu.com/p/502880424
https://zhuanlan.zhihu.com/p/506698319
https://www.modb.pro/db/590924