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

二分图是指图的所有顶点可分为两个集合，每条边对应的两个顶点分别属于这两个集合。对于一个二分图 G，M 是它的一个子图，如果 M的边集中任意两条边都不依附于同一个顶点，则称 M 为一个匹配。二分图匹配常用于有明确供需关系场景（如交友网站等）下的信息匹配行为。
算法描述，如下所示：

• 
  从左边第 1 个顶点开始，挑选未匹配点进行搜索，寻找增广路。
• 
  如果经过一个未匹配点，说明寻找成功。
• 
  更新路径信息，匹配边数 +1，停止搜索。
• 
  如果一直没有找到增广路，则不再从这个点开始搜索。

1. import java.io.DataInput;
2. import java.io.DataOutput;
3. import java.io.IOException;
4. import java.util.Random;
6. import com.aliyun.odps.data.TableInfo;
7. import com.aliyun.odps.graph.ComputeContext;
8. import com.aliyun.odps.graph.GraphJob;
9. import com.aliyun.odps.graph.MutationContext;
10. import com.aliyun.odps.graph.WorkerContext;
11. import com.aliyun.odps.graph.Vertex;
12. import com.aliyun.odps.graph.GraphLoader;
13. import com.aliyun.odps.io.LongWritable;
14. import com.aliyun.odps.io.NullWritable;
15. import com.aliyun.odps.io.Text;
16. import com.aliyun.odps.io.Writable;
17. import com.aliyun.odps.io.WritableRecord;
19. public class BipartiteMatching {
21.   private static final Text UNMATCHED = new Text("UNMATCHED");
23.   public static class TextPair implements Writable {
25.     public Text first;
26.     public Text second;
28.     public TextPair() {
29.       first = new Text();
30.       second = new Text();
31.     }
33.     public TextPair(Text first, Text second) {
34.       this.first = new Text(first);
35.       this.second = new Text(second);
36.     }
38.     @Override
39.     public void write(DataOutput out) throws IOException {
40.       first.write(out);
41.       second.write(out);
42.     }
44.     @Override
45.     public void readFields(DataInput in) throws IOException {
46.       first = new Text();
47.       first.readFields(in);
48.       second = new Text();
49.       second.readFields(in);
50.     }
52.     @Override
53.     public String toString() {
54.       return first + ": " + second;
55.     }
57.   }
59.   public static class BipartiteMatchingVertexReader extends
60.       GraphLoader<Text, TextPair, NullWritable, Text> {
62.     @Override
63.     public void load(LongWritable recordNum, WritableRecord record,
64.         MutationContext<Text, TextPair, NullWritable, Text> context)
65.         throws IOException {
66.       BipartiteMatchingVertex vertex = new BipartiteMatchingVertex();
67.       vertex.setId((Text) record.get(0));
68.       vertex.setValue(new TextPair(UNMATCHED, (Text) record.get(1)));
70.       String[] adjs = record.get(2).toString().split(",");
71.       for (String adj : adjs) {
72.         vertex.addEdge(new Text(adj), null);
73.       }
74.       context.addVertexRequest(vertex);
75.     }
77.   }
79.   public static class BipartiteMatchingVertex extends
80.       Vertex<Text, TextPair, NullWritable, Text> {
82.     private static final Text LEFT = new Text("LEFT");
83.     private static final Text RIGHT = new Text("RIGHT");
85.     private static Random rand = new Random();
87.     @Override
88.     public void compute(
89.         ComputeContext<Text, TextPair, NullWritable, Text> context,
90.         Iterable<Text> messages) throws IOException {
91.       if (isMatched()) {
92.         voteToHalt();
93.         return;
94.       }
96.       switch ((int) context.getSuperstep() % 4) {
97.       case 0:
98.         if (isLeft()) {
99.           context.sendMessageToNeighbors(this, getId());
100.         }
101.         break;
102.       case 1:
103.         if (isRight()) {
104.           Text luckyLeft = null;
105.           for (Text message : messages) {
106.             if (luckyLeft == null) {
107.               luckyLeft = new Text(message);
108.             } else {
109.               if (rand.nextInt(1) == 0) {
110.                 luckyLeft.set(message);
111.               }
112.             }
113.           }
114.           if (luckyLeft != null) {
115.             context.sendMessage(luckyLeft, getId());
116.           }
117.         }
118.         break;
119.       case 2:
120.         if (isLeft()) {
121.           Text luckyRight = null;
122.           for (Text msg : messages) {
123.             if (luckyRight == null) {
124.               luckyRight = new Text(msg);
125.             } else {
126.               if (rand.nextInt(1) == 0) {
127.                 luckyRight.set(msg);
128.               }
129.             }
130.           }
131.           if (luckyRight != null) {
132.             setMatchVertex(luckyRight);
133.             context.sendMessage(luckyRight, getId());
134.           }
135.         }
136.         break;
137.       case 3:
138.         if (isRight()) {
139.           for (Text msg : messages) {
140.             setMatchVertex(msg);
141.           }
142.         }
143.         break;
144.       }
145.     }
147.     @Override
148.     public void cleanup(
149.         WorkerContext<Text, TextPair, NullWritable, Text> context)
150.         throws IOException {
151.       context.write(getId(), getValue().first);
152.     }
154.     private boolean isMatched() {
155.       return !getValue().first.equals(UNMATCHED);
156.     }
158.     private boolean isLeft() {
159.       return getValue().second.equals(LEFT);
160.     }
162.     private boolean isRight() {
163.       return getValue().second.equals(RIGHT);
164.     }
166.     private void setMatchVertex(Text matchVertex) {
167.       getValue().first.set(matchVertex);
168.     }
169.   }
171.   private static void printUsage() {
172.     System.err.println("BipartiteMatching <input> <output> [maxIteration]");
173.   }
175.   public static void main(String[] args) throws IOException {
176.     if (args.length < 2) {
177.       printUsage();
178.     }
180.     GraphJob job = new GraphJob();
182.     job.setGraphLoaderClass(BipartiteMatchingVertexReader.class);
183.     job.setVertexClass(BipartiteMatchingVertex.class);
185.     job.addInput(TableInfo.builder().tableName(args[0]).build());
186.     job.addOutput(TableInfo.builder().tableName(args[1]).build());
187.     int maxIteration = 30;
188.     if (args.length > 2) {
189.       maxIteration = Integer.parseInt(args[2]);
190.     }
191.     job.setMaxIteration(maxIteration);
193.     job.run();
194.   }
196. }