cleanup of code, bugfixes

Author: ddecap

halvade/src/be/ugent/intec/halvade/HalvadeOptions.java

@@ -31,7 +31,7 @@
 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
 import org.apache.hadoop.util.Tool;
 import be.ugent.intec.halvade.utils.Logger;
-import be.ugent.intec.halvade.utils.MyConf;
+import be.ugent.intec.halvade.utils.HalvadeConf;
 import be.ugent.intec.halvade.utils.Timer;
 import fi.tkk.ics.hadoop.bam.VCFInputFormat;
 import java.net.URI;
@@ -54,23 +54,14 @@ public int run(String[] strings) throws Exception {
             halvadeOpts = new HalvadeOptions();
             int optR = halvadeOpts.GetOptions(strings, halvadeConf);
             if (optR != 0) return optR;
-            // initialise MapReduce - copy ref to each node!
+            // initialise MapReduce - copy ref to each node??
 
 
-            // only put files or continue?
-            if(halvadeOpts.justPut)
-                return 0;
-            
-            
-            Job halvadeJob = new Job(halvadeConf, "Halvade");
-            // add to dist cache with job
-            halvadeJob.addCacheArchive(new URI(halvadeOpts.halvadeDir + "bin.tar.gz"));   
-            
+            Job halvadeJob = Job.getInstance(halvadeConf, "Halvade");
+            halvadeJob.addCacheArchive(new URI(halvadeOpts.halvadeBinaries));
 
-            halvadeJob.setJarByClass(be.ugent.intec.halvade.hadoop.mapreduce.BWAMemMapper.class);
-            // specify input and output dirs
-            // check if input is a file or directory
 
+            halvadeJob.setJarByClass(be.ugent.intec.halvade.hadoop.mapreduce.HalvadeMapper.class);
             FileSystem fs = FileSystem.get(new URI(halvadeOpts.in), halvadeConf);
             try {
                 if (fs.getFileStatus(new Path(halvadeOpts.in)).isDirectory()) {
@@ -90,32 +81,32 @@ public int run(String[] strings) throws Exception {
             }
             FileOutputFormat.setOutputPath(halvadeJob, new Path(halvadeOpts.out));
 
-            // specify a mapper       
-            if (halvadeOpts.aln) halvadeJob.setMapperClass(be.ugent.intec.halvade.hadoop.mapreduce.BWAAlnMapper.class);
-            else halvadeJob.setMapperClass(be.ugent.intec.halvade.hadoop.mapreduce.BWAMemMapper.class);
+            if(halvadeOpts.rnaPipeline) 
+                halvadeJob.setMapperClass(be.ugent.intec.halvade.hadoop.mapreduce.StarAlignMapper.class);
+            else {
+                if (halvadeOpts.aln) halvadeJob.setMapperClass(be.ugent.intec.halvade.hadoop.mapreduce.BWAAlnMapper.class);
+                else halvadeJob.setMapperClass(be.ugent.intec.halvade.hadoop.mapreduce.BWAMemMapper.class);            }
             halvadeJob.setMapOutputKeyClass(ChromosomeRegion.class);
             halvadeJob.setMapOutputValueClass(SAMRecordWritable.class);
-            halvadeJob.setInputFormatClass(TextInputFormat.class); 
-//            halvadeJob.setInputFormatClass(FastqInputFormat.class);
-            
-            // per chromosome && region
+            halvadeJob.setInputFormatClass(TextInputFormat.class);             
             halvadeJob.setPartitionerClass(ChrRgPartitioner.class);
             halvadeJob.setSortComparatorClass(ChrRgPositionComparator.class);
             halvadeJob.setGroupingComparatorClass(ChrRgRegionComparator.class);
 
-            // # reducers
             if(halvadeOpts.justAlign)
                 halvadeJob.setNumReduceTasks(0);
             else
-                halvadeJob.setNumReduceTasks(halvadeOpts.reducers);
-            // specify a reducer
-            halvadeJob.setReducerClass(be.ugent.intec.halvade.hadoop.mapreduce.GATKReducer.class);
+                halvadeJob.setNumReduceTasks(halvadeOpts.reducers);            
+            if(halvadeOpts.rnaPipeline)
+                halvadeJob.setReducerClass(be.ugent.intec.halvade.hadoop.mapreduce.RnaGATKReducer.class);
+            else
+                halvadeJob.setReducerClass(be.ugent.intec.halvade.hadoop.mapreduce.DnaGATKReducer.class);            
             halvadeJob.setOutputKeyClass(Text.class);
             halvadeJob.setOutputValueClass(VariantContextWritable.class);
-//            job.setOutputFormatClass(VCFOutputFormat.class);
 
             if(halvadeOpts.dryRun) 
                 return 0;
+            
             Timer timer = new Timer();
             timer.start();
             ret = halvadeJob.waitForCompletion(true) ? 0 : 1;
@@ -127,9 +118,9 @@ public int run(String[] strings) throws Exception {
                 Logger.DEBUG("combining output");
                 Configuration combineConf = getConf();
                 if(!halvadeOpts.out.endsWith("/")) halvadeOpts.out += "/";  
-                MyConf.setInputDir(combineConf, halvadeOpts.out);
-                MyConf.setOutDir(combineConf, halvadeOpts.out + "combinedVCF/");
-                MyConf.setReportAllVariant(combineConf, halvadeOpts.reportAll);
+                HalvadeConf.setInputDir(combineConf, halvadeOpts.out);
+                HalvadeConf.setOutDir(combineConf, halvadeOpts.out + "combinedVCF/");
+                HalvadeConf.setReportAllVariant(combineConf, halvadeOpts.reportAll);
                 Job combineJob = new Job(combineConf, "HalvadeCombineVCF");            
                 combineJob.setJarByClass(be.ugent.intec.halvade.hadoop.mapreduce.VCFCombineMapper.class);
 

halvade/src/be/ugent/intec/halvade/MapReduceRunner.java

@@ -0,0 +1,56 @@
+/*
+ * To change this license header, choose License Headers in Project Properties.
+ * To change this template file, choose Tools | Templates
+ * and open the template in the editor.
+ */
+
+package be.ugent.intec.halvade.hadoop.mapreduce;
+
+import be.ugent.intec.halvade.tools.GATKTools;
+import be.ugent.intec.halvade.tools.PreprocessingTools;
+import be.ugent.intec.halvade.tools.QualityException;
+import be.ugent.intec.halvade.utils.ChromosomeRange;
+import be.ugent.intec.halvade.utils.Logger;
+import be.ugent.intec.halvade.utils.HalvadeConf;
+import be.ugent.intec.halvade.utils.SAMRecordIterator;
+import fi.tkk.ics.hadoop.bam.SAMRecordWritable;
+import java.io.IOException;
+import java.net.URISyntaxException;
+
+/**
+ *
+ * @author ddecap
+ */
+public class DnaGATKReducer extends GATKReducer {
+
+    @Override
+    protected void processAlignments(Iterable<SAMRecordWritable> values, Context context, PreprocessingTools tools, GATKTools gatk) throws IOException, InterruptedException, URISyntaxException, QualityException {
+        long startTime = System.currentTimeMillis();
+        // temporary files
+        String region = tmpFileBase + "-region.intervals";
+        String preprocess = tmpFileBase + ".bam";
+        String tmpFile1 = tmpFileBase + "-2.bam";
+        String tmpFile2 = tmpFileBase + "-3.bam";
+        String snps = tmpFileBase + ".vcf";    
+        boolean useElPrep = HalvadeConf.getUseIPrep(context.getConfiguration());
+        ChromosomeRange r = new ChromosomeRange();
+        SAMRecordIterator SAMit = new SAMRecordIterator(values.iterator(), header, r);
+        
+        if(useElPrep)
+            elPrepPreprocess(context, tools, SAMit, preprocess);
+        else 
+            PicardPreprocess(context, tools, SAMit, preprocess); 
+        region = makeRegionFile(context, r, tools, region);
+        if(region == null) return;        
+        
+        indelRealignment(context, region, gatk, preprocess, tmpFile1);        
+        baseQualityScoreRecalibration(context, region, r, tools, gatk, tmpFile1, tmpFile2);        
+        DnaVariantCalling(context, region, gatk, tmpFile2, snps);     
+        variantFiles.add(snps);
+           
+        removeLocalFile(region);
+        long estimatedTime = System.currentTimeMillis() - startTime;
+        Logger.DEBUG("total estimated time: " + estimatedTime / 1000);
+    }
+    
+}