MaxCompute用户指南：图模型：示例程序：连通分量

两个顶点之间存在路径，称两个顶点为连通的。如果无向图 G 中任意两个顶点都是连通的，则称 G 为连通图，否则称为非连通图。其顶点个数极大的连通子图称为连通分量。
本算法计算每个点的连通分量成员，最后输出顶点值中包含最小顶点 ID 的连通分量。将最小顶点 ID 沿着边传播到连通分量的所有顶点。

代码示例


连通分量的代码，如下所示：

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.GraphJob;
6. import com.aliyun.odps.graph.GraphLoader;
7. import com.aliyun.odps.graph.MutationContext;
8. import com.aliyun.odps.graph.Vertex;
9. import com.aliyun.odps.graph.WorkerContext;
10. import com.aliyun.odps.graph.examples.SSSP.MinLongCombiner;
11. import com.aliyun.odps.io.LongWritable;
12. import com.aliyun.odps.io.NullWritable;
13. import com.aliyun.odps.io.WritableRecord;
15. /**
16. * Compute the connected component membership of each vertex and output
17. * each vertex which's value containing the smallest id in the connected
18. * component containing that vertex.
19. *
20. * Algorithm: propagate the smallest vertex id along the edges to all
21. * vertices of a connected component.
22. *
23. */
24. public class ConnectedComponents {
26.   public static class CCVertex extends
27.     Vertex<LongWritable, LongWritable, NullWritable, LongWritable> {
29.     @Override
30.     public void compute(
31.         ComputeContext<LongWritable, LongWritable, NullWritable, LongWritable> context,
32.         Iterable<LongWritable> msgs) throws IOException {
34.       if (context.getSuperstep() == 0L) {
35.         this.setValue(getId());
36.         context.sendMessageToNeighbors(this, getValue());
37.         return;
38.       }
40.       long minID = Long.MAX_VALUE;
41.       for (LongWritable id : msgs) {
42.         if (id.get() < minID) {
43.           minID = id.get();
44.         }
45.       }
47.       if (minID < this.getValue().get()) {
48.         this.setValue(new LongWritable(minID));
49.         context.sendMessageToNeighbors(this, getValue());
50.       } else {
51.         this.voteToHalt();
52.       }
53.     }
55.     /**
56.      * Output Table Description:
57.      * +-----------------+----------------------------------------+
58.      * | Field | Type    | Comment                                |
59.      * +-----------------+----------------------------------------+
60.      * | v     | bigint  | vertex id                              |
61.      * | minID | bigint  | smallest id in the connected component |
62.      * +-----------------+----------------------------------------+
63.      */
64.     @Override
65.     public void cleanup(
66.         WorkerContext<LongWritable, LongWritable, NullWritable, LongWritable> context)
67.         throws IOException {
68.       context.write(getId(), getValue());
69.     }
70.   }
73.   /**
74.    * Input Table Description:
75.    * +-----------------+----------------------------------------------------+
76.    * | Field | Type    | Comment                                            |
77.    * +-----------------+----------------------------------------------------+
78.    * | v     | bigint  | vertex id                                          |
79.    * | es    | string  | comma separated target vertex id of outgoing edges |
80.    * +-----------------+----------------------------------------------------+
81.    *
82.    * Example:
83.    * For graph:
84.    *       1 ----- 2
85.    *       |       |
86.    *       3 ----- 4
87.    * Input table:
88.    * +-----------+
89.    * | v  | es   |
90.    * +-----------+
91.    * | 1  | 2,3  |
92.    * | 2  | 1,4  |
93.    * | 3  | 1,4  |
94.    * | 4  | 2,3  |
95.    * +-----------+
96.    */
97.   public static class CCVertexReader extends
98.     GraphLoader<LongWritable, LongWritable, NullWritable, LongWritable> {
100.     @Override
101.     public void load(
102.         LongWritable recordNum,
103.         WritableRecord record,
104.         MutationContext<LongWritable, LongWritable, NullWritable, LongWritable> context)
105.     throws IOException {
106.       CCVertex vertex = new CCVertex();
108.       vertex.setId((LongWritable) record.get(0));
110.       String[] edges = record.get(1).toString().split(",");
111.       for (int i = 0; i < edges.length; i++) {
112.         long destID = Long.parseLong(edges);
113.         vertex.addEdge(new LongWritable(destID), NullWritable.get());
114.       }
116.       context.addVertexRequest(vertex);
117.     }
119.   }
121.   public static void main(String[] args) throws IOException {
122.     if (args.length < 2) {
123.       System.out.println("Usage: <input> <output>");
124.       System.exit(-1);
125.     }
127.     GraphJob job = new GraphJob();
128.     job.setGraphLoaderClass(CCVertexReader.class);
129.     job.setVertexClass(CCVertex.class);
130.     job.setCombinerClass(MinLongCombiner.class);
132.     job.addInput(TableInfo.builder().tableName(args[0]).build());
133.     job.addOutput(TableInfo.builder().tableName(args[1]).build());
134.     long startTime = System.currentTimeMillis();
135.     job.run();
136.     System.out.println("Job Finished in "
137.         + (System.currentTimeMillis() - startTime) / 1000.0 + " seconds");
138.   }
139. }