package org.slc.sli.aggregation;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import com.mongodb.BasicDBObject;
import com.mongodb.DB;
import com.mongodb.DBCollection;
import com.mongodb.DBObject;
import com.mongodb.Mongo;
import com.mongodb.MongoURI;
import com.mongodb.hadoop.MongoInputFormat;
import com.mongodb.hadoop.MongoOutputFormat;
import com.mongodb.hadoop.io.BSONWritable;
import com.mongodb.hadoop.util.MongoConfigUtil;


/**
 * Aggregate the highest ever score
 */
public class HighestEver extends Configured implements Tool {
    
    public static void main(String[] args) throws Exception {
        int res = ToolRunner.run(new Configuration(), new HighestEver(), args);
        System.exit(res);
    }

    @Override
    public int run(String[] args) throws Exception {
        Configuration conf = getConf();
        String[] remainingArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
        
        if (remainingArgs.length < 1) {
            System.err.println("Usage: HighestEver <assmtIDCode>");
            ToolRunner.printGenericCommandUsage(System.err);
            return 1;
        }
        
        String assmtIDCode = remainingArgs[0];
        Mongo m = new Mongo("localhost");
        DB db = m.getDB("sli");
        DBCollection assessments = db.getCollection("assessment");
        DBObject assmt = assessments.findOne(new BasicDBObject("body.assessmentIdentificationCode.ID", assmtIDCode));
        String assmtId = (String) assmt.get("_id");
        MongoURI input = new MongoURI("mongodb://localhost/sli.studentAssessmentAssociation");
        MongoConfigUtil.setInputURI(conf, input);
        MongoConfigUtil.setQuery(conf, new BasicDBObject("body.assessmentId", assmtId));
        MongoConfigUtil.setOutputURI(conf, "mongodb://localhost/sli.aggregatesAssessments");
        conf.set(ScoreMapper.SCORE_TYPE, "Scale score");
        
        Job job = new Job(conf, "HighestEver");
        job.setJarByClass(getClass());
        
        job.setMapperClass(ScoreMapper.class);
        job.setCombinerClass(Highest.class);
        job.setReducerClass(Highest.class);
        
        job.setInputFormatClass(MongoInputFormat.class);
        job.setOutputKeyClass(BSONWritable.class);
        job.setOutputValueClass(DoubleWritable.class);
        job.setOutputFormatClass(MongoOutputFormat.class);
        
        boolean success = job.waitForCompletion(true);
        
        return success ? 0 : 1;
    }
}

@@ -27,16 +27,55 @@
 ation;%0A%0A
+import java.net.UnknownHostException;%0A%0A
 import o
@@ -107,24 +107,24 @@
 figuration;%0A
-
 import org.a
@@ -1485,308 +1485,52 @@
     
-Mongo m = new Mongo(%22localhost%22);%0A        DB db = m.getDB(%22sli%22);%0A        DBCollection assessments = db.getCollection(%22assessment%22);%0A        DBObject
+String
  assmt
+Id
  = 
-assessments.findOne(new BasicDBObject(%22body.assessmentIdentificationCode.ID%22, assmtIDCode));%0A        String assmtId = (String) assmt.get(%22_id%22
+getAssessmentId(assmtIDCode
 );%0A 
@@ -2449,32 +2449,32 @@
 true);%0A        %0A
-
         return s
@@ -2494,11 +2494,440 @@
 1;%0A    %7D
+%0A%0A    protected String getAssessmentId(String assmtIDCode) throws UnknownHostException %7B%0A        Mongo m = new Mongo(%22localhost%22);%0A        DB db = m.getDB(%22sli%22);%0A        DBCollection assessments = db.getCollection(%22assessment%22);%0A        DBObject assmt = assessments.findOne(new BasicDBObject(%22body.assessmentIdentificationCode.ID%22, assmtIDCode));%0A        String assmtId = (String) assmt.get(%22_id%22);%0A        return assmtId;%0A    %7D
 %0A%7D%0A

encapsulate the getAssessmentId method