View Javadoc

1   /*
2    * Licensed to the Apache Software Foundation (ASF) under one
3    * or more contributor license agreements.  See the NOTICE file
4    * distributed with this work for additional information
5    * regarding copyright ownership.  The ASF licenses this file
6    * to you under the Apache License, Version 2.0 (the
7    * "License"); you may not use this file except in compliance
8    * with the License.  You may obtain a copy of the License at
9    *
10   *     http://www.apache.org/licenses/LICENSE-2.0
11   *
12   * Unless required by applicable law or agreed to in writing, software
13   * distributed under the License is distributed on an "AS IS" BASIS,
14   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15   * See the License for the specific language governing permissions and
16   * limitations under the License.
17   */
18  package org.apache.giraph.io.gora;
19  
20  import java.io.IOException;
21  import java.util.Set;
22  
23  import org.apache.giraph.edge.Edge;
24  import org.apache.giraph.edge.EdgeFactory;
25  import org.apache.giraph.graph.Vertex;
26  import org.apache.giraph.io.gora.generated.GVertex;
27  import org.apache.hadoop.io.DoubleWritable;
28  import org.apache.hadoop.io.FloatWritable;
29  import org.apache.hadoop.io.LongWritable;
30  import org.apache.hadoop.mapreduce.InputSplit;
31  import org.apache.hadoop.mapreduce.TaskAttemptContext;
32  
33  /**
34   * Example implementation of a specific reader for a generated data bean.
35   */
36  public class GoraGVertexVertexInputFormat
37    extends GoraVertexInputFormat<LongWritable, DoubleWritable,
38            FloatWritable> {
39  
40    /**
41     * DEfault constructor
42     */
43    public GoraGVertexVertexInputFormat() {
44    }
45  
46    /**
47     * Creates specific vertex reader to be used inside Hadoop.
48     * @param split split to be read.
49     * @param context JobContext to be used.
50     * @return GoraVertexReader Vertex reader to be used by Hadoop.
51     */
52    @Override
53    public GoraVertexReader createVertexReader(
54        InputSplit split, TaskAttemptContext context) throws IOException {
55      return new GoraGVertexVertexReader();
56    }
57  
58    /**
59     * Gora vertex reader
60     */
61    protected class GoraGVertexVertexReader extends GoraVertexReader {
62  
63      /**
64       * Transforms a GoraObject into a Vertex object.
65       * @param goraObject Object from Gora to be translated.
66       * @return Vertex Result from transforming the gora object.
67       */
68      @Override
69      protected Vertex<LongWritable, DoubleWritable, FloatWritable>
70      transformVertex(Object goraObject) {
71        Vertex<LongWritable, DoubleWritable, FloatWritable> vertex;
72        /* create the actual vertex */
73        vertex = getConf().createVertex();
74        GVertex tmpGVertex = (GVertex) goraObject;
75  
76        LongWritable vrtxId = new LongWritable(
77          Long.parseLong(tmpGVertex.getVertexId().toString()));
78        DoubleWritable vrtxValue = new DoubleWritable(
79          tmpGVertex.getVertexValue());
80        vertex.initialize(vrtxId, vrtxValue);
81        if (tmpGVertex.getEdges() != null && !tmpGVertex.getEdges().isEmpty()) {
82          Set<CharSequence> keyIt = tmpGVertex.getEdges().keySet();
83          for (CharSequence key : keyIt) {
84            String keyVal = key.toString();
85            String valVal = tmpGVertex.getEdges().get(key).toString();
86            Edge<LongWritable, FloatWritable> edge;
87            if (!keyVal.contains("vertexId") && !keyVal.contains("value")) {
88              edge = EdgeFactory.create(
89                  new LongWritable(Long.parseLong(keyVal)),
90                  new FloatWritable(Float.parseFloat(valVal)));
91              vertex.addEdge(edge);
92            }
93          }
94        }
95        return vertex;
96      }
97    }
98  }