Java hadoop question

given Friends.java

import java.io.IOException; import java.util.StringTokenizer; import java.util.Set; import java.util.HashSet; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; 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.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.apache.hadoop.util.GenericOptionsParser; public class FriendsOfFriends { public static class FriendMapper extends Mapper{ private int mark_red = -1; public void map(Object key, Text value, Context context ) throws IOException, InterruptedException { StringTokenizer itr = new StringTokenizer(value.toString()); String one_string = itr.nextToken(); String two_string = itr.nextToken(); int one_integer = Integer.parseInt(one_string); int two_integer = Integer.parseInt(two_string); /* << your code here >> */ context.write(new IntWritable(one_integer), new IntWritable(two_integer)); } } public static class FriendReducer extends Reducer { private int mark_red = -1; public void reduce(IntWritable key, Iterable values, Context context ) throws IOException, InterruptedException { Set srcs = new HashSet(); Set dest = new HashSet(); /* << your code here >> */ } } public static void main(String[] args) throws Exception { Configuration conf = new Configuration(); String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs(); if (otherArgs.length < 2) { System.err.println("Usage: FriendsOfFriends  [...] "); System.exit(2); } Job job = Job.getInstance(conf, "Friends of Friends 2023"); job.setJarByClass(FriendsOfFriends.class); job.setMapperClass(FriendMapper.class); job.setReducerClass(FriendReducer.class); job.setOutputKeyClass(IntWritable.class); job.setOutputValueClass(IntWritable.class); for (int i = 0; i < otherArgs.length - 1; ++i) { FileInputFormat.addInputPath(job, new Path(otherArgs[i])); } FileOutputFormat.setOutputPath(job, new Path(otherArgs[otherArgs.length - 1])); System.exit(job.waitForCompletion(true) ? 0 : 1); } } 

Friend.txt

1 2 2 3 3 4 2 5

The input file contains a list of friendship between users. A small sample is shown below.

1 2

2 3

3 4

2 5

For example, the first row encodes User 2 is a friend of User 1.

The expected output file is shown below:

1 3

1 5

2 4

Please help me finish the Firends.java