图计算系统GraphScope背景(一)

Jin's Blog

2021-09-20 (Updated: 2021-12-06)

图

图计算的框架简单介绍完上文后, 接着要来看看业内目前最先进的实现之一, 首当其冲的就是最新更新的 GraphScope (出自阿里)

长期计划,待更新

0x00. 前言

首先尽可能多的阅读官方文档的框架以及 FAQ 部分, 内置算法, 架构说明,

你可以通过 Github 登录给一个在线体验的个人云端环境, 因为 GraphScope 整合了几个很大的项目, 细看一下发现每个都是一个团队才能做出来的, 所以本篇转为介绍 GraphScope 整体以及相关论文阅读, 其他的核心子项目拆分为不同的文档单独来说.

maxgraph (JNA/ffi 封装的持久化存储, 用于 TP)
gaia(x) (实时并行计算框架, 侧重 TP)
v6d (分布式内存存储, 用于 AP)
grape-lite (全图计算引擎, 侧重 AP)
graphlean (炼金引擎, 用于 AI)

然后 GS(GraphScope, 后续都以此简称) 官网首页有一些可以试试的:

playground: 给一个直接可用的交互环境, 操作和数据都一键准备好了, 可以一键体验 (还挺不错的, 不过这样只能看到前端)
blogs: 有一些发版信息和论文分拆, 后续可能会补充更多的架构代码相关信息.

此部分暂时搁置, 列下参考资料, 后续把其他核心抽空看后再补充上来…

0x01. A One-Stop Large-Scale Graph Computing System

这是 GS 整体的白皮书, 应该是综合了 paper 内容后的版本, 先来看看吧

0x02. A Review of Programming Models for Parallel Graph Processing

如标题所示, 是一篇讲述图计算历史/上下文的文章, 推荐阅读学习.

首先文章开头两端说了下 GAIA 论文背景, 就是上千/万亿的边在生产环境是无法单机计算PageRank, 单源最短路径这类算法的, 业内应声出现了不少图计算的思想和框架, 来一一回顾一下:

1. Think like a vertex (Pregel/GAS)

以 2010 年 Google 的经典论文 Pregel 为代表, 提出了上述的设计模型, 每个点(Vertex) 会存储它自身 + 邻边的信息, 每个算法可以有多轮(superstep)完全一样的计算, 这些计算都是在每个顶点上重复进行的, 也就是说通常需要编写一个 UDF (称为vertex program)然后每个点, 重复的执行这个运算逻辑, 它定义每个点如何接收到上一轮迭代的计算结果, 以及如何传递给其他邻接点, 默认每一轮所有点都只走一步, 然后等待全图的点执行完毕后, 开始下一轮(步), 直到满足某个条件提前终止或没有需要发送的消息终止, 这种计算模式也被称为 BSP 模型 (类似木桶效应, 最慢的计算决定最终的计算时间.)

这样的模型下编写算法似乎还算简单, 比如要写一个带权重单点出发的最短路:

# 每轮进行计算的 computer() 接口实现
def VertexProgramForSSSP(vertex, incoming_msgs):
    # 收到其他点的消息, 然后选出其中"最小"的权值/路径长
    merged_msg = Reduce(incoming_msgs, MIN)
    
    # 如果收到的权重和最小比当前权重小, 则更新当前为最小
    if dist > merged_msg:
        dist = merged_msg
    
    # 如果已有权重 + 当前边权重比neighbor_dist小, 则继续发送给当前点的所有出点
    for neighbor in neighbors:
        if dist + edge_weight < neighbor_dist:
            SendMessage(neighbor, dist + edge_weight)

Pregel 以点作为计算核心, 这样遇到社交类分布的图 (比如 Twitter/Weibo) 就会非常麻烦, 因为大量的超级点(及其出边)都在一个分区内, 然而每轮计算都需要等待这个最慢的分区计算完成, 也就是说会把木桶效应放得很大, 为了避免这个问题, PowerGraph 就提出了一个名为 GAS(Gather-Apply-Scatter) 的编程模型

首先这里引入了一个关键的镜像(mirror)点的概念,

Gather 函数在每个分区本地运行, Apply 函数在 master 节点运行, 最后 Scatter 函数阶段, 那么同样是单点带权最短路径, 写法如下, 它直观看起来比 pregel 的模型要复杂一些,

def GASForSSSP(vertex, ?):
    # gather_nbrs: ALL_NBRS
    def gather(D_u, D(u,v), D_v):
        return D_v + D(v,u)
    
    def sum(a, b): 
        return min(a, b)
    
    def apply(D_u, new_dist):
        D_u = new_dist
        
    # scatter_nbrs: ALL_NBRS
    def scatter(D_u,D(u,v),D_v):
        # If changed, activate neighbor
        if(changed(D_u)):
            Activate(v)
        if(increased(D_u)):
            return NULL
        else:
            return D_u + D(u,v)

这里把 pregel和改进 gas 模型放在一起, 它们的利弊如下:

优点
1. 对多种图算法有更好的实现能力
2. 核心的计算过程(vertex program) 并行度更高? (why?)
缺点
1. 原本单机模式实现的算法必须进行改造, 以符合点为中心的计算模式(think like a vertex)
2. 模型设计每次(vertex program)计算只能传递 1 跳, 深度大的算法可能会非常缓慢, 效率也低. (Short-sighted)

2. Think like a (sub)graph (PIE)

为了改善上面提到的 pregel 的模型里每次计算只能传递1跳的问题, 提出了以子图为计算单位的模型 (又称block-centric/partition centric), 简而言之就是以一部分数据为单位, 而不再以每个点为单位进行计算. 那么类似的, 之前在 pregel 中每轮只走一步的做法, 在这里通过预分区等优化, 让相关联的点尽可能在一个子图/分区内, 然后每个点在1轮计算内可以多次通信和传递消息 (而不是仅能发给自己的邻点), 这样来大幅减少通信成本 & 超级步的数量. 那么最短路在此方案下的实现:

def ComputeSSSP(subgraph, messages):
    # vertices with improved distances
    openset = NULL 
    if superstep == 1: # computer0 (初始化步)
        for v in subgraph.internal_vertices:
        if v == SOURCE:
            v.value = 0 # set distance to source as 0
            openset.add(v) # distance has improved
        else:
        v.value = MAX_INT # not source vertex
    for m in messages: # process input messages
        if subgraph.vertices[m.vertex].value > m.value:
            subgraph.vertices[m.vertex].value = m.value
            openset.add(m.vertex) # distance improved
    boundarySet = dijkstra(openset)
    # Send new distances to boundary vertices
    for (boundarySG, vertex, value) in boundarySet:
        SendToSubgraphVertex(boundarySG, vertex, value)
    VoteToHalt()

def dijkstra(openset):
    boundaryOpenset = NULL
    while openset is not NULL:
    v = GetShortestVertex(openset)
    for v2 in v.neighbors:
            # update neighbors, notify if boundary.
        if v2.isBoundary():
        boundaryOpenset.add(v2.subgraph,v2,v.value+dis(v,v2))
        else if v2.value > v.value + dis(v,v2):
            v2.value = v.value + dis(v,v2)
            openset.add(v2) # distance has improved
    openset.remove(v) # done with this local vertex
    return boundaryOpenset

优点:

有着与点为计算中心同样的表达能力
通信/调度/内存使用开销都会更低
对子图类算法支持更好, 例如 wcc, 各子图算完合并起来即可. (个人补充)

不足:

同样需要改变原有单机的图算法实现, 让它适应 “Think like a graph”
实现更为复杂, 开发者还需要深刻理解例如 “内部点/边界点”的概念
子图为计算单位存储时会有点边冗余信息, 对数据预先分区要求很高, 如果是完全离散的图, 可能比点中心效率还低得多, 并且它的预处理过程耗时一般也挺长(个人补充)

3. Think sequential, run parallel (Grape)

可以理解为串行思考/开发算法, 运行时自动并行, 在樊前辈的论文里提出的一个 PIE (PEval-IncEval-Assemble, 如何理解?)模型, 在这个模型里开发者/用户只需要提供 3 个部分:

PEval: 单机可以直接运行的图查询算法, 例如 TP 里就写好的图算法么? (教科书实现)
IncEval: 一个连续递增的函数(?), 它可以将输入的消息当做更新来修改原有的输出
Assemble: 收集部分结果集, 最后合并为一个最终结果 (reduce?)

看起来也是基于一个分区/子图为计算单位, 每个计算节点先在本地的分区计算得到部分结果, 然后 worker 线程通过消息传递和其他的 worker 交换信息, 收到信息时, 每个工作线程都将增量地计算 incEval, 迭代直到没有新的消息产生. 最后, Assemble 阶段汇总多个局部结果得到一个最终结果. 和之前的计算方式/模型相比, PIE 模型下最大的特点就是用户无需单独为计算改写原本串行的单机图算法, 从而大大减少/降低了开发难度和门槛.

那么, 求全图的最短路径, 在原有的已经提供的情况下, 只需要实现简单两步:

def PEval(source, g, vals, updates):
    for v in vertices:
        updates[v] = MAX_INT # 准备阶段初始化子图所有点
    updates[source] = 0
    dijkstra(g, vals, updates)
    
def IncEval(source, g, vals, updates):
    dijkstra(g, vals, updates) # 增量更新?

# 单机最短路实现代码(略)
def dijkstra(g, vals, updates):
    heap = VertexHeap()
    for i in updates: # 遍历所有 update 数组值
      vals[i] = updates[i]
      heap.push(i, vals[i])
        
    updates.clear()
    
    while not heap.empty():
      u = heap.top().vid
      distu = heap.top().val
      heap.pop()
      for e in g.get_outgoing_edges(u):
        v = e.get_neighbor()
        distv = distu + e.data()
        if vals[v] > distv:
          vals[v] = distv
            if g.is_inner_vertex(v):
              heap.push(v, distv)
              updates[v] = distv       
    return updates

最后贴了一下 PIE 整合在 GraphScope 中后, 4台节点在 LDBC 数据集/测试下的结果, 对比了 Power/Gemini/Plato 三个主流图计算框架, 看起来在 10 亿数据集的情况下, 最长的聚集系数(lcc) 耗时在 200s 左右, 其他绝大部分都在10s 内返回结果. 初步看应该还是挺快的 (但是不知道数据准备工作, 预优化时间有多长, 这里好像没提到.)

0x03.Towards a Swiss Army Knife for a Continuous Life Cycle of Big Graph Analytics

这篇应该是说 GS 的设计就像瑞士军刀, 整合了多个有利的功能为一体. 细节待看

这里开头先说了一下做 GrapScope 项目的原因, 想做一个一站式(整合)图学习/图计算和图存储的平台, 对外提供一个统一的图服务(Graph Service – 个人理解), 然后基于现有的 Python 接口 + Gremlin 语法支持两个方式面向用户. 这里提到了 V6d 作为与现有大数据设施连接的接口/分布式内存存储.

然后接下来是一个重点, 讲述了 GS 设计的核心理念, 以及对 TP 场景和 AP 场景的整体考量和交互, 如下图所示:

关于图的 AP 和 TP 场景不再重述, 这里想说的是如何打通:

从 TP –> AP, 这里如图所示想使用快照的机制, 每 N 小时/天建立一次快照, 然后 AP 那边构建一个不可变的内存图
从 AP –> TP, 这里没有细说, 估计是可选提供回写, 或者是同时输出文件等进行其他的使用

那么传统的做法是怎么样呢, 最早图的 TP 和 AP 是完全解耦的, 也就是说几乎没啥关联, 这样带来许多数据转换开销, 后续打通存储 + 计算的做法一般是从存储端实时读数据, 或者是写入 HDFS 等文件系统读文件, 这样带来大量数据重复加载的开销 (耗时占比相当高).

下面说的是 GraphScope V0.5 引入的关于图存储的两个主要特性:

1. 持久化的图存储(persistent graph)

也就是说 V6d 之外, 提供了一个 MaxGraph 作为图存储的方案, 用于实时 OLTP 查询, 类似一个分布式的 RocksDB, 然后 RocksDB 内置的 MVCC 机制在存储每行时都会携带一个 snapshotID(应该是说的lsn-日志序列号, 从0开始每次写+1) 作为版本号, 由此可避免写阻塞读 . 对写入来说, 在一致性和高吞吐之间做了一个折中选择, 类似 Kineograph, 下面来重点看看:

同一个 session 内的写入会被自动攒一批发一次? 类似一次写 500 条数据, 然后由底层 rocksdb 的序列号来保证写入顺序, 但是这样似乎和传统写入也没啥区别? 让我来看看 Kineograph 的 paper,

这样在保证高写入吞吐的同时仍具有一定的有序性/隔离性(虽然比传统 DB 的严格快照隔离弱), 后续这个持久化存储(应该就是 maxgraph) 看起来会投入更多精力让它更加通用.

2. 延时计算 (lazy evaluation)

最早应该是在函数式编程里提出了 eager evaluation 和 lazy evaluation 的定义, FP 特点之一也就是惰性计算, 函数在没有真正执行时都不会提前被计算处理, 而是会按照一个类似 pipeline graph (DAG)的模式自上而下的进行, TensorFlow 也在系统里运用了这个方式, 那么 GS 中引入它有三方面好处:

类似 TF 的做法, 一个计算任务/作业会被先转换为一个 DAG(就是一个执行顺序图), 那么就是按需(用到再)计算
同样由于这种 plan 的思想 (类似 Gremlin 也有), 可以优化/合并重复的计算链路, 减少不必要重复计算
同上, 这样还可以把多个算子合并为一个整体(类似 GAIA 最后提到的 batch 发送优化)

在 GS 中, 开启一个 session 时就可以指定当前回话的策略是 lazy 和是 eager, 开发时可以选eager, 生产阶段可以使用 lazy 获得更好性能.

参考资料: