MaxCompute用户指南：图模型：示例程序：PageRank

PageRank 算法是计算网页排名的经典算法：输入是一个有向图 G，其中顶点表示网页，如果存在网页 A 到网页 B 的链接，那么存在连接 A 到 B 的边。
算法基本原理，如下所示：

• 
  初始化：点值表示 PageRank 的 rank 值（double 类型），初始时，所有点取值为 1/TotalNumVertices。
• 
  迭代公式：PageRank(i)=0.15/TotalNumVertices+0.85*sum，其中 sum 为所有指向 i 点的点（设为 j） PageRank(j)/out_degree(j) 的累加值。

1. import java.io.IOException;
3. import org.apache.log4j.Logger;
5. import com.aliyun.odps.io.WritableRecord;
6. import com.aliyun.odps.graph.ComputeContext;
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.DoubleWritable;
13. import com.aliyun.odps.io.LongWritable;
14. import com.aliyun.odps.io.NullWritable;
15. import com.aliyun.odps.data.TableInfo;
16. import com.aliyun.odps.io.Text;
17. import com.aliyun.odps.io.Writable;
19. public class PageRank {
21.   private final static Logger LOG = Logger.getLogger(PageRank.class);
23.   public static class PageRankVertex extends
24.       Vertex<Text, DoubleWritable, NullWritable, DoubleWritable> {
26.     @Override
27.     public void compute(
28.         ComputeContext<Text, DoubleWritable, NullWritable, DoubleWritable> context,
29.         Iterable<DoubleWritable> messages) throws IOException {
30.       if (context.getSuperstep() == 0) {
31.         setValue(new DoubleWritable(1.0 / context.getTotalNumVertices()));
32.       } else if (context.getSuperstep() >= 1) {
33.         double sum = 0;
34.         for (DoubleWritable msg : messages) {
35.           sum += msg.get();
36.         }
37.         DoubleWritable vertexValue = new DoubleWritable(
38.             (0.15f / context.getTotalNumVertices()) + 0.85f * sum);
39.         setValue(vertexValue);
40.       }
41.       if (hasEdges()) {
42.         context.sendMessageToNeighbors(this, new DoubleWritable(getValue()
43.             .get() / getEdges().size()));
44.       }
45.     }
47.     @Override
48.     public void cleanup(
49.         WorkerContext<Text, DoubleWritable, NullWritable, DoubleWritable> context)
50.         throws IOException {
51.       context.write(getId(), getValue());
52.     }
53.   }
55.   public static class PageRankVertexReader extends
56.       GraphLoader<Text, DoubleWritable, NullWritable, DoubleWritable> {
58.     @Override
59.     public void load(
60.         LongWritable recordNum,
61.         WritableRecord record,
62.         MutationContext<Text, DoubleWritable, NullWritable, DoubleWritable> context)
63.         throws IOException {
64.       PageRankVertex vertex = new PageRankVertex();
65.       vertex.setValue(new DoubleWritable(0));
66.       vertex.setId((Text) record.get(0));
67.       System.out.println(record.get(0));
69.       for (int i = 1; i < record.size(); i++) {
70.         Writable edge = record.get(i);
71.         System.out.println(edge.toString());
72.         if (!(edge.equals(NullWritable.get()))) {
73.           vertex.addEdge(new Text(edge.toString()), NullWritable.get());
74.         }
75.       }
76.       LOG.info("vertex edgs size: "
77.           + (vertex.hasEdges() ? vertex.getEdges().size() : 0));
78.       context.addVertexRequest(vertex);
79.     }
81.   }
83.   private static void printUsage() {
84.     System.out.println("Usage: <in> <out> [Max iterations (default 30)]");
85.     System.exit(-1);
86.   }
88.   public static void main(String[] args) throws IOException {
89.     if (args.length < 2)
90.       printUsage();
92.     GraphJob job = new GraphJob();
94.     job.setGraphLoaderClass(PageRankVertexReader.class);
95.     job.setVertexClass(PageRankVertex.class);
96.     job.addInput(TableInfo.builder().tableName(args[0]).build());
97.     job.addOutput(TableInfo.builder().tableName(args[1]).build());
99.     // default max iteration is 30
100.     job.setMaxIteration(30);
101.     if (args.length >= 3)
102.       job.setMaxIteration(Integer.parseInt(args[2]));
104.     long startTime = System.currentTimeMillis();
105.     job.run();
106.     System.out.println("Job Finished in "
107.         + (System.currentTimeMillis() - startTime) / 1000.0 + " seconds");
108.   }
109. }

• 
  第 23 行：定义 PageRankVertex ，其中：
  点值表示该点（网页）的当前 PageRank 取值。
• 
  compute() 方法使用迭代公式：PageRank(i)=0.15/TotalNumVertices+0.85*sum更新点值。
• 
  cleanup() 方法把点及其 PageRank 取值写到结果表中。