package org.apache.flink.graph.partition.centric.utils;

/**
 * Created by nikola on 30.11.15.
 */
public class GraphSSSPRunner {
}

@@ -54,79 +54,3060 @@
 s;%0A%0A
-/**%0A * Created by nikola on 30.11.15.%0A */%0Apublic class GraphSSSPRunner %7B
+import org.apache.flink.api.common.JobExecutionResult;%0Aimport org.apache.flink.api.common.accumulators.Histogram;%0Aimport org.apache.flink.api.common.accumulators.LongCounter;%0Aimport org.apache.flink.api.java.ExecutionEnvironment;%0Aimport org.apache.flink.core.fs.FileSystem;%0Aimport org.apache.flink.graph.Graph;%0Aimport org.apache.flink.graph.library.ConnectedComponents;%0Aimport org.apache.flink.graph.library.SingleSourceShortestPaths;%0Aimport org.apache.flink.graph.partition.centric.PCConnectedComponents;%0Aimport org.apache.flink.graph.partition.centric.PCSingleSourceShortestPaths;%0Aimport org.apache.flink.graph.partition.centric.PartitionCentricConfiguration;%0Aimport org.apache.flink.graph.partition.centric.PartitionCentricIteration;%0Aimport org.apache.flink.types.NullValue;%0A%0Aimport java.util.HashMap;%0Aimport java.util.Map;%0A%0A/**%0A * Utility class to run SSSP algorithm on different graphs%0A */%0Apublic class GraphSSSPRunner %7B%0A%0A    public static void detectComponent(%0A            ExecutionEnvironment environment,%0A            Graph%3CLong, Double, Double%3E graph,%0A            Long srcVertexId,%0A            String partitionCentricOutput,%0A            String vertexCentricOutput) throws Exception %7B%0A%0A        JobExecutionResult result;%0A%0A        PartitionCentricConfiguration configuration = new PartitionCentricConfiguration();%0A%0A        configuration.registerAccumulator(PCConnectedComponents.MESSAGE_SENT_CTR, new LongCounter());%0A        configuration.registerAccumulator(PCConnectedComponents.MESSAGE_SENT_ITER_CTR, new Histogram());%0A        configuration.registerAccumulator(PCConnectedComponents.ITER_CTR, new LongCounter());%0A        configuration.registerAccumulator(PCConnectedComponents.ACTIVE_VER_ITER_CTR, new Histogram());%0A        configuration.registerAccumulator(PartitionCentricIteration.ITER_TIMER, new IterationTimer());%0A%0A        environment.startNewSession();%0A%0A        PCSingleSourceShortestPaths%3CLong, Double%3E algo = new PCSingleSourceShortestPaths%3CLong, Double%3E(srcVertexId, Integer.MAX_VALUE, configuration);%0A%0A        algo.run(graph).writeAsCsv(partitionCentricOutput, FileSystem.WriteMode.OVERWRITE);%0A        result = environment.execute();%0A        Map%3CString, String%3E fields = new HashMap%3C%3E();%0A%0A        fields.put(PCConnectedComponents.MESSAGE_SENT_CTR, %22Total messages sent%22);%0A        fields.put(PCConnectedComponents.MESSAGE_SENT_ITER_CTR, %22Messages sent%22);%0A        fields.put(PCConnectedComponents.ITER_CTR, %22Iteration count%22);%0A        fields.put(PCConnectedComponents.ACTIVE_VER_ITER_CTR, %22Active vertices%22);%0A        fields.put(PartitionCentricIteration.ITER_TIMER, %22Elapse time%22);%0A%0A        Telemetry.printTelemetry(%22Partition centric%22, result, fields);%0A%0A        environment.startNewSession();%0A        SingleSourceShortestPaths%3CLong%3E vcAlgo = new SingleSourceShortestPaths%3C%3E(srcVertexId, Integer.MAX_VALUE);%0A        vcAlgo.run(graph).writeAsCsv(vertexCentricOutput, FileSystem.WriteMode.OVERWRITE);%0A        result = environment.execute();%0A        fields.clear();%0A        Telemetry.printTelemetry(%22Vertex centric%22, result, fields);%0A    %7D
 %0A%7D%0A

Add SSSP utility class for graphs