图系统Hugegraph-K步邻居解析

图
, , ,

从数据结构和函数式风格的图查询语言Gremlin走来, 不难看出图的遍历方式极其灵活, 且原生图论里就有许多经典的图算法的. 在Hugegraph中同样也针对常用的一些图算法, 单独进行了实现和封装.

今天就先来看看传统图算法里, 使用频率最高之一的K-neighbor (K步邻居)算法的实现源码, 从而分析如果要提高查询速度, 还能从哪些方面入手优化.

0x00. 概览

1. 简介

K步邻居之所以在实际业务场景中使用评率很高, 有很大一部分原因是因为它是最裸的一种图查询算法, 业务就算没有学习过任何图查询算法, 因为它的广遍历特点, 也能快速从K步邻居的结果里提取到需要的信息 (简单直观).

它的实际效果是: 从某个点出发, 遍历出K步(层)内的所有关联点. 再简单一点说, 就像是从某个点出发的K次迭代循环. 举个实际的例子就像你的微信里有140个好友, 每个好友又分别有数百个好友, 通过K步邻居, 就能一次得到从你出发, 所有好友的展开关系网, 然后从中就可以进行筛选和分析, 得到想要的信息.

补充: 需要注意的是, 准确来说, 图的常规算法里是没有kneighbor这个说法的, 学术名字上与它最接近的是KNN(K-Nearest Neighbor ), 这个是机器学习里的常见算法之一, 跟我们这说的不是一个, 所以你也可以称它为kstep 或者想做是BFS的一种限定条件实现. 只不过为了方便沟通, 目前称它为K步邻居:(

2. 访问

从官方文档中可以看到大体的访问方式, 最简单的查询某同学的3层所有朋友的写法可以是:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
//Get请求,URL如下
http://ip/graphs/graphName/traversers/kneighbor?source=“1:jin”&max_depth=3

//返回值,仅有点IDs
{
    "vertices":[
        "2:doge",
        "1:tom",
        "1:tom2",
        "1:tom3",
        "1:jerry",
        "2:cat",
        .....
    ]
}

但是你会发现这个返回结果很有点摸不着头脑, 因为这并不是之前预期的那样—-它告诉我1层(自己)有哪些朋友, 我2层(朋友的朋友)是哪些, 而是一股脑把所有名字都输出给了我, 并且没有点对应的边的信息. 这都是后续可以改进的地方~

0x01. 分析

通过之前性能测试就已经发现, 当深度变大(>3层)的时候, K步邻居的查询速度会明显变长, 甚至于出错无法得到结果. 那么分析的核心自然就是找到它的耗时点, 然后分析瓶颈到底在什么地方, 是HugeServer , 还是后端存储?

通过对比JanusGraph和Tinkerpop的实现方式, 对K步邻居性能上的关注点集中在以下几个地方:

  1. 通过gremlin查询K步邻居和使用kneighbor查询本质区别, 包括gremlin的limit() 实现方式 (待完成….)
  2. 通过kneighbor查询的时候, 每一层是通过上一层缓存的点一次性去查下一层(1次RPC), 还是说每个点依次迭代(N次RPC ).

然后带着这些问题, 再结合源码去验证和进一步思考.

0x02. 源码

首先, 所有的定制图算法在结构上都是属于Traverser 的. 包括最短路径, 全路径, PageRank, 三角计数等等… 而在Huge中这个结构也挺清晰, 分为以下几个核心步骤:

  1. 构造API
  2. 编写Traverser算法 (核心)
  3. 针对后端定制 (可选)

就以Kneighbor为例, 首先在hugegraph-api 模块中, 它会有一个KneighborAPI , 然后提供网络I/O的接口, 解析通过HTTP请求发送的Kneighbor参数, 调用kneighbor算法, 最后序列化返回结果, 这个过程简化的模型如下:

1
2
3
4
5
6
7
8
9
10
    @GET
public String get(@QueryParam("source") String sourceV,@QueryParam("direction") String dire,....params) {
        // 转换各种传入的string参数为对象
        transformInputParams(sourceV, dire, params....);
        
        HugeGraph g = graph(manager, graph);
        Set<Id> ids = traverser.kneighbor(source, dir, edgeLabel, depth, degree, limit);
        
        return serializer(g).writeJsonList("vertices", ids);
    }

所以关键自然是看kneighbor的具体实现了, 其中degree指单个顶点最大可遍历边数目.

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
public Set<Id> kneighbor(Id sourceV, Directions dir,
                         String label, int depth, long degree, long limit) {
        checkParams(sourceV,dir....);
        // Q1:如果不传,则查所有边是在哪实现的... ("" or null)
        Id labelId = this.getEdgeLabelId(label);
        
        // 初始化两个的Set, 添加初始顶点
        Set<Id> latest = newSet(sourceV);
        Set<Id> all = newSet(sourceV);
    
        // 按每层遍历
        while (depth-- > 0) {
            long remaining = limit == NO_LIMIT ? NO_LIMIT : limit - all.size();
            // 每次更新latest集合加入相邻顶点(核心)
            latest = adjacentVertices(latest, dir, labelId, all, degree, remaining);
            all.addAll(latest);
            
            if (limit != NO_LIMIT && all.size() >= limit) break; //遍历点数超过上限则跳出
        }
        return all;
    }

private Set<Id> adjacentVertices(Set<Id> vertices, Directions dir,Id label,
                                 Set<Id> excluded, long degree, long limit) {
        if (limit == 0) return ImmutableSet.of();

        Set<Id> neighbors = newSet();
        // 依次遍历latest顶点 (比如在第二层, 则遍历第一层的所有点+起点)
        for (Id source : vertices) {
            // 拿到从这个点出发的所有边,,时间复杂度至少O(n)?
            Iterator<Edge> edges = edgesOfVertex(source, dir,label, degree);
            while (edges.hasNext()) {
                HugeEdge e = (HugeEdge) edges.next();
                // 获得每条边指向的顶点ID
                Id target = e.id().otherVertexId();
                // 跳过or添加这个点
                if (excluded != null && excluded.contains(target)) continue;
                neighbors.add(target);
                
                if (limit != NO_LIMIT && neighbors.size() >= limit) return neighbors;
            }
        }
        return neighbors;
}

Iterator<Edge> edgesOfVertex(Id source, Directions dir, Id label, long limit) {
        Id[] labels = {};
        if (label != null) labels = new Id[]{label}; //Q2:为何下面不直接传label
    
        //通过"fromV + 方向 + 边label"查询边, 这里确定是一个ConditionQuery (查的细节在后面.)
        Query query = GraphTransaction.constructEdgesQuery(source, dir, labels);
        if (limit != NO_LIMIT) query.limit(limit);
    
        return this.graph.edges(query);
    }

所以从上面可以很清楚的看到目前K步邻居的整个逻辑:

  1. 给定两个Set集合, 一个记录每层遍历的点(latest), 一个记录整个过程遍历的点(all).
  2. 从第一层开始, 依次遍历当前层的所有点 + 之前层的点.
    • 首先获得当前点的所有边(方向可指定)
    • 遍历每一条边, 然后获得边指向的另一个顶点
    • 判断指向的点是否记录过, 没有就加进去.
  3. 依次循环, 直到遍历数达到上限, 或遍历完成.

过程可以理解为: 用两个Set替代了传统队列的BFS遍历, 只不过目前的Set集合只进不出, 只是能自动去重. 所以在多层迭代遍历的时候效率较低, 因为每次遍历不仅要遍历当前层所有点, 还有遍历之前已经遍历过的所有点 (虽然有跳过逻辑)

整个遍历过程:

  • 算法复杂度接近: O(n) + O(n+n^2^) +…. O(n+n^2^+….n^k^) ≈ O(n^k^) (K代表遍历层数, 当然因为**图缓存**的关系, 所以低层数并不会很慢.)
  • 时间复杂度: S(2n)

并且当前只是一股脑把所有点去重获得了, 并不知道实际用途… 所以如果要用它替代目前gremlin自己实现的K步邻居, 还有比较多需要改动的地方,

**补充: **

因为Kneighbor有个孪生兄弟Kout. 所以不妨也看看它实现的区别. (检查省去)

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
// 
public Set<Id> kout(Id sourceV, Directions dir, String label,int depth,
                    boolean nearest,long degree, long capacity, long limit) {
        // 查找从起始顶点出发恰好depth步可达的顶点
        Set<Id> latest = newSet();
        latest.add(sourceV);

        Set<Id> all = newSet();
        all.add(sourceV);

        long remaining = capacity == NO_LIMIT ? NO_LIMIT : capacity - latest.size();
        while (depth-- > 0) {
            // Just get limit nodes in last layer if limit < remaining capacity
            if (depth == 0 && limit != NO_LIMIT &&(limit < remaining || remaining == NO_LIMIT)) remaining = limit;
            
            // 为true时，代表起始顶点到达结果顶点的最短路径长度为depth，不存在更短的路径,否则允许环
            if (nearest) {
                latest = this.adjacentVertices(latest, dir, labelId, all,degree, remaining);
                all.addAll(latest);
            } else {
                latest = this.adjacentVertices(latest, dir, labelId, null,degree, remaining);
            }
            if (capacity != NO_LIMIT) {
                // Update 'remaining' value to record remaining capacity
                remaining -= latest.size();

                if (remaining <= 0 && depth > 0) exit; // >limit                    
       }}
        return latest;
    }

这里面有一个比较核心的操作其实就是, 获取某个点的所有边这个操作, 现在来跟一下它的实现方式, 到底如何去后端查询的: (这里直接到深处~)

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
//关键查询后端在于此,后续具体分不同后端操作,这里以Hbase/Binary序列化为例
   @Override
   public Iterator<BackendEntry> query(Query query) {
        if (!(query instanceof ConditionQuery)) {
            return super.query(query); //非条件查询,writeQueryEdgePrefixCondition
        }

        QueryList queries = new QueryList(this.graph(), query,
                                          q -> super.query(q));
        for (ConditionQuery cq: ConditionQueryFlatten.flatten(
                                (ConditionQuery) query)) {
            Query q = this.optimizeQuery(cq);
            /*
             * NOTE: There are two possibilities for this query:
             * 1.sysprop-query, which would not be empty.
             * 2.index-query result(ids after optimization), which may be empty.
             */
            if (q == null) {
                queries.add(this.indexQuery(cq));
            } else if (!q.empty()) {
                queries.add(q);
            }
        }

        return !queries.empty() ? queries.fetch() : Collections.emptyIterator();
    }

//上面的query()方法,到Hbase前的查询条件如下(可以看到总共有5种方式)
    @Override
    public Iterator<BackendEntry> query(Session session, Query query) {
        if (query.limit() == 0 && query.limit() != Query.NO_LIMIT) return ImmutableList.<BackendEntry>of().iterator();

        // Query all
        if (query.empty()) return newEntryIterator(this.queryAll(session, query), query);

        // Query by prefix  (查"某点的四出边"属于这种,比如"1:jin"的出边 )
        if (query instanceof IdPrefixQuery) {
            IdPrefixQuery pq = (IdPrefixQuery) query;
            return newEntryIterator(this.queryByPrefix(session, pq), query);
        }

        // Query by range 
        // Query by id (非条件查询)
        // Query by condition (or condition + id)  
        ..... //这几种的实现都省去,因为暂时不涉及
    }

//条件查询到Hbase使用的是scan by rowkey查询
   public RowIterator scan(String table, byte[] startRow,
                           boolean inclusiveStart, byte[] prefix) {          
  // Hbase-client的scan, 这里filter就是序列化后的通过id去前缀匹配rowkey
  Scan scan = new Scan().withStartRow(startRow, inclusiveStart).setFilter(new PrefixFilter(prefix));
  return this.scan(table, scan);
}

所以可以确定这里只需要一次RPC, 就能从Hbase中读取到点对应的边数据, 不过这里starRow是在哪设置的什么来着….

待补…

0x03. 改进

改进分两个层次: 功能 & 性能, 先列一下功能上的:

  1. 返回值中不单返回点ID, 而是返回完整点信息. (或给一个选择)
  2. max_degree能指定每个边(带方向)返回的最大数, 而不只是总数.
  3. 允许传入多条边(目前只能一条or全部), direction同理, 形成类似 map<label,direction>的参数传入.
  4. 返回点对应的边, 先考虑返回全部, 再考虑最好每层能分开返回[e.g 第一层点+边, 第二层点+边 …] (复杂度: 待定)
  5. 支持同时传入多个顶点ID (复杂度: 待定)

性能上:

  1. 遍历的时候不应重复选取已遍历过的点, 改为队列或先去重后的点集(差集).
  2. 遍历的时候尽可能多的允许使用剪枝 (主要是条件剪枝..)
  3. 如果不用输出每层的点边, 可以一次性去后端读下一层数据, 而不是一个个点去查.
  4. 重复查询的时候, 尽可能多的利用缓存 (比如先查某人的2层邻居, 再查4层, 则应该直接从3层开始查.)

最关键的优化点是: K步邻居/BFS是一个完全可并发的过程, 所以应该用多个线程, 或者是其他并发实现去查询, 而不应是单点 ,然后就是加大缓存的利用, 详细参考下一篇Hugegraph-缓存设计.

更新: 关于分布式执行的尝试参考之后的协处理器文章. (计算下推)

参考资料:

  1. 启发式寻路算法(A*,BFS,Dijkstra)
  2. 可视化の图遍历(可绘图)