vearch源码阅读——http重要接口一览 基本名词解释

发表于

基本名词解释

Vearch 是对大规模深度学习向量进行高性能相似搜索的弹性分布式系统。可以做一个和mysql类比的抽象理解，vearch就是一个分布式数据库，只不过存的数据的某些属性可能是向量。下面对vearch里数据相关一些名词做解释：

  1. db :一个库，类似mysql的一个数据库
  2. space :一个表空间，对应mysql的一个table
  3. partition :一个space上的数据分布在多个partition上，每个partition只能存在一台机器上，一个机器可以有多个partition
  4. doc :一个doc可以理解成mysql存的一条数据
  5. field :doc的属性，一个field可以理解成mysql table中的一列。
    • field有自己的type，如果type是vector，表明这个field是一个向量

在verach运行过程中，机器分为三种角色：

  1. router ：接收对数据(doc)增删改查的请求，然后把请求发往ps
  2. master ：接收对集群的操作,比如对db、space、partition的增删,还有对集群的维护操作
  3. ps ：ps上存着一个或多个partition,ps的工作就是接收router的请求并调用c++ faiss库负责具体的向量运算了，是真正的worker

image.png

主要接口与问题

向量召回的步骤简单理解为这样:

  1. 模型同学训练好自己的模型
  2. 把所有物料经过计算算出向量，存在某个地方
  3. 线上服务机器把上一步算好的所有物料向量都拉到本地存着
  4. 一条请求过来，用户特征经过模型计算出一个用户向量
  5. 用户向量在步骤[3]中存的物料向量里搜索，取出最相近的topn个物料向量
  6. 召回完成

如果有了vearch，上述步骤[3]和步骤[5]都可以变成一条rpc请求

  • 对于步骤[3], 离线任务可以将自己训练的物料向量通过vearch的插入&批量插入接口存在vearch里
  • 对于步骤[5], 线上服务可以将本地搜索过程也变成一个rpc或http请求，通过vearch的查询（search）接口找到topn相近的向量
    这么做的好处是：
  • 避免线上服务定时拉取更新物料向量
  • 线上服务机器不需要存储物料向量，省内存
    但也存在一些担忧:
  • 通过rpc请求进行向量搜索时间大概率是没有本地搜索快
  • 正常召回只需要返回topN物料的ID以及用户向量与物料向量的内积即可，但是如果有的业务需要返回物料的向量，比如我要top3000物料的向量，那体积过大，明显不现实,只有把物料向量存在本地才能这么玩

在上述过程中，业务方主要需要使用的这几个接口：

  • 单条插入&批量插入
  • 查询
    下面对vearch接口工作流程进行一个梳理

vearch工作步骤

Vearch对用户http请求调用步骤大致都是：

  1. router收到指令，开始构建RouterRequest
  2. 解析请求，填充RouterRequest物料相关的信息
  3. 根据缓存和ETCD，找到对应的partition的具体机器信息
  4. 发送RouterRequest给partition所在的ps机器
  5. partition所在的ps机器接收到请求，调用gamma引擎执行c++代码进行向量运算

这里提出几个问题，最后回答

  1. 创建新的doc，怎么选择对应存储的partition
  2. router的缓存都有什么
  3. 怎么根据doc的id确认对应的partition

单条插入

单条插入的请求如下

1
2
3
4
5
6
7
8
9
vbnet复制代码curl -XPOST -H "content-type: application/json"  -d'
{
    "field1": "value1",
    "field2": "value2",
    "field3": {
        "feature": [0.1, 0.2]
    }
}
' http://router_server/$db_name/$space_name

请求中的field与space的field一一对应，如果field类型是向量，通过feature:[xxx, xxx]写入，并且维数要与space中该field对应

对应调用handleUpdateDoc()方法,这个方法通过传入的http请求初始化一个UpdateRequest,这里需要注意的是为UpdateRequest设置PKey这里

1
2
ini复制代码URLParamID        = "_id"
args.Doc.PKey = params.ByName(URLParamID)

如果传入参数有 _id,那么PKey就等于传入的_id,否则为””（空）。我们这里是插入，不需要传入_id,所以_id为空。而当调用查询之类接口时，会传入_id。插入新物料时，后续的SetDocsField(docs)方法中会为插入的物料自动生成一个id，方法大概就是自增，这里不深究，只要知道router会为新插入的物料生成唯一id就行了。

完成后，调用updateDoc()函数处理初始化的UpdateRequest。

updateDoc()

updateDoc 函数通过传入的 pb UpdateRequest构建一个RouterRequest并发送给partition，步骤分为：

  1. 装填RouterRequest
  2. 发送请求给partition

RouterRequest的结构

RouterRequest结构如下

image.png

  • head里是请求的基本信息：包括用户名，密码，目标dbname以及spacename
  • md是一个map，记录了请求的方法和id
    • key为HandlerType时，value表示该请求对应的方法（增删改查）
    • key为MessageID时，value表示本条请求的唯一id
  • docs是本条请求的物料信息
  • space是本条请求对应的space的信息
  • sendMap的key是partitionID,value是要发给这个partition的信息,其中items包含了doc信息，其他别的借口时候再补充

updateDoc 关键的步骤代码如下:

1
2
3
4
5
6
7
8
9
css复制代码request := client.NewRouterRequest(ctx, docService.client)
request.SetMsgID()
        .SetMethod(client.ReplaceDocHandler)
        .SetHead(args.Head)
        .SetSpace()
        .SetDocs(docs)
        .SetDocsField()
        .PartitionDocs()
items := request.Execute()

装填

首先通过一串函数装填RouterRequest

  • SetMsgID()为本条request生成唯一id，填入md[MessageID]。
  • SetMethod()填写md[HandlerType]，表明本request是一条更新请求
  • SetSpace()填写request的space字段，获取方式是先从router本地缓存找，找不到就去etcd里拿
  • SetDocs()填写request的docs数组
  • SetDocsField()为docs数组里每一个doc填写PKey和Fields字段，
    • 创建时，PKey为空，generateUUID为doc自动生成一个id
    • 这个函数在Fields里加了一个PKey
  • PartitionDocs()填充sendMap字段，就是把docs字段里的doc都加进sendmap[id]的items数组里
    • id是根据doc的PKey做哈希算出来的，这决定了该doc存在哪个partition上
      RouterRequest装填完毕，下一步就是发送了

router发送给partition

发送是通过RouterRequest的Execute()函数

所有要给partition发送请求的接口最后都会落在这个函数上，这个函数分为两步

  1. 构建rpcClient
  2. 正式发送请求

先来看构建rpcClient

上一步中，我们已经确认了要发往每个partition的数据，存在RouterRequest的sendMap成员里

首先通过partitionID获取对应partition的信息，包括机器地址等信息。获取的方式就是先从本地缓存中取，如果没有就从ETCD里拿,这里介绍一下router的本地缓存，router本地缓存如下图所示：

image.png

缓存相关都在router.client.masterclient.cliCache下面

  • router.client.masterclient.cliCache本身继承了sync.map,存储了nodeID对应的rpcClient，避免多次创建
  • partitionCache里存放了partitionID对应的partition相关信息，包括机器节点ID（没有地址）
  • serverCache里存放了NodeID对应的机器信息，包括IP、端口等
  • 如果在缓存里没有找到，router会访问etcd获取相关数据，router.client.masterclient.store就是etcd相关

继续回到构建rpcClient，构建它的关键就是填写ip和端口。从缓存和etcd拿到nodeID后，调用GetOrCreateRPCClient(ctx, nodeID)，同样从缓存和etcd拿到nodeID对应的具体机器信息（地址、端口），并构造一个rpcClient，完毕

构建好以后，发送就完事了,远程调用的方法是UnaryHandler

partition处理请求

方法入口是UnaryHandler.Execute,根据请求是插入，调用update(ctx, store, req.Items)
这里贴一段代码,后面进入gamma引擎了，这里不做研究

1
2
3
4
5
6
7
8
9
10
11
12
go复制代码func update(ctx context.Context, store PartitionStore, items []*vearchpb.Item) {
    item := items[0]
    docGamma := &gamma.Doc{Fields: item.Doc.Fields}
    docBytes := docGamma.Serialize()
    docCmd := &vearchpb.DocCmd{Type: vearchpb.OpType_REPLACE, Doc: docBytes}
    if err := store.Write(ctx, docCmd, nil, nil); err != nil {
            log.Error("Add doc failed, err: [%s]", err.Error())
            item.Err = vearchpb.NewError(vearchpb.ErrorEnum_INTERNAL_ERROR, err).GetError()
    } else {
            item.Doc.Fields = nil
    }
}

批量插入

批量插入的请求如下,每一个插入的物料要两行

  • 第一行固定 {“index”:{“_id”:”xxx”}} \n,_id可以为空，router会自动生成唯一id
  • 第二行里是物料每一列的值
  • 每一行用结尾需要‘\n’分开
1
2
3
4
5
6
vbnet复制代码curl -H "content-type: application/json" -XPOST -d'
{"index": {"_id": "v1"}}\n
{"field1": "value", "field2": {"feature": []}}\n
{"index": {"_id": "v2"}}\n
{"field1": "value", "field2": {"feature": []}}\n
' http://router_server/$db_name/$space_name/_bulk

router中对应的处理方法是handleBulk(),该方法通过http请求初始化一个BulkRequest,主要就是解析请求中每一个doc,把他们填入BulkRequest.docs里，BulkRequest结构如下：

1
2
3
4
ini复制代码message BulkRequest{
    RequestHead head = 1;
    repeated Document docs = 4;
}

填充完后，调用bulk()方法填充一个RouterRequest并发送,步骤和单条插入里的updateDoc()方法类似。

1
css复制代码reply := handler.docService.bulk(ctx, args)

与单条插入不同的是，批量插入的rpc请求中call的方法是BatchHandler，ps接到router批量插入请求，调用对应的处理方法是bulk()

查询

查询接口示例如下

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
vbnet复制代码curl -H "content-type: application/json" -XPOST -d'
{
    "query": {
        "sum": [{
            "field": "field_name",
            "feature": [0.1, 0.2, 0.3, 0.4, 0.5],
            "min_score": 0.9,
            "boost": 0.5
        }],
        "filter": [{
            "range": {
                "field_name": {
                    "gte": 160,
                    "lte": 180
                }
            }
        },
        {
             "term": {
                 "field_name": ["100", "200", "300"],
                 "operator": "or"
             }
        }]
    },
    "retrieval_param": {
        "nprobe": 20
    },
    "fields": ["field1", "field2"],
    "is_brute_search": 0,
    "online_log_level": "debug",
    "quick": false,
    "vector_value": false,
    "client_type": "leader",
    "l2_sqrt": false,
    "sort": [{"field1":{"order": "asc"}}],
    "size": 10
}
' http://router_server/$db_name/$space_name/_search

工作的方法与前面大同小异，无非是构造请求然后发送，这里mark一下重要的参数

  • sum:跟需要查询的特征，下面又有几个参数:
    1. field 指定创建表时特征字段的名称。类似 select col from table中的 col
    2. feature 传递特征，维数和定义表结构时维数必须相同。
    3. min_score 指定返回结果中分值必须大于等于0.9，两个向量计算结果相似度在0-1之间，min_score可以指定返回结果分值最小值，max_score可以指定最大值。如设置： “min_score”: 0.8，“max_score”: 0.95 代表过滤0.8<= 分值<= 0.95 的结果。同时另外一种分值过滤的方式是使用: “symbol”:”>=”，”value”:0.9 这种组合方式，symbol支持的值类型包含: > 、 >= 、 <、 <= 4种，value及min_score、max_score值在0到1之间。
    4. boost指定相似度的权重，比如两个向量相似度分值是0.7，boost设置成0.5之后,返回的结果中会将分值0.7乘以0.5即0.35。
  • size指定最多返回的结果数量，通过这个参数指定topN。若请求url中设置了size值http://router\_server/$db\_name/$space\_name/\_search?size=20优先使用url中指定的size值。
  • quick搜索结果默认将PQ召回向量进行计算和精排，为了加快服务端处理速度设置成true可以指定只召回，不做计算和精排。(这个不是很理解)

问题回答

  1. 创建新的doc，怎么选择对应存储的partition
  • A: router为doc生成新的唯一id，然后通过hash函数计算partition
  1. router的缓存都有什么
  • A: 缓存相关都在router.client.masterclient.cliCache下面
    • router.client.masterclient.cliCache本身继承了sync.map,存储了nodeID对应的rpcClient，避免多次创建
    • partitionCache里存放了partitionID对应的partition相关信息，包括机器节点ID（没有地址）
    • serverCache里存放了NodeID对应的机器信息，包括IP、端口等
    • 如果在缓存里没有找到，router会访问etcd获取相关数据，router.client.masterclient.store就是etcd相关
  1. 怎么根据doc的id确认对应的partition
  • A:hash函数，代码对应这句:
1
css复制代码partitionID := r.space.PartitionId(murmur3.Sum32WithSeed(cbbytes.StringToByte(doc.PKey), 0))

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

0%