MaxCompute用户指南：图模型：示例程序：单源最短距离

Dijkstra 算法是求解有向图中单源最短距离（Single Source Shortest Path，简称为 SSSP）的经典算法。
最短距离：对一个有权重的有向图 G=(V,E)，从一个源点 s 到汇点 v 有很多路径，其中边权和最小的路径，称从 s 到 v 的最短距离。
算法基本原理，如下所示：

• 
  初始化：源点 s 到 s 自身的距离（d[s]=0），其他点 u 到 s 的距离为无穷（d=∞）。
• 
  迭代：若存在一条从 u 到 v 的边，那么从 s 到 v 的最短距离更新为：d[v]=min(d[v], d+weight(u, v))，直到所有的点到 s 的距离不再发生变化时，迭代结束。

1. import java.io.IOException;
3. import com.aliyun.odps.io.WritableRecord;
4. import com.aliyun.odps.graph.Combiner;
5. import com.aliyun.odps.graph.ComputeContext;
6. import com.aliyun.odps.graph.Edge;
7. import com.aliyun.odps.graph.GraphJob;
8. import com.aliyun.odps.graph.GraphLoader;
9. import com.aliyun.odps.graph.MutationContext;
10. import com.aliyun.odps.graph.Vertex;
11. import com.aliyun.odps.graph.WorkerContext;
12. import com.aliyun.odps.io.LongWritable;
13. import com.aliyun.odps.data.TableInfo;
15. public class SSSP {
17.   public static final String START_VERTEX = "sssp.start.vertex.id";
19.   public static class SSSPVertex extends
20.       Vertex<LongWritable, LongWritable, LongWritable, LongWritable> {
22.     private static long startVertexId = -1;
24.     public SSSPVertex() {
25.       this.setValue(new LongWritable(Long.MAX_VALUE));
26.     }
28.     public boolean isStartVertex(
29.         ComputeContext<LongWritable, LongWritable, LongWritable, LongWritable> context) {
30.       if (startVertexId == -1) {
31.         String s = context.getConfiguration().get(START_VERTEX);
32.         startVertexId = Long.parseLong(s);
33.       }
34.       return getId().get() == startVertexId;
35.     }
37.     @Override
38.     public void compute(
39.         ComputeContext<LongWritable, LongWritable, LongWritable, LongWritable> context,
40.         Iterable<LongWritable> messages) throws IOException {
41.       long minDist = isStartVertex(context) ? 0 : Integer.MAX_VALUE;
43.       for (LongWritable msg : messages) {
44.         if (msg.get() < minDist) {
45.           minDist = msg.get();
46.         }
47.       }
49.       if (minDist < this.getValue().get()) {
50.         this.setValue(new LongWritable(minDist));
51.         if (hasEdges()) {
52.           for (Edge<LongWritable, LongWritable> e : this.getEdges()) {
53.             context.sendMessage(e.getDestVertexId(), new LongWritable(minDist
54.                 + e.getValue().get()));
55.           }
56.         }
57.       } else {
58.         voteToHalt();
59.       }
60.     }
62.     @Override
63.     public void cleanup(
64.         WorkerContext<LongWritable, LongWritable, LongWritable, LongWritable> context)
65.         throws IOException {
66.       context.write(getId(), getValue());
67.     }
68.   }
70.   public static class MinLongCombiner extends
71.       Combiner<LongWritable, LongWritable> {
73.     @Override
74.     public void combine(LongWritable vertexId, LongWritable combinedMessage,
75.         LongWritable messageToCombine) throws IOException {
76.       if (combinedMessage.get() > messageToCombine.get()) {
77.         combinedMessage.set(messageToCombine.get());
78.       }
79.     }
81.   }
83.   public static class SSSPVertexReader extends
84.       GraphLoader<LongWritable, LongWritable, LongWritable, LongWritable> {
86.     @Override
87.     public void load(
88.         LongWritable recordNum,
89.         WritableRecord record,
90.         MutationContext<LongWritable, LongWritable, LongWritable, LongWritable> context)
91.         throws IOException {
92.       SSSPVertex vertex = new SSSPVertex();
93.       vertex.setId((LongWritable) record.get(0));
94.       String[] edges = record.get(1).toString().split(",");
95.       for (int i = 0; i < edges.length; i++) {
96.         String[] ss = edges.split(":");
97.         vertex.addEdge(new LongWritable(Long.parseLong(ss[0])),
98.             new LongWritable(Long.parseLong(ss[1])));
99.       }
101.       context.addVertexRequest(vertex);
102.     }
104.   }
106.   public static void main(String[] args) throws IOException {
107.     if (args.length < 2) {
108.       System.out.println("Usage: <startnode> <input> <output>");
109.       System.exit(-1);
110.     }
112.     GraphJob job = new GraphJob();
113.     job.setGraphLoaderClass(SSSPVertexReader.class);
114.     job.setVertexClass(SSSPVertex.class);
115.     job.setCombinerClass(MinLongCombiner.class);
117.     job.set(START_VERTEX, args[0]);
118.     job.addInput(TableInfo.builder().tableName(args[1]).build());
119.     job.addOutput(TableInfo.builder().tableName(args[2]).build());
121.     long startTime = System.currentTimeMillis();
122.     job.run();
123.     System.out.println("Job Finished in "
124.         + (System.currentTimeMillis() - startTime) / 1000.0 + " seconds");
125.   }
126. }

• 
  第 19 行：定义 SSSPVertex ，其中：
  点值表示该点到源点 startVertexId 的当前最短距离。
• 
  compute() 方法使用迭代公式：d[v]=min(d[v], d+weight(u, v)) 更新点值。
• 
  cleanup() 方法把点及其到源点的最短距离写到结果表中。