1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18 package org.apache.giraph.jython;
19
20 import org.apache.giraph.graph.AbstractComputation;
21 import org.apache.giraph.graph.GraphType;
22 import org.apache.giraph.graph.Language;
23 import org.apache.giraph.graph.Vertex;
24 import org.apache.giraph.jython.wrappers.JythonWritableWrapper;
25 import org.apache.hadoop.io.Writable;
26 import org.apache.hadoop.io.WritableComparable;
27 import org.python.core.PyObject;
28
29 import com.google.common.base.Preconditions;
30
31 import java.io.IOException;
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48 public class JythonGiraphComputation<I extends WritableComparable,
49 V extends Writable, E extends Writable, M1 extends Writable,
50 M2 extends Writable>
51 extends AbstractComputation<I, V, E, M1, M2> {
52
53 private final JythonComputation jythonComputation;
54
55
56
57
58
59
60 public JythonGiraphComputation(JythonComputation jythonComputation) {
61 this.jythonComputation = jythonComputation;
62 }
63
64 @Override public void compute(Vertex<I, V, E> vertex, Iterable<M1> messages)
65 throws IOException {
66 jythonComputation.compute(vertex, messages);
67 }
68
69
70
71
72
73
74
75 public WritableComparable wrapIdIfNecessary(Object object) {
76 return wrapIfNecessary(object, GraphType.VERTEX_ID);
77 }
78
79
80
81
82
83
84
85
86
87 public <W extends Writable> W
88 wrapIfNecessary(Object object, GraphType graphType) {
89 if (graphType.interfaceClass().isInstance(object)) {
90 return (W) object;
91 }
92 if (getConf().getValueLanguages().get(graphType) == Language.JYTHON &&
93 getConf().getValueNeedsWrappers().get(graphType)) {
94 Preconditions.checkArgument(object instanceof PyObject);
95 return (W) new JythonWritableWrapper((PyObject) object);
96 } else {
97 return (W) object;
98 }
99 }
100 }