文档章节

基于MaxCompute的图计算实践分享-Resolver简介

_夜枫
 _夜枫
发布于 2017/04/06 22:01
字数 2047
阅读 19
收藏 0

【推荐】2019 Java 开发者跳槽指南.pdf(吐血整理) >>>

Resolver简介

在学习使用MaxCompute-Graph计算模型时,resolver是一个不容易理解的概念。在MaxCompute帮助文档 https://help.aliyun.com/document_detail/27903.html?spm=5176.doc27901.6.662.O0sGBI 中对于Resolver有相关介绍:VertexResolver 用于自定义图拓扑修改时的冲突处理逻辑.

​在Graph的迭代计算模型中,图的拓扑结构会在下面两个场景中发生变化:

​1,在图加载阶段(LOAD),worker会读取输入数据,然后发送修改图拓扑结构的请求;在RESOLVE_LOADING_MUTATION阶段,worker会根据LOAD阶段收到的图拓扑结构变更请求,对图的拓扑结构进行修改;

​2,在每个superstep的COMPUTE阶段,worker也会发送修改图拓扑结构的请求;在下一个superstep的RESOLVE_COMPUTING_MUTATION阶段,会处理收集到上个superstep中发送到该Vertex的请求,然后对图的拓扑结构进行修改。

 

图拓扑结构变化的流程

​下图是COMPUTE阶段图拓扑结构发生变化的示例:

alt

1,在superstep:k的compute阶段,vertex会试图改变图的拓扑结构,修改图拓扑结构的请求包括以下四种:

  1. ​addVertexRequest:添加顶点
  2. ​removeVertexRequest:删除顶点
  3. ​addEdgeRequest:添加边
  4. ​removeEdgeRequest:删除边

这些改变图拓扑结构的请求并不会马上执行并返回,而只是记录在了目标顶点所在的worker上。

2,这些改变图拓扑结构的请求会根据操作顶点的id,partition分发到对应worker上,每个worker会将在superstep:k中收集到的图变更请求保存在serverData.partitionMutations中。这样在superstep:k+1时,每个顶点可以从serverData中获取到对应自己操作的所有请求。

3,在superstep:K+1的RESOLVE_COMPUTING_MUTATION阶段,worker会从serverData获取到对应每个vertexID的vertexChanges(即上一个superstep中对改顶点的图修改请求集合),分别对应然后将其作为参数调用用户定义的resole方法,在resolve方法中完成对当前Vertex结构的修改。

4,完成RESOLVE_COMPUTING_MUTATION阶段之后,开始superstep:K+1的compute阶段。

 

如何定义resolver

用户可以自定义VertexResolver的实现,即自定义类继承VertexResolver并重写VertexResolver.resolve方法,然后在提交作业时通过:

  1. ​job.setLoadingVertexResolverClass(VertexResolver.class)
  2. job.setComputingVertexResolverClass(VertexResolver.class)

 

分别指定在图加载阶段和COMPUTE阶段处理图拓扑结构修改请求的resolver实现类。对于LoadingVertexResolver,如果用户没有指定,框架会提供一个默认的DefaultLoadingVertexResolver实现;但是对于ComputingVertexResolver,如果没有指定,是没有默认实现的,就是说如果用户需要在compute阶段对图的拓扑结构进行修改,则必须指定Resolver实现,否则会抛异常。

关于“冲突”

为什么graph的框架需要用户去自定义Resolver的实现,一个很重要的原因是在处理修改图拓扑结构时可能会出现各种“冲突”。

在graph模型中,修改图拓扑结构是一个异“步(superstep)”的请求,在某次superstep中发起的请求,需要到下一个superstep中才能生效。同时对于同一个顶点的修改请求,可能来自各个不同的worker,彼此之间不能相互感知,来自不同worker的请求先后顺序也是不确定的,所以会出现很多冲突的场景,例如对通过一个顶点进行多次添加,删除不存在的边等等。

在默认的DefaultLoadingVertexResolver实现中,对于下面这些场景会认为是“冲突”,然后会抛出异常:

  • 添加重复点:通常使用 addVertexRequest 添加点,而相同 ID 的点已经存在。
  • 添加重复边:通常使用 addEdgeRequest 添加边时,而相同起点和终点的边已经存在;或者使用 addVertexRequest 添加点时,待添加点中存在重复边。
  • 删除不存在的边:通常使用 removeEdgeRequest 删除边时,待删除的边不存在。
  • 删除不存在的点:通常使用 removeVertexRequest 删除点时,待删除的点不存在。
  • 发送消息到不存在的点:通常使用 sendMessage 发送消息时,目标点不存在。

用户自定义的Resolver中可以自行定义冲突场景,或者对上述冲突场景做其他处理。

​SSSP_Split的例子

​在graph提供的example中提供的三个例子都没有自定义resolver,也没有在迭代过程中改变图拓扑结构。下面这个例子是在原先最简单的例子SSSP(单源最短路径)的基础上增加了分裂顶点的功能,即在某轮迭代中发现某个顶点的出边数大于某个阈值时,将其分裂成两个顶点,这两个顶点之间的距离为0。

 

package com.test;

import java.io.IOException;
import java.util.Iterator;
import java.util.List;
import com.aliyun.odps.Record;
import com.aliyun.odps.graph.Combiner;
import com.aliyun.odps.graph.ComputeContext;
import com.aliyun.odps.graph.DefaultLoadingVertexResolver;
import com.aliyun.odps.graph.Edge;
import com.aliyun.odps.graph.GraphJob;
import com.aliyun.odps.graph.GraphLoader;
import com.aliyun.odps.graph.MutationContext;
import com.aliyun.odps.graph.Vertex;
import com.aliyun.odps.graph.VertexChanges;
import com.aliyun.odps.graph.VertexResolver;
import com.aliyun.odps.graph.WorkerContext;
import com.aliyun.odps.io.LongWritable;
import com.aliyun.odps.io.TableInfo;
import com.aliyun.odps.io.Text;
import com.aliyun.odps.io.WritableComparable;

public class SSSP_Split {

  public static final String START_VERTEX = "sssp.start.vertex.id";
  public static final String MAX_DEGREE = "sssp.vertex.max.degree";

  public static class SSSPVertex extends
      Vertex<LongWritable, LongWritable, LongWritable, LongWritable> {

    private static long startVertexId = -1;
    private static long STEP_BASE = 100;
    
    public SSSPVertex() {
      this.setValue(new LongWritable(Long.MAX_VALUE));
    }

    public boolean isStartVertex(
        ComputeContext<LongWritable, LongWritable, LongWritable, LongWritable> context) {
      if (startVertexId == -1) {
        String s = context.getConfiguration().get(START_VERTEX);
        startVertexId = Long.parseLong(s);
      }
      return getId().get() == startVertexId;
    }

    @Override
    public void compute(
        ComputeContext<LongWritable, LongWritable, LongWritable, LongWritable> context,
        Iterable<LongWritable> messages) throws IOException {
      long minDist = isStartVertex(context) ? 0 : Long.MAX_VALUE;

      for (LongWritable msg : messages) {
        if (msg.get() < minDist) {
          minDist = msg.get();
        }
      }
      
      if (minDist < this.getValue().get()) {
        this.setValue(new LongWritable(minDist));
        if (hasEdges()) {
          for (Edge<LongWritable, LongWritable> e : this.getEdges()) {
            context.sendMessage(e.getDestVertexId(), new LongWritable(minDist
                + e.getValue().get()));
          }
        }
      } else {
        voteToHalt();
      }
      // 这里执行分裂顶点的逻辑,当顶点的出边数大于阈值sssp.vertex.max.degree时执行分裂,
      // 并且分裂行为只在前3轮发生.
      if (this.getEdges().size() > context.getConfiguration().getInt(
          MAX_DEGREE, Integer.MAX_VALUE)
          && context.getSuperstep() < 3) {
        SSSPVertex splitVertex = new SSSPVertex();
        // 分裂的顶点id为原id+100*2^迭代轮次,这样避免id出现冲突(初始顶点数小于100)
        // 即顶点3在第0步时分裂出来顶点103,顶点103在第2轮时分裂出顶点503
        splitVertex.setId(new LongWritable(getId().get()
            + (STEP_BASE << context.getSuperstep())));
        splitVertex.setValue(getValue());
        context.addVertexRequest(splitVertex);
        // 添加原始顶点与分裂顶点之间的边,距离为0
        context.addEdgeRequest(splitVertex.getId(),
            new Edge<LongWritable, LongWritable>(this.getId(),
                new LongWritable(0)));
        context.addEdgeRequest(getId(), new Edge<LongWritable, LongWritable>(
            splitVertex.getId(), new LongWritable(0)));
        // 将原始顶点上一半的出边转移到分裂顶点上。
        for (int i = 0; i < this.getEdges().size(); i++) {
          if (i % 2 == 0) {
            context.removeEdgeRequest(getId(), getEdges().get(i).getDestVertexId());
            context.addEdgeRequest(splitVertex.getId(), getEdges().get(i));
          } 
        }
      }
    }

    @Override
    public void cleanup(
        WorkerContext<LongWritable, LongWritable, LongWritable, LongWritable> context)
        throws IOException {
      String edges = "";
      for (int i =0;i<getEdges().size();i++){
        edges+=getEdges().get(i).getDestVertexId()+":";
      }
      // 输出每个顶点的id,出边,和与目标顶点的距离
      context.write(getId(), new Text(edges), getValue());
    }
  }

  public static class MinLongCombiner extends
      Combiner<LongWritable, LongWritable> {

    @Override
    public void combine(LongWritable vertexId, LongWritable combinedMessage,
        LongWritable messageToCombine) throws IOException {
      if (combinedMessage.get() > messageToCombine.get()) {
        combinedMessage.set(messageToCombine.get());
      }
    }
  }
  
  public static class SSSPVertexReader extends
      GraphLoader<LongWritable, LongWritable, LongWritable, LongWritable> {
    @Override
    public void load(
        LongWritable recordNum,
        Record record,
        MutationContext<LongWritable, LongWritable, LongWritable, LongWritable> context)
        throws IOException {
      SSSPVertex vertex = new SSSPVertex();
      vertex.setId((LongWritable) record.get(0));
      String[] edges = record.get(1).toString().split(",");
      for (int i = 0; i < edges.length; i++) {
        String[] ss = edges[i].split(":");
        vertex.addEdge(new LongWritable(Long.parseLong(ss[0])),
            new LongWritable(Long.parseLong(ss[1])));
      }
      context.addVertexRequest(vertex);
    }
  }
  
  // 这里定义computer阶段的resolver
  public static class SSSPVertexResolver extends VertexResolver{
    @Override
    public Vertex resolve(WritableComparable vertexId, Vertex vertex,
        VertexChanges vertexChanges, boolean hasMessages) throws IOException {
      // 处理添加顶点的请求
      if (vertexChanges.getAddedVertexList() != null
          && vertexChanges.getAddedVertexList().size() > 0) {
        vertex = (Vertex) vertexChanges.getAddedVertexList().get(0);
      }
      // 处理添加边的请求
      if (vertexChanges.getAddedEdgeList() != null
          && vertexChanges.getAddedEdgeList().size() > 0) {
        for (Edge<LongWritable, LongWritable> edge : (List<Edge<LongWritable, LongWritable>>) (vertexChanges
            .getAddedEdgeList())) {
          vertex.addEdge(edge.getDestVertexId(), edge.getValue());
        }
      }
      // 处理删除边的请求
      if (vertexChanges.getRemovedEdgeList() != null
          && vertexChanges.getRemovedEdgeList().size() > 0) {
        for (LongWritable removedDestVertex : (List<LongWritable>)vertexChanges.getRemovedEdgeList()) {
          List<Edge<LongWritable, LongWritable>> edgeList = vertex.getEdges();
          for (Iterator<Edge<LongWritable, LongWritable>> edges = edgeList.iterator(); edges.hasNext();) {
            Edge<LongWritable, LongWritable> edge = edges.next();
            if (edge.getDestVertexId().equals(removedDestVertex)) {
              edges.remove();
            }
          }
        }
      }
      // 处理删除顶点的请求
      if (vertexChanges.getRemovedVertexCount() > 0) {
        // do nothing
      }
      return vertex;
    }
    
  }
  public static void main(String[] args) throws IOException {
    if (args.length < 3) {
      System.out.println("Usage: <startnode> <max degree> <input> <output>");
      System.exit(-1);
    }

    GraphJob job = new GraphJob();
    job.setGraphLoaderClass(SSSPVertexReader.class);
    job.setVertexClass(SSSPVertex.class);
    job.setCombinerClass(MinLongCombiner.class);
    // 设置compute阶段的resolver
    job.setComputingVertexResolverClass(SSSPVertexResolver.class);
    job.set(START_VERTEX, args[0]);
    job.set(MAX_DEGREE, args[1]);
    job.addInput(new TableInfo(args[2]));
    job.addOutput(new TableInfo(args[3]));

    long startTime = System.currentTimeMillis();
    job.run();
    System.out.println("Job Finished in "
        + (System.currentTimeMillis() - startTime) / 1000.0 + " seconds");
  }
}


 

输入表数据ssp_in:

1,"2:2,3:1,4:4,5:5"
2,"1:2,3:2,4:1,5:4"
3,"1:1,2:2,4:2,5:3"
4,"1:4,2:1,3:2,5:1"
5,"1:5,2:4,3:1,4:1"

输出表数据sssp_out:

102,2:1:4:,2
103,3:1:4:,1
1,3:5:101:,0
2,3:5:102:,2
101,1:2:4:,0
3,2:5:103:,1
4,2:5:104:,3
5,2:4:105:,4
104,4:1:3:,3
105,5:1:3:,4

输出数据中,分裂出来的点与源点(顶点103和顶点3)与目秒顶点的距离相同,两者实际仍然是同一个点。

本文转载自:https://yq.aliyun.com/articles/69104

_夜枫
粉丝 10
博文 506
码字总数 0
作品 0
朝阳
后端工程师
私信 提问
【袋鼠云技术团队沙龙第13期】大数据计算服务MaxCompute线下Meetup-杭州站

活动简介: 大数据时代,计算的力量支撑着数据的价值。2018年双11 ,作为阿里巴巴统一的计算平台,MaxCompute单日数据处理超过500PB,平稳支撑电商混布单元在线流量洪峰12万笔/s交易。MaxCo...

墨羽MOYU
2018/11/21
8
0
【袋鼠云技术团队沙龙第13期】大数据计算服务MaxCompute线下Meetup-杭州站

活动简介: 大数据时代,计算的力量支撑着数据的价值。2018年双11 ,作为阿里巴巴统一的计算平台,MaxCompute单日数据处理超过500PB,平稳支撑电商混布单元在线流量洪峰12万笔/s交易。MaxCo...

墨羽MOYU
2018/11/21
46
0
阿里巴巴大数据计算平台MaxCompute(原名ODPS)全套攻略(持续更新20171127)

概况介绍 大数据计算服务(MaxCompute,原名ODPS,产品地址:https://www.aliyun.com/product/odps)是一种快速、完全托管的TB/PB级数据仓库解决方案。MaxCompute向用户提供了完善的数据导入方...

隐林
2017/05/05
0
0
干货满满的大数据技术沙龙来了

豪华讲师阵容,大牛云集,云栖线下沙龙大数据专场即将开启,为你揭秘大数据与人工智能技术的最佳实践,2月24日邀您到现场,与大牛面对面。https://yq.aliyun.com/promotion/154 大数据和人工...

阿里云官方博客
2017/02/23
439
1
【8月17日】MaxCompute携手袋鼠云:“云计算&大数据实践”杭州技术沙龙

阿里巴巴主力计算平台的大数据计算平台MaxCompute本周五在杭州携手袋鼠云与大家再次见面,感兴趣的朋友们赶快报名参加吧。 活动简介 为响应浙江省万企上云号召,抓住新一轮科技革命和产业变革...

晋恒
2018/08/14
0
0

没有更多内容

加载失败,请刷新页面

加载更多

apache httpClient实现代理发送Post请求

CredentialsProvider credsProvider = new BasicCredentialsProvider(); credsProvider.setCredentials( new AuthScope("host", port), new UsernamePasswordCredentials(username, password......

huangkejie
5分钟前
1
0
SpringCloud

单体应用存在的问题 ● 随着业务的发展,开发变得越来越复杂。 ● 修改、新增某个功能,需要对整个系统进行测试,重新部署。 ● 一个模块出现问题,很可能导致整个系统崩溃。 ● 多个开发团队...

Star永恒
16分钟前
2
0
contos7安装elasticsearch6.8

基于最近安装elasticsearch过程遇到的坑在这里总结一下,记录一下。 环境: elasticsearch基于Java,需要有Java环境,最新的es需要jdk11,这选择安装的事6.8.0,需要jdk8。 版本: 安装去官网...

eric_le
19分钟前
2
0
PHP 全链路追踪 Molten-Docker

#DockerfileFROM daocloud.io/php:7.2-fpm-alpineENV TIME_ZONE=Asiz/Shanghai RUN apk add linux-headers# Compile MoltenRUN set -xe && \ curl -LO https://git......

MrMyj
21分钟前
2
0
聊聊rocketmq producer的batch

序 本文主要研究一下rocketmq producer的batch batch rocketmq-client-4.6.0-sources.jar!/org/apache/rocketmq/client/producer/DefaultMQProducer.java public class DefaultMQProducer ex......

go4it
昨天
5
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部