MaxCompute用户指南：图模型：示例程序：输入点表

输入点表的代码，如下所示：

1. import java.io.IOException;
3. import com.aliyun.odps.conf.Configuration;
4. import com.aliyun.odps.data.TableInfo;
5. import com.aliyun.odps.graph.ComputeContext;
6. import com.aliyun.odps.graph.GraphJob;
7. import com.aliyun.odps.graph.GraphLoader;
8. import com.aliyun.odps.graph.Vertex;
9. import com.aliyun.odps.graph.VertexResolver;
10. import com.aliyun.odps.graph.MutationContext;
11. import com.aliyun.odps.graph.VertexChanges;
12. import com.aliyun.odps.graph.Edge;
13. import com.aliyun.odps.io.LongWritable;
14. import com.aliyun.odps.io.WritableComparable;
15. import com.aliyun.odps.io.WritableRecord;
17. /**
18. * 本示例是用于展示，对于不同类型的数据类型，如何编写图作业程序载入数据。主要展示GraphLoader和
19. * VertexResolver的配合完成图的构建。
20. *
21. * ODPS Graph的作业输入都为ODPS的Table，假设作业输入有两张表，一张存储点的信息，一张存储边的信息。
22. * 存储点信息的表的格式，如：
23. * +------------------------+
24. * | VertexID | VertexValue |
25. * +------------------------+
26. * |       id0|            9|
27. * +------------------------+
28. * |       id1|            7|
29. * +------------------------+
30. * |       id2|            8|
31. * +------------------------+
32. *
33. * 存储边信息的表的格式，如
34. * +-----------------------------------+
35. * | VertexID | DestVertexID| EdgeValue|
36. * +-----------------------------------+
37. * |       id0|          id1|         1|
38. * +-----------------------------------+
39. * |       id0|          id2|         2|
40. * +-----------------------------------+
41. * |       id2|          id1|         3|
42. * +-----------------------------------+
43. *
44. * 结合两张表的数据，表示id0有两条出边，分别指向id1和id2；id2有一条出边，指向id1；id1没有出边。
45. *
46. * 对于此种类型的数据，在GraphLoader::load(LongWritable, Record, MutationContext)
47. * ，可以使用 MutationContext#addVertexRequest(Vertex)向图中请求添加点，使用
48. * link MutationContext#addEdgeRequest(WritableComparable, Edge)向图中请求添加边，然后，在
49. * link VertexResolver#resolve(WritableComparable, Vertex, VertexChanges, boolean)
50. * 中，将load 方法中添加的点和边，合并到一个Vertex对象中，作为返回值，添加到最后参与计算的图中。
51. *
52. **/
53. public class VertexInputFormat {
55.   private final static String EDGE_TABLE = "edge.table";
57.   /**
58.    * 将Record解释为Vertex和Edge，每个Record根据其来源，表示一个Vertex或者一条Edge。
59.    *
60.    * 类似于com.aliyun.odps.mapreduce.Mapper#map
61.    * ，输入Record，生成键值对，此处的键是Vertex的ID，
62.    * 值是Vertex或Edge，通过上下文Context写出，这些键值对会在LoadingVertexResolver出根据Vertex的ID汇总。
63.    *
64.    * 注意：此处添加的点或边只是根据Record内容发出的请求，并不是最后参与计算的点或边，只有在随后的VertexResolver
65.    * 中添加的点或边才参与计算。
66.    **/
67.   public static class VertexInputLoader extends
68.       GraphLoader<LongWritable, LongWritable, LongWritable, LongWritable> {
70.     private boolean isEdgeData;
72.     /**
73.      * 配置VertexInputLoader。
74.      *
75.      * @param conf
76.      *          作业的配置参数，在main中使用GraphJob配置的，或者在console中set的
77.      * @param workerId
78.      *          当前工作的worker的序号，从0开始，可以用于构造唯一的vertex id
79.      * @param inputTableInfo
80.      *          当前worker载入的输入表信息，可以用于确定当前输入是哪种类型的数据，即Record的格式
81.      **/
82.     @Override
83.     public void setup(Configuration conf, int workerId, TableInfo inputTableInfo) {
84.       isEdgeData = conf.get(EDGE_TABLE).equals(inputTableInfo.getTableName());
85.     }
87.     /**
88.      * 根据Record中的内容，解析为对应的边，并请求添加到图中。
89.      *
90.      * @param recordNum
91.      *          记录序列号，从1开始，每个worker上单独计数
92.      * @param record
93.      *          输入表中的记录，三列，分别表示初点、终点、边的权重
94.      * @param context
95.      *          上下文，请求将解释后的边添加到图中
96.      **/
97.     @Override
98.     public void load(
99.         LongWritable recordNum,
100.         WritableRecord record,
101.         MutationContext<LongWritable, LongWritable, LongWritable, LongWritable> context)
102.         throws IOException {
103.       if (isEdgeData) {
104.         /**
105.          * 数据来源于存储边信息的表。
106.          *
107.          * 1、第一列表示初始点的ID
108.          **/
109.         LongWritable sourceVertexID = (LongWritable) record.get(0);
111.         /**
112.          * 2、第二列表示终点的ID
113.          **/
114.         LongWritable destinationVertexID = (LongWritable) record.get(1);
116.         /**
117.          * 3、地三列表示边的权重
118.          **/
119.         LongWritable edgeValue = (LongWritable) record.get(2);
121.         /**
122.          * 4、创建边，由终点ID和边的权重组成
123.          **/
124.         Edge<LongWritable, LongWritable> edge = new Edge<LongWritable, LongWritable>(
125.             destinationVertexID, edgeValue);
127.         /**
128.          * 5、请求给初始点添加边
129.          **/
130.         context.addEdgeRequest(sourceVertexID, edge);
132.         /**
133.          * 6、如果每条Record表示双向边，重复4与5 Edge<LongWritable, LongWritable> edge2 = new
134.          * Edge<LongWritable, LongWritable>( sourceVertexID, edgeValue);
135.          * context.addEdgeRequest(destinationVertexID, edge2);
136.          **/
137.       } else {
138.         /**
139.          * 数据来源于存储点信息的表。
140.          *
141.          * 1、第一列表示点的ID
142.          **/
143.         LongWritable vertexID = (LongWritable) record.get(0);
145.         /**
146.          * 2、第二列表示点的值
147.          **/
148.         LongWritable vertexValue = (LongWritable) record.get(1);
150.         /**
151.          * 3、创建点，由点的ID和点的值组成
152.          **/
153.         MyVertex vertex = new MyVertex();
155.         /**
156.          * 4、初始化点
157.          **/
158.         vertex.setId(vertexID);
159.         vertex.setValue(vertexValue);
161.         /**
162.          * 5、请求添加点
163.          **/
164.         context.addVertexRequest(vertex);
165.       }
167.     }
169.   }
171.   /**
172.    * 汇总GraphLoader::load(LongWritable, Record, MutationContext)生成的键值对，类似于
173.    * com.aliyun.odps.mapreduce.Reducer中的reduce。对于唯一的Vertex ID，所有关于这个ID上
174.    * 添加\删除、点\边的行为都会放在VertexChanges中。
175.    *
176.    * 注意：此处并不只针对load方法中添加的有冲突的点或边才调用（冲突是指添加多个相同Vertex对象，添加重复边等），
177.    * 所有在load方法中请求生成的ID都会在此处被调用。
178.    **/
179.   public static class LoadingResolver extends
180.       VertexResolver<LongWritable, LongWritable, LongWritable, LongWritable> {
182.     /**
183.      * 处理关于一个ID的添加或删除、点或边的请求。
184.      *
185.      * VertexChanges有四个接口，分别与MutationContext的四个接口对应：
186.      * VertexChanges::getAddedVertexList()与
187.      * MutationContext::addVertexRequest(Vertex)对应，
188.      * 在load方法中，请求添加的ID相同的Vertex对象，会被汇总在返回的List中
189.      * VertexChanges::getAddedEdgeList()与
190.      * MutationContext::addEdgeRequest(WritableComparable, Edge)
191.      * 对应，请求添加的初始点ID相同的Edge对象，会被汇总在返回的List中
192.      * VertexChanges::getRemovedVertexCount()与
193.      * MutationContext::removeVertexRequest(WritableComparable)
194.      * 对应，请求删除的ID相同的Vertex，汇总的请求删除的次数作为返回值
195.      * VertexChanges#getRemovedEdgeList()与
196.      * MutationContext#removeEdgeRequest(WritableComparable, WritableComparable)
197.      * 对应，请求删除的初始点ID相同的Edge对象，会被汇总在返回的List中
198.      *
199.      * 用户通过处理关于这个ID的变化，通过返回值声明此ID是否参与计算，如果返回的Vertex不为null，
200.      * 则此ID会参与随后的计算，如果返回null，则不会参与计算。
201.      *
202.      * @param vertexId
203.      *          请求添加的点的ID，或请求添加的边的初点ID
204.      * @param vertex
205.      *          已存在的Vertex对象，数据载入阶段，始终为null
206.      * @param vertexChanges
207.      *          此ID上的请求添加\删除、点\边的集合
208.      * @param hasMessages
209.      *          此ID是否有输入消息，数据载入阶段，始终为false
210.      **/
211.     @Override
212.     public Vertex<LongWritable, LongWritable, LongWritable, LongWritable> resolve(
213.         LongWritable vertexId,
214.         Vertex<LongWritable, LongWritable, LongWritable, LongWritable> vertex,
215.         VertexChanges<LongWritable, LongWritable, LongWritable, LongWritable> vertexChanges,
216.         boolean hasMessages) throws IOException {
217.       /**
218.        * 1、获取Vertex对象，作为参与计算的点。
219.        **/
220.       MyVertex computeVertex = null;
221.       if (vertexChanges.getAddedVertexList() == null
222.           || vertexChanges.getAddedVertexList().isEmpty()) {
223.         computeVertex = new MyVertex();
224.         computeVertex.setId(vertexId);
225.       } else {
226.         /**
227.          * 此处假设存储点信息的表中，每个Record表示唯一的点。
228.          **/
229.         computeVertex = (MyVertex) vertexChanges.getAddedVertexList().get(0);
230.       }
232.       /**
233.        * 2、将请求给此点添加的边，添加到Vertex对象中，如果数据有重复的可能，根据算法需要决定是否去重。
234.        **/
235.       if (vertexChanges.getAddedEdgeList() != null) {
236.         for (Edge<LongWritable, LongWritable> edge : vertexChanges
237.             .getAddedEdgeList()) {
238.           computeVertex.addEdge(edge.getDestVertexId(), edge.getValue());
239.         }
240.       }
242.       /**
243.        * 3、将Vertex对象返回，添加到最终的图中参与计算。
244.        **/
245.       return computeVertex;
246.     }
248.   }
250.   /**
251.    * 确定参与计算的Vertex的行为。
252.    *
253.    **/
254.   public static class MyVertex extends
255.       Vertex<LongWritable, LongWritable, LongWritable, LongWritable> {
257.     /**
258.      * 将vertex的边，按照输入表的格式再写到结果表。输入表与输出表的格式和数据都相同。
259.      *
260.      * @param context
261.      *          运行时上下文
262.      * @param messages
263.      *          输入消息
264.      **/
265.     @Override
266.     public void compute(
267.         ComputeContext<LongWritable, LongWritable, LongWritable, LongWritable> context,
268.         Iterable<LongWritable> messages) throws IOException {
269.       /**
270.        * 将点的ID和值，写到存储点的结果表
271.        **/
272.       context.write("vertex", getId(), getValue());
274.       /**
275.        * 将点的边，写到存储边的结果表
276.        **/
277.       if (hasEdges()) {
278.         for (Edge<LongWritable, LongWritable> edge : getEdges()) {
279.           context.write("edge", getId(), edge.getDestVertexId(),
280.               edge.getValue());
281.         }
282.       }
284.       /**
285.        * 只迭代一轮
286.        **/
287.       voteToHalt();
288.     }
290.   }
292.   /**
293.    * @param args
294.    * @throws IOException
295.    */
296.   public static void main(String[] args) throws IOException {
298.     if (args.length < 4) {
299.       throw new IOException(
300.           "Usage: VertexInputFormat <vertex input> <edge input> <vertex output> <edge output>");
301.     }
303.     /**
304.      * GraphJob用于对Graph作业进行配置
305.      */
306.     GraphJob job = new GraphJob();
308.     /**
309.      * 1、指定输入的图数据，并指定存储边数据所在的表。
310.      */
311.     job.addInput(TableInfo.builder().tableName(args[0]).build());
312.     job.addInput(TableInfo.builder().tableName(args[1]).build());
313.     job.set(EDGE_TABLE, args[1]);
315.     /**
316.      * 2、指定载入数据的方式，将Record解释为Edge，此处类似于map，生成的 key为vertex的ID，value为edge。
317.      */
318.     job.setGraphLoaderClass(VertexInputLoader.class);
320.     /**
321.      * 3、指定载入数据阶段，生成参与计算的vertex。此处类似于reduce，将map 生成的edge合并成一个vertex。
322.      */
323.     job.setLoadingVertexResolverClass(LoadingResolver.class);
325.     /**
326.      * 4、指定参与计算的vertex的行为。每轮迭代执行vertex.compute方法。
327.      */
328.     job.setVertexClass(MyVertex.class);
330.     /**
331.      * 5、指定图作业的输出表，将计算生成的结果写到结果表中。
332.      */
333.     job.addOutput(TableInfo.builder().tableName(args[2]).label("vertex").build());
334.     job.addOutput(TableInfo.builder().tableName(args[3]).label("edge").build());
336.     /**
337.      * 6、提交作业执行。
338.      */
339.     job.run();
340.   }
342. }