Giraph源码分析(八)—— 统计每个SuperStep中参与计算的顶点数目

简介: 科研中,需要分析在每次迭代过程中参与计算的顶点数目,来进一步优化系统。比如,在SSSP的compute()方法最后一行,都会把当前顶点voteToHalt,即变为InActive状态。所以每次迭代完成后,所有顶点都是InActive状态。

作者|白松

目的:科研中,需要分析在每次迭代过程中参与计算的顶点数目,来进一步优化系统。比如,在SSSP的compute()方法最后一行,都会把当前顶点voteToHalt,即变为InActive状态。所以每次迭代完成后,所有顶点都是InActive状态。在大同步后,收到消息的顶点会被激活,变为Active状态,然后调用顶点的compute()方法。本文的目的就是统计每次迭代过程中,参与计算的顶点数目。下面附上SSSP的compute()方法:

@Override
  public void compute(Iterable messages) {
    if (getSuperstep() == 0) {
      setValue(new DoubleWritable(Double.MAX_VALUE));
    }
    double minDist = isSource() ? 0d : Double.MAX_VALUE;
    for (DoubleWritable message : messages) {
      minDist = Math.min(minDist, message.get());
    }
    if (minDist < getValue().get()) {
      setValue(new DoubleWritable(minDist));
      for (Edge edge : getEdges()) {
        double distance = minDist + edge.getValue().get();
        sendMessage(edge.getTargetVertexId(), new DoubleWritable(distance));
      }
    }
    //把顶点置为InActive状态
    voteToHalt();
  }

附:giraph中算法的终止条件是:没有活跃顶点且worker间没有消息传递。

hama-0.6.0中算法的终止条件只是:判断是否有活跃顶点。不是真正的pregel思想,半成品。

修改过程如下:

  1. org.apache.giraph.partition. PartitionStats 类

添加变量和方法,用来统计每个Partition在每个超步中参与计算的顶点数目。添加的变量和方法如下:

/** computed vertices in this partition */
private long computedVertexCount=0;
 
/**
* Increment the computed vertex count by one.
*/
public void incrComputedVertexCount() {
    ++ computedVertexCount;
}
 
/**
 * @return the computedVertexCount
 */
public long getComputedVertexCount() {
    return computedVertexCount;
}

修改readFields()和write()方法,每个方法追加最后一句。当每个Partition计算完成后,会把自己的computedVertexCount发送给Master,Mater再读取汇总。

@Override
public void readFields(DataInput input) throws IOException {
    partitionId = input.readInt();
    vertexCount = input.readLong();
    finishedVertexCount = input.readLong();
    edgeCount = input.readLong();
    messagesSentCount = input.readLong();
    //添加下条语句
    computedVertexCount=input.readLong();
}
 
@Override
public void write(DataOutput output) throws IOException {
    output.writeInt(partitionId);
    output.writeLong(vertexCount);
    output.writeLong(finishedVertexCount);
    output.writeLong(edgeCount);
    output.writeLong(messagesSentCount);
    //添加下条语句
    output.writeLong(computedVertexCount);
}
  1. org.apache.giraph.graph. GlobalStats 类

    
    添加变量和方法,用来统计每个超步中参与计算的顶点总数目,包含每个Worker上的所有Partitions。
    
    
 /** computed vertices in this partition 
  *  Add by BaiSong 
  */
  private long computedVertexCount=0;
     /**
     * @return the computedVertexCount
     */
    public long getComputedVertexCount() {
        return computedVertexCount;
    }

修改addPartitionStats(PartitionStats partitionStats)方法,增加统计computedVertexCount功能。

/**
  * Add the stats of a partition to the global stats.
  *
  * @param partitionStats Partition stats to be added.
  */
  public void addPartitionStats(PartitionStats partitionStats) {
    this.vertexCount += partitionStats.getVertexCount();
    this.finishedVertexCount += partitionStats.getFinishedVertexCount();
    this.edgeCount += partitionStats.getEdgeCount();
    //Add by BaiSong,添加下条语句
    this.computedVertexCount+=partitionStats.getComputedVertexCount();
 }

当然为了Debug方便,也可以修改该类的toString()方法(可选),修改后的如下:

public String toString() {
        return "(vtx=" + vertexCount + ", computedVertexCount="
                + computedVertexCount + ",finVtx=" + finishedVertexCount
                + ",edges=" + edgeCount + ",msgCount=" + messageCount
                + ",haltComputation=" + haltComputation + ")";
    }
  1. org.apache.giraph.graph. ComputeCallable

添加统计功能。在computePartition()方法中,添加下面一句。

if (!vertex.isHalted()) {
        context.progress();
        TimerContext computeOneTimerContext = computeOneTimer.time();
        try {
            vertex.compute(messages);
        //添加下面一句,当顶点调用完compute()方法后,就把该Partition的computedVertexCount加1
            partitionStats.incrComputedVertexCount();
        } finally {
           computeOneTimerContext.stop();
        }
……
  1. 添加Counters统计,和我的博客Giraph源码分析(七)—— 添加消息统计功能) 类似,此处不再详述。添加的类为:org.apache.giraph.counters.GiraphComputedVertex,下面附上该类的源码:

package org.apache.giraph.counters;
 
import java.util.Iterator;
import java.util.Map;
 
import org.apache.hadoop.mapreduce.Mapper.Context;
import com.google.common.collect.Maps;
 
/**
 * Hadoop Counters in group "Giraph Messages" for counting every superstep
 * message count.
 */
 
public class GiraphComputedVertex extends HadoopCountersBase {
    /** Counter group name for the giraph Messages */
    public static final String GROUP_NAME = "Giraph Computed Vertex";
 
    /** Singleton instance for everyone to use */
    private static GiraphComputedVertex INSTANCE;
 
    /** superstep time in msec */
    private final Map superstepVertexCount;
 
    private GiraphComputedVertex(Context context) {
        super(context, GROUP_NAME);
        superstepVertexCount = Maps.newHashMap();
    }
 
    /**
     * Instantiate with Hadoop Context.
     * 
     * @param context
     *            Hadoop Context to use.
     */
    public static void init(Context context) {
        INSTANCE = new GiraphComputedVertex(context);
    }
 
    /**
     * Get singleton instance.
     * 
     * @return singleton GiraphTimers instance.
     */
    public static GiraphComputedVertex getInstance() {
        return INSTANCE;
    }
 
    /**
     * Get counter for superstep messages
     * 
     * @param superstep
     * @return
     */
    public GiraphHadoopCounter getSuperstepVertexCount(long superstep) {
        GiraphHadoopCounter counter = superstepVertexCount.get(superstep);
        if (counter == null) {
            String counterPrefix = "Superstep: " + superstep+" ";
            counter = getCounter(counterPrefix);
            superstepVertexCount.put(superstep, counter);
        }
        return counter;
    }
 
    @Override
    public Iterator iterator() {
        return superstepVertexCount.values().iterator();
    }
}
  1. 实验结果,运行程序后。会在终端输出每次迭代参与计算的顶点总数目。 测试SSSP(SimpleShortestPathsVertex类),输入图中共有9个顶点和12条边。输出结果如下:

上图测试中,共有6次迭代。红色框中,显示出了每次迭代过冲参与计算的顶点数目,依次是:9,4,4,3,4,0

解释:在第0个超步,每个顶点都是活跃的,所有共有9个顶点参与计算。在第5个超步,共有0个顶点参与计算,那么就不会向外发送消息,加上每个顶点都是不活跃的,所以算法迭代终止。

阅读更多文章请访问数澜社区

相关文章
|
7月前
|
算法 测试技术 C#
区间合并|LeetCode2963:统计好分割方案的数目
区间合并|LeetCode2963:统计好分割方案的数目
|
2月前
Leetcode第40题(组合总和2)
LeetCode第40题“组合总和II”的解题方法,使用了回溯法来找出所有可能的组合,并对重复元素进行了处理。
28 0
|
2月前
【LeetCode 51】216.组合总和III
【LeetCode 51】216.组合总和III
13 1
|
2月前
LeetCode第39题(组合总和)
LeetCode第39题要求找出一个无重复元素整数数组中所有和为给定目标数的不同组合,可以使用回溯法解决。
55 0
|
4月前
|
算法
LeetCode第40题组合总和II
LeetCode第40题"组合总和II"的解题策略,涉及排序、去重和使用标记数组避免重复组合,通过回溯法实现递归组合。
LeetCode第40题组合总和II
|
4月前
|
算法
LeetCode第39题组合总和
LeetCode第39题"组合总和"的解题思路和技巧,采用回溯法通过递归代替多层嵌套循环,有效解决组合问题。
LeetCode第39题组合总和
|
7月前
|
Java
leetcode-40:组合总和 II
leetcode-40:组合总和 II
53 0
|
7月前
|
Java
leetcode-377:组合总和 Ⅳ
leetcode-377:组合总和 Ⅳ
43 0
|
机器学习/深度学习 算法 安全
LeetCode - #40 组合总和 II
不积跬步,无以至千里;不积小流,无以成江海,Swift社区 伴你前行。如果大家有建议和意见欢迎在文末留言,我们会尽力满足大家的需求。
leetcode:39.组合总和
给定一个无重复元素的数组 candidates和一个目标数 target,找出 candidates中所有可以使数字和为 target的组合。
74 0