/*
 * Copyright 2015 OpenCB
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.opencb.hpg.bigdata.tools.stats.alignment.mr;

import java.io.IOException;

import org.apache.avro.mapred.AvroKey;
import org.apache.avro.mapreduce.AvroJob;
import org.apache.avro.mapreduce.AvroKeyInputFormat;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
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.ga4gh.models.ReadAlignment;
import org.opencb.hpg.bigdata.tools.io.ReadAlignmentStatsWritable;
import org.opencb.hpg.bigdata.tools.io.ReadStatsWritable;

public class ReadAlignmentStatsMR {

	public static class ReadAlignmentStatsMapper extends Mapper<AvroKey<ReadAlignment>, NullWritable, LongWritable, ReadAlignmentStatsWritable> {

		@Override
		public void map(AvroKey<ReadAlignment> key, NullWritable value, Context context) throws IOException, InterruptedException {
			ReadAlignmentStatsWritable stats = new ReadAlignmentStatsWritable();
			stats.updateByReadAlignment(key.datum());
			context.write(new LongWritable(1), stats);
		}
	}

	public static class ReadAlignmentStatsReducer extends Reducer<LongWritable, ReadAlignmentStatsWritable, Text, NullWritable> {

		public void reduce(LongWritable key, Iterable<ReadAlignmentStatsWritable> values, Context context) throws IOException, InterruptedException {
			ReadAlignmentStatsWritable stats = new ReadAlignmentStatsWritable();
			for (ReadAlignmentStatsWritable value : values) {
				stats.update(value);
			}
			context.write(new Text(stats.toJSON()), NullWritable.get());
		}
	}

	public static int run(String input, String output) throws Exception {
		Configuration conf = new Configuration();

		Job job = Job.getInstance(conf, "ReadAlignmentStatsMR");		
		job.setJarByClass(ReadAlignmentStatsMR.class);

		// input
		AvroJob.setInputKeySchema(job, ReadAlignment.getClassSchema());
		FileInputFormat.setInputPaths(job, new Path(input));
		job.setInputFormatClass(AvroKeyInputFormat.class);
				
		// output
		FileOutputFormat.setOutputPath(job, new Path(output));
		job.setOutputKeyClass(ReadStatsWritable.class);
		job.setOutputValueClass(NullWritable.class);
		
		// mapper
		job.setMapperClass(ReadAlignmentStatsMapper.class);
		job.setMapOutputKeyClass(LongWritable.class);
		job.setMapOutputValueClass(ReadAlignmentStatsWritable.class);
		
		// reducer
		job.setReducerClass(ReadAlignmentStatsReducer.class);
		job.setNumReduceTasks(1);

		return (job.waitForCompletion(true) ? 0 : 1);
	}
}

@@ -1601,16 +1601,168 @@
 ble%3E %7B%0A%0A
+%09%09int newKey;%0A%09%09int numRecords;%0A%09%09final int MAX_NUM_AVRO_RECORDS = 1000;%0A%0A%09%09public void setup(Context context) %7B%0A%09%09%09newKey = 0;%0A%09%09%09numRecords = 0;%0A%09%09%7D%0A%0A
 %09%09@Overr
@@ -1765,16 +1765,16 @@
 verride%0A
-
 %09%09public
@@ -2008,16 +2008,666 @@
 tum());%0A
+%09%09%09context.write(new LongWritable(newKey), stats);%0A%0A%09%09%09// count records and update new key%0A%09%09%09numRecords++;%0A%09%09%09if (numRecords %3E= MAX_NUM_AVRO_RECORDS) %7B%0A%09%09%09%09newKey++;%0A%09%09%09%09numRecords = 0;%0A%09%09%09%7D%0A%09%09%7D%0A%09%7D%0A%0A%09public static class ReadAlignmentStatsCombiner extends Reducer%3CLongWritable, ReadAlignmentStatsWritable, LongWritable, ReadAlignmentStatsWritable%3E %7B%0A%0A%09%09public void reduce(LongWritable key, Iterable%3CReadAlignmentStatsWritable%3E values, Context context) throws IOException, InterruptedException %7B%0A%09%09%09ReadAlignmentStatsWritable stats = new ReadAlignmentStatsWritable();%0A%09%09%09for (ReadAlignmentStatsWritable value : values) %7B%0A%09%09%09%09stats.update(value);%0A%09%09%09%7D%0A
 %09%09%09conte
@@ -2704,28 +2704,27 @@
 stats);%0A%09%09%7D%0A
+
 %09%7D%0A
-%0A
 %09public stat
@@ -3970,26 +3970,97 @@
 ble.class);%0A
+%0A
 %09%09
+// combiner%0A%09%09job.setCombinerClass(ReadAlignmentStatsCombiner.class);%0A
 %0A%09%09// reduce

add a combiner function to compute BAM stats in MapReduce. #9