package com.ibm.giraph.graph.example.mapred;


import java.io.*;
import java.util.Vector;

import com.ibm.giraph.graph.example.ioformats.KVBinaryOutputFormat;
import com.ibm.giraph.graph.example.ioformats.SkeletonNeighborhood;

import org.apache.giraph.graph.LongDoubleDoubleNeighborhood;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.*;
import org.apache.hadoop.mapreduce.lib.output.*;
import org.apache.hadoop.util.GenericOptionsParser;

public class TextBinaryEdgeValueMerge {
	public static class SMapper extends
			Mapper<LongWritable, Text, LongWritable, Text> {

		public void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			
			String[] tokens = value.toString().split("\\s+");
			
			context.write(new LongWritable(Long.parseLong(tokens[0])), value);
			
		}
	}
    public static class SReducer extends
			Reducer<LongWritable, Text, LongWritable, LongDoubleDoubleNeighborhood> {

		public void reduce(LongWritable key, Iterable<Text> values, Context context)
				throws IOException, InterruptedException {
			
			LongDoubleDoubleNeighborhood val = new LongDoubleDoubleNeighborhood();
			
			
			String[] nbs = values.iterator().next().toString().split("\\s+");
			
			long[] edges = new long[Integer.parseInt(nbs[1])];
			double[] edgeValues = new double[edges.length];
			for(int i = 2 ; i < nbs.length ; i +=2)
			{
				edges[i] =Long.parseLong(nbs[i]);
				edgeValues[i] =Long.parseLong(nbs[i+1]);
			}
			
			val.setSimpleEdges(edges, edgeValues);
			 
			context.write(new LongWritable(Long.parseLong(nbs[0])), val);
			
		}
	}
	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		Job job = new Job(conf, "TextBinaryMerge");
		String[] otherArgs = new GenericOptionsParser(conf, args)
				.getRemainingArgs();
		if (otherArgs.length != 3) {
			System.err.println("Args: <in> <out> <num>");
			System.exit(3);
		}
		
		job.setJarByClass(TextBinaryEdgeValueMerge.class);
		
		FileInputFormat.addInputPath(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job,new Path(args[1]));
		
		job.setMapperClass(SMapper.class);
		job.setInputFormatClass(TextInputFormat.class);
		job.setMapOutputKeyClass(LongWritable.class);// cnt
		job.setMapOutputValueClass(Text.class);// vid
		
		job.setReducerClass(SReducer.class);
		
		job.setOutputFormatClass(KVBinaryOutputFormat.class);
		job.setOutputKeyClass(LongWritable.class);
		job.setOutputValueClass(LongDoubleDoubleNeighborhood.class);
		
		job.setNumReduceTasks(Integer.parseInt(args[2]));
		
		System.exit(job.waitForCompletion(true) ? 0 : 1);
	}
}
@@ -1457,60 +1457,89 @@
 %0A%09%09%09
-%0A%09%09%09long%5B%5D edges = new long%5BInteger.parseInt(nbs%5B1%5D)
+int numofEdges =  Integer.parseInt(nbs%5B1%5D);%0A%09%09%09long%5B%5D edges = new long%5BnumofEdges
 %5D;%0A%09
@@ -1577,20 +1577,18 @@
 ble%5B
-edges.length
+numofEdges
 %5D;%0A%09
@@ -1605,17 +1605,17 @@
 i = 
-2
+0
  ; i %3C n
 bs.l
@@ -1614,25 +1614,24 @@
  %3C n
-bs.length
+umofEdges
  ; i +
-=2
++
 )%0A%09%09
@@ -1666,17 +1666,21 @@
 ong(nbs%5B
-i
+2*i+2
 %5D);%0A%09%09%09%09
@@ -1717,11 +1717,13 @@
 nbs%5B
+2*
 i+
-1
+3
 %5D);%0A

fix a bug