1/*2 * Licensed to the Apache Software Foundation (ASF) under one3 * or more contributor license agreements. See the NOTICE file4 * distributed with this work for additional information5 * regarding copyright ownership. The ASF licenses this file6 * to you under the Apache License, Version 2.0 (the7 * "License"); you may not use this file except in compliance8 * with the License. You may obtain a copy of the License at9 *10 * http://www.apache.org/licenses/LICENSE-2.011 *12 * Unless required by applicable law or agreed to in writing, software13 * 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 and16 * limitations under the License.17 */18package org.apache.giraph.io.gora;
1920import java.io.IOException;
21import java.util.Set;
2223import org.apache.giraph.edge.Edge;
24import org.apache.giraph.edge.EdgeFactory;
25import org.apache.giraph.graph.Vertex;
26import org.apache.giraph.io.gora.generated.GVertex;
27import org.apache.hadoop.io.DoubleWritable;
28import org.apache.hadoop.io.FloatWritable;
29import org.apache.hadoop.io.LongWritable;
30import org.apache.hadoop.mapreduce.InputSplit;
31import org.apache.hadoop.mapreduce.TaskAttemptContext;
3233/**34 * Example implementation of a specific reader for a generated data bean.35 */36publicclassGoraGVertexVertexInputFormat37extends GoraVertexInputFormat<LongWritable, DoubleWritable,
38 FloatWritable> {
3940/**41 * DEfault constructor42 */43publicGoraGVertexVertexInputFormat() {
44 }
4546/**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
53publicGoraVertexReader createVertexReader(
54 InputSplit split, TaskAttemptContext context) throws IOException {
55returnnewGoraGVertexVertexReader();
56 }
5758/**59 * Gora vertex reader60 */61protectedclassGoraGVertexVertexReaderextendsGoraVertexReader {
6263/**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
69protected Vertex<LongWritable, DoubleWritable, FloatWritable>
70 transformVertex(Object goraObject) {
71 Vertex<LongWritable, DoubleWritable, FloatWritable> vertex;
72/* create the actual vertex */73 vertex = getConf().createVertex();
74GVertex tmpGVertex = (GVertex) goraObject;
7576 LongWritable vrtxId = new LongWritable(
77 Long.parseLong(tmpGVertex.getVertexId().toString()));
78 DoubleWritable vrtxValue = new DoubleWritable(
79 tmpGVertex.getVertexValue());
80 vertex.initialize(vrtxId, vrtxValue);
81if (tmpGVertex.getEdges() != null && !tmpGVertex.getEdges().isEmpty()) {
82 Set<CharSequence> keyIt = tmpGVertex.getEdges().keySet();
83for (CharSequence key : keyIt) {
84 String keyVal = key.toString();
85 String valVal = tmpGVertex.getEdges().get(key).toString();
86 Edge<LongWritable, FloatWritable> edge;
87if (!keyVal.contains("vertexId") && !keyVal.contains("value")) {
88 edge = EdgeFactory.create(
89new LongWritable(Long.parseLong(keyVal)),
90new FloatWritable(Float.parseFloat(valVal)));
91 vertex.addEdge(edge);
92 }
93 }
94 }
95return vertex;
96 }
97 }
98 }