Giraph源碼分析（八）—— 統(tǒng)計(jì)每個(gè)SuperStep中參與計(jì)算的頂點(diǎn)數(shù)目

作者|白松

目的：科研中，需要分析在每次迭代過(guò)程中參與計(jì)算的頂點(diǎn)數(shù)目，來(lái)進(jìn)一步優(yōu)化系統(tǒng)。比如，在SSSP的compute()方法最后一行，都會(huì)把當(dāng)前頂點(diǎn)voteToHalt，即變?yōu)镮nActive狀態(tài)。所以每次迭代完成后，所有頂點(diǎn)都是InActive狀態(tài)。在大同步后，收到消息的頂點(diǎn)會(huì)被激活，變?yōu)锳ctive狀態(tài)，然后調(diào)用頂點(diǎn)的compute()方法。本文的目的就是統(tǒng)計(jì)每次迭代過(guò)程中，參與計(jì)算的頂點(diǎn)數(shù)目。下面附上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));
      }
    }
    //把頂點(diǎn)置為InActive狀態(tài)
    voteToHalt();
  }

附：giraph中算法的終止條件是：沒(méi)有活躍頂點(diǎn)且worker間沒(méi)有消息傳遞。

hama-0.6.0中算法的終止條件只是：判斷是否有活躍頂點(diǎn)。不是真正的pregel思想，半成品。

修改過(guò)程如下：

1. org.apache.giraph.partition. PartitionStats 類

添加變量和方法，用來(lái)統(tǒng)計(jì)每個(gè)Partition在每個(gè)超步中參與計(jì)算的頂點(diǎn)數(shù)目。添加的變量和方法如下：

/** 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()方法，每個(gè)方法追加最后一句。當(dāng)每個(gè)Partition計(jì)算完成后，會(huì)把自己的computedVertexCount發(fā)送給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();
    //添加下條語(yǔ)句
    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);
    //添加下條語(yǔ)句
    output.writeLong(computedVertexCount);
}

2. org.apache.giraph.graph. GlobalStats 類

   添加變量和方法，用來(lái)統(tǒng)計(jì)每個(gè)超步中參與計(jì)算的頂點(diǎn)總數(shù)目，包含每個(gè)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)方法，增加統(tǒng)計(jì)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，添加下條語(yǔ)句
    this.computedVertexCount+=partitionStats.getComputedVertexCount();
 }

當(dāng)然為了Debug方便，也可以修改該類的toString()方法（可選），修改后的如下：

public String toString() {
        return "(vtx=" + vertexCount + ", computedVertexCount="
                + computedVertexCount + ",finVtx=" + finishedVertexCount
                + ",edges=" + edgeCount + ",msgCount=" + messageCount
                + ",haltComputation=" + haltComputation + ")";
    }

3. org.apache.giraph.graph. ComputeCallable<I,V,E,M>

添加統(tǒng)計(jì)功能。在computePartition()方法中，添加下面一句。

if (!vertex.isHalted()) {
        context.progress();
        TimerContext computeOneTimerContext = computeOneTimer.time();
        try {
            vertex.compute(messages);
        //添加下面一句，當(dāng)頂點(diǎn)調(diào)用完compute()方法后，就把該P(yáng)artition的computedVertexCount加1
            partitionStats.incrComputedVertexCount();
        } finally {
           computeOneTimerContext.stop();
        }
……

4. 添加Counters統(tǒng)計(jì)，和我的博客Giraph源碼分析（七）—— 添加消息統(tǒng)計(jì)功能 類似，此處不再詳述。添加的類為：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();
    }
}

5. 實(shí)驗(yàn)結(jié)果，運(yùn)行程序后。會(huì)在終端輸出每次迭代參與計(jì)算的頂點(diǎn)總數(shù)目。 測(cè)試SSSP（SimpleShortestPathsVertex類），輸入圖中共有9個(gè)頂點(diǎn)和12條邊。輸出結(jié)果如下：

上圖測(cè)試中，共有6次迭代。紅色框中，顯示出了每次迭代過(guò)沖參與計(jì)算的頂點(diǎn)數(shù)目，依次是：9,4,4,3,4,0

解釋：在第0個(gè)超步，每個(gè)頂點(diǎn)都是活躍的，所有共有9個(gè)頂點(diǎn)參與計(jì)算。在第5個(gè)超步，共有0個(gè)頂點(diǎn)參與計(jì)算，那么就不會(huì)向外發(fā)送消息，加上每個(gè)頂點(diǎn)都是不活躍的，所以算法迭代終止。

【閱讀更多文章請(qǐng)?jiān)L問(wèn)數(shù)瀾社區(qū)】