MaxCompute用户指南：图模型：示例程序：拓扑排序

对于有向边（u,v），定义所有满足 u<v 的顶点序列为拓扑序列，拓扑排序就是求一个有向图的拓扑序列的算法。
算法步骤如下：

1. 
   从图中找到一个没有入边的顶点，并输出。
2. 
   从图中删除该点，及其所有出边。
3. 
   重复以上步骤，直到所有点都已输出。

1. import java.io.IOException;
3. import org.apache.commons.logging.Log;
4. import org.apache.commons.logging.LogFactory;
6. import com.aliyun.odps.data.TableInfo;
7. import com.aliyun.odps.graph.Aggregator;
8. import com.aliyun.odps.graph.Combiner;
9. import com.aliyun.odps.graph.ComputeContext;
10. import com.aliyun.odps.graph.GraphJob;
11. import com.aliyun.odps.graph.GraphLoader;
12. import com.aliyun.odps.graph.MutationContext;
13. import com.aliyun.odps.graph.Vertex;
14. import com.aliyun.odps.graph.WorkerContext;
15. import com.aliyun.odps.io.LongWritable;
16. import com.aliyun.odps.io.NullWritable;
17. import com.aliyun.odps.io.BooleanWritable;
18. import com.aliyun.odps.io.WritableRecord;
20. public class TopologySort {
22.   private final static Log LOG = LogFactory.getLog(TopologySort.class);
24.   public static class TopologySortVertex extends
25.       Vertex<LongWritable, LongWritable, NullWritable, LongWritable> {
27.     @Override
28.     public void compute(
29.         ComputeContext<LongWritable, LongWritable, NullWritable, LongWritable> context,
30.         Iterable<LongWritable> messages) throws IOException {
31.       // in superstep 0, each vertex sends message whose value is 1 to its
32.       // neighbors
33.       if (context.getSuperstep() == 0) {
34.         if (hasEdges()) {
35.           context.sendMessageToNeighbors(this, new LongWritable(1L));
36.         }
37.       } else if (context.getSuperstep() >= 1) {
38.         // compute each vertex's indegree
39.         long indegree = getValue().get();
40.         for (LongWritable msg : messages) {
41.           indegree += msg.get();
42.         }
43.         setValue(new LongWritable(indegree));
44.         if (indegree == 0) {
45.           voteToHalt();
46.           if (hasEdges()) {
47.             context.sendMessageToNeighbors(this, new LongWritable(-1L));
48.           }
49.           context.write(new LongWritable(context.getSuperstep()), getId());
50.           LOG.info("vertex: " + getId());
51.         }
52.         context.aggregate(new LongWritable(indegree));
53.       }
54.     }
55.   }
57.   public static class TopologySortVertexReader extends
58.       GraphLoader<LongWritable, LongWritable, NullWritable, LongWritable> {
60.     @Override
61.     public void load(
62.         LongWritable recordNum,
63.         WritableRecord record,
64.         MutationContext<LongWritable, LongWritable, NullWritable, LongWritable> context)
65.         throws IOException {
66.       TopologySortVertex vertex = new TopologySortVertex();
67.       vertex.setId((LongWritable) record.get(0));
68.       vertex.setValue(new LongWritable(0));
70.       String[] edges = record.get(1).toString().split(",");
71.       for (int i = 0; i < edges.length; i++) {
72.         long edge = Long.parseLong(edges);
73.         if (edge >= 0) {
74.           vertex.addEdge(new LongWritable(Long.parseLong(edges)),
75.               NullWritable.get());
76.         }
77.       }
78.       LOG.info(record.toString());
79.       context.addVertexRequest(vertex);
80.     }
82.   }
84.   public static class LongSumCombiner extends
85.       Combiner<LongWritable, LongWritable> {
87.     @Override
88.     public void combine(LongWritable vertexId, LongWritable combinedMessage,
89.         LongWritable messageToCombine) throws IOException {
90.       combinedMessage.set(combinedMessage.get() + messageToCombine.get());
91.     }
93.   }
95.   public static class TopologySortAggregator extends
96.       Aggregator<BooleanWritable> {
98.     @SuppressWarnings("rawtypes")
99.     @Override
100.     public BooleanWritable createInitialValue(WorkerContext context)
101.         throws IOException {
102.       return new BooleanWritable(true);
103.     }
105.     @Override
106.     public void aggregate(BooleanWritable value, Object item)
107.         throws IOException {
108.       boolean hasCycle = value.get();
109.       boolean inDegreeNotZero = ((LongWritable) item).get() == 0 ? false : true;
110.       value.set(hasCycle && inDegreeNotZero);
111.     }
113.     @Override
114.     public void merge(BooleanWritable value, BooleanWritable partial)
115.         throws IOException {
116.       value.set(value.get() && partial.get());
117.     }
119.     @SuppressWarnings("rawtypes")
120.     @Override
121.     public boolean terminate(WorkerContext context, BooleanWritable value)
122.         throws IOException {
123.       if (context.getSuperstep() == 0) {
124.         // since the initial aggregator value is true, and in superstep we don't
125.         // do aggregate
126.         return false;
127.       }
128.       return value.get();
129.     }
131.   }
133.   public static void main(String[] args) throws IOException {
134.     if (args.length != 2) {
135.       System.out.println("Usage : <inputTable> <outputTable>");
136.       System.exit(-1);
137.     }
139.     // 输入表形式为
140.     // 0 1，2
141.     // 1 3
142.     // 2 3
143.     // 3 -1
144.     // 第一列为vertexid，第二列为该点边的destination vertexid，若值为-1，表示该点无出边
145.     // 输出表形式为
146.     // 0 0
147.     // 1 1
148.     // 1 2
149.     // 2 3
150.     // 第一列为supstep值，隐含了拓扑顺序，第二列为vertexid
151.     // TopologySortAggregator用来判断图中是否有环
152.     // 若输入的图有环，则当图中active的点入度都不为0时，迭代结束
153.     // 用户可以通过输入表和输出表的记录数来判断一个有向图是否有环
154.     GraphJob job = new GraphJob();
155.     job.setGraphLoaderClass(TopologySortVertexReader.class);
156.     job.setVertexClass(TopologySortVertex.class);
157.     job.addInput(TableInfo.builder().tableName(args[0]).build());
158.     job.addOutput(TableInfo.builder().tableName(args[1]).build());
159.     job.setCombinerClass(LongSumCombiner.class);
160.     job.setAggregatorClass(TopologySortAggregator.class);
162.     long startTime = System.currentTimeMillis();
163.     job.run();
164.     System.out.println("Job Finished in "
165.         + (System.currentTimeMillis() - startTime) / 1000.0 + " seconds");
166.   }
167. }