import java.io.IOException;
// import java.util.StringTokenizer;
// import org.apache.hadoop.io.LongWritable;;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.Configuration;
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 kmer

{

    public static class  Map extends Mapper<LongWritable,Text,Text,IntWritable>

    {


        public void map(LongWritable k,Text v,Context con)throws IOException,InterruptedException

        {
            String line =v.toString();
			for(int i=0;i<(line.length()-1);i=i+2)
			{
				String  y = line.substring(i,2);
				con.write(new Text(y), new IntWritable(sum));
			}
        }

    }

	public static class Reduce extends Reducer<Text,IntWritable,Text,IntWritable>
    {
        public void reduce(Text y,Iterable<IntWritable> values,Context con)throws IOException,InterruptedException
        {
            int sum=0;
            for (IntWritable val : values)
			{
				sum += val.get();
			}
			con.write(y, sum);

        }

    }

	 public static void main(String[] args)throws IOException, InterruptedException, ClassNotFoundException
    {
        Configuration c = new Configuration();
        String[] files = new GenericOptionsParser(c,args).getRemainingArgs();
        Path p1 = new Path(files[0]);
        Path p2 =new Path(files[1]);
        Job j = new Job(c,"kmer");
        j.setJarByClass(kmer.class);
        j.setMapperClass(Map.class);
        j.setReducerClass(Reduce.class);
        j.setOutputKeyClass(Text.class);
        j.setOutputValueClass(IntWritable.class);
        FileInputFormat.addInputPath(j,p1);
        FileOutputFormat.setOutputPath(j,p2);
        System.exit(j.waitForCompletion(true) ? 0:1);

    }
}
