MaxCompute用户指南：图模型：示例程序：三角形计数

三角形计数算法用于计算通过每个顶点的三角形个数。
算法实现的流程如下：

1. 
   每个顶点将其 ID 发送给所有出边邻居。
2. 
   存储入边和出边邻居并发送给出边邻居。
3. 
   对每条边计算其终点的交集数量，并求和，结果输出到表。
4. 
   将表中的输出结果求和并除以三，即得到三角形个数。

1. import java.io.IOException;
3. import com.aliyun.odps.data.TableInfo;
4. import com.aliyun.odps.graph.ComputeContext;
5. import com.aliyun.odps.graph.Edge;
6. import com.aliyun.odps.graph.GraphJob;
7. import com.aliyun.odps.graph.GraphLoader;
8. import com.aliyun.odps.graph.MutationContext;
9. import com.aliyun.odps.graph.Vertex;
10. import com.aliyun.odps.graph.WorkerContext;
11. import com.aliyun.odps.io.LongWritable;
12. import com.aliyun.odps.io.NullWritable;
13. import com.aliyun.odps.io.Tuple;
14. import com.aliyun.odps.io.Writable;
15. import com.aliyun.odps.io.WritableRecord;
17. /**
18. * Compute the number of triangles passing through each vertex.
19. *
20. * The algorithm can be computed in three supersteps:
21. * I. Each vertex sends a message with its ID to all its outgoing
22. * neighbors.
23. * II. The incoming neighbors and outgoing neighbors are stored and
24. * send to outgoing neighbors.
25. * III. For each edge compute the intersection of the sets at destination
26. * vertex and sum them, then output to table.
27. *
28. * The triangle count is the sum of output table and divide by three since
29. * each triangle is counted three times.
30. *
31. **/
32. public class TriangleCount {
34.   public static class TCVertex extends
35.     Vertex<LongWritable, Tuple, NullWritable, Tuple> {
37.     @Override
38.     public void setup(
39.         WorkerContext<LongWritable, Tuple, NullWritable, Tuple> context)
40.       throws IOException {
41.       // collect the outgoing neighbors
42.       Tuple t = new Tuple();
43.       if (this.hasEdges()) {
44.         for (Edge<LongWritable, NullWritable> edge : this.getEdges()) {
45.           t.append(edge.getDestVertexId());
46.         }
47.       }
49.       this.setValue(t);
50.     }
52.     @Override
53.     public void compute(
54.         ComputeContext<LongWritable, Tuple, NullWritable, Tuple> context,
55.         Iterable<Tuple> msgs) throws IOException {
57.       if (context.getSuperstep() == 0L) {
58.         // sends a message with its ID to all its outgoing neighbors
59.         Tuple t = new Tuple();
60.         t.append(getId());
61.         context.sendMessageToNeighbors(this, t);
62.       } else if (context.getSuperstep() == 1L) {
63.         // store the incoming neighbors
64.         for (Tuple msg : msgs) {
65.           for (Writable item : msg.getAll()) {
66.             if (!this.getValue().getAll().contains((LongWritable)item)) {
67.               this.getValue().append((LongWritable)item);
68.             }
69.           }
70.         }
72.         // send both incoming and outgoing neighbors to all outgoing neighbors
73.         context.sendMessageToNeighbors(this, getValue());
74.       } else if (context.getSuperstep() == 2L) {
75.         // count the sum of intersection at each edge
76.         long count = 0;
77.         for (Tuple msg : msgs) {
78.           for (Writable id : msg.getAll()) {
79.             if (getValue().getAll().contains(id)) {
80.               count ++;
81.             }
82.           }
83.         }
85.         // output to table
86.         context.write(getId(), new LongWritable(count));
87.         this.voteToHalt();
88.       }
89.     }
90.   }
93.   public static class TCVertexReader extends
94.   GraphLoader<LongWritable, Tuple, NullWritable, Tuple> {
96.     @Override
97.     public void load(
98.         LongWritable recordNum,
99.         WritableRecord record,
100.         MutationContext<LongWritable, Tuple, NullWritable, Tuple> context)
101.     throws IOException {
102.       TCVertex vertex = new TCVertex();
104.       vertex.setId((LongWritable) record.get(0));
106.       String[] edges = record.get(1).toString().split(",");
107.       for (int i = 0; i < edges.length; i++) {
108.         try {
109.           long destID = Long.parseLong(edges);
110.           vertex.addEdge(new LongWritable(destID), NullWritable.get());
111.         } catch(NumberFormatException nfe) {
112.           System.err.println("Ignore " + nfe);
113.         }
114.       }
115.       context.addVertexRequest(vertex);
116.     }
117.   }
119.   public static void main(String[] args) throws IOException {
120.     if (args.length < 2) {
121.       System.out.println("Usage: <input> <output>");
122.       System.exit(-1);
123.     }
125.     GraphJob job = new GraphJob();
126.     job.setGraphLoaderClass(TCVertexReader.class);
127.     job.setVertexClass(TCVertex.class);
129.     job.addInput(TableInfo.builder().tableName(args[0]).build());
130.     job.addOutput(TableInfo.builder().tableName(args[1]).build());
131.     long startTime = System.currentTimeMillis();
132.     job.run();
133.     System.out.println("Job Finished in "
134.         + (System.currentTimeMillis() - startTime) / 1000.0 + " seconds");
135.   }
136. }