Added the ability for MarkDuplicatesSpark to accept multiple inputs #5430
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change | ||||||
|---|---|---|---|---|---|---|---|---|
| @@ -1,8 +1,9 @@ | ||||||||
| package org.broadinstitute.hellbender.engine.spark; | ||||||||
|
|
||||||||
| import com.google.common.annotations.VisibleForTesting; | ||||||||
| import htsjdk.samtools.*; | ||||||||
| import htsjdk.samtools.reference.ReferenceSequenceFileFactory; | ||||||||
| import htsjdk.samtools.util.CloseableIterator; | ||||||||
| import htsjdk.samtools.util.GZIIndex; | ||||||||
| import htsjdk.samtools.util.IOUtil; | ||||||||
| import htsjdk.variant.vcf.VCFHeaderLine; | ||||||||
|
|
@@ -25,6 +26,7 @@ | |||||||
| import org.broadinstitute.hellbender.engine.filters.WellformedReadFilter; | ||||||||
| import org.broadinstitute.hellbender.engine.spark.datasources.ReadsSparkSink; | ||||||||
| import org.broadinstitute.hellbender.engine.spark.datasources.ReadsSparkSource; | ||||||||
| import org.broadinstitute.hellbender.exceptions.GATKException; | ||||||||
| import org.broadinstitute.hellbender.exceptions.UserException; | ||||||||
| import org.broadinstitute.hellbender.tools.walkers.annotator.Annotation; | ||||||||
| import org.broadinstitute.hellbender.utils.SequenceDictionaryUtils; | ||||||||
|
|
@@ -121,7 +123,7 @@ public abstract class GATKSparkTool extends SparkCommandLineProgram { | |||||||
|
|
||||||||
| private ReadsSparkSource readsSource; | ||||||||
| private SAMFileHeader readsHeader; | ||||||||
| private LinkedHashMap<String, SAMFileHeader> readInputs; | ||||||||
| private ReferenceMultiSparkSource referenceSource; | ||||||||
| private SAMSequenceDictionary referenceDictionary; | ||||||||
| private List<SimpleInterval> userIntervals; | ||||||||
|
|
@@ -158,6 +160,20 @@ public boolean requiresReads() { | |||||||
| return false; | ||||||||
| } | ||||||||
|
|
||||||||
| /** | ||||||||
| * Does this tool support multiple inputs? Tools that do should override this method with the desired {@link ReadInputMergingPolicy}. | ||||||||
| * | ||||||||
| * @return doNotMerge by default | ||||||||
| */ | ||||||||
| public ReadInputMergingPolicy getReadInputMergingPolicy() { | ||||||||
| return ReadInputMergingPolicy.doNotMerge; | ||||||||
| } | ||||||||
|
|
||||||||
| public static enum ReadInputMergingPolicy { | ||||||||
| doNotMerge, | ||||||||
| concatMerge | ||||||||
| } | ||||||||
|
There was a problem hiding this comment. The |
||||||||
|
|
||||||||
| /** | ||||||||
| * Does this tool require intervals? Tools that do should override to return true. | ||||||||
| * | ||||||||
|
|
@@ -274,24 +290,39 @@ public JavaRDD<GATKRead> getUnfilteredReads() { | |||||||
| } | ||||||||
| traversalParameters = new TraversalParameters(getIntervals(), traverseUnmapped); | ||||||||
| } else { | ||||||||
| traversalParameters = null; | ||||||||
| } | ||||||||
|
|
||||||||
| JavaRDD<GATKRead> output = null; | ||||||||
| ReadsSparkSource source = readsSource; | ||||||||
| for (String input : readInputs.keySet()) { | ||||||||
| if (output == null) { | ||||||||
| output = getGatkReadJavaRDD(traversalParameters, source, input); | ||||||||
| } else { | ||||||||
| output = output.union(getGatkReadJavaRDD(traversalParameters, source, input)); | ||||||||
| } | ||||||||
| } | ||||||||
| return output; | ||||||||
| } | ||||||||
|
|
||||||||
| protected JavaRDD<GATKRead> getGatkReadJavaRDD(TraversalParameters traversalParameters, ReadsSparkSource source, String input) { | ||||||||
| JavaRDD<GATKRead> output; | ||||||||
| // TODO: This if statement is a temporary hack until #959 gets resolve | ||||||||
| if (input.endsWith(".adam")) { | ||||||||
| try { | ||||||||
| output = source.getADAMReads(input, traversalParameters, getHeaderForReads()); | ||||||||
| } catch (IOException e) { | ||||||||
| throw new UserException("Failed to read ADAM file " + input, e); | ||||||||
| } | ||||||||
|
|
||||||||
| } else { | ||||||||
| if (hasCramInput() && !hasReference()){ | ||||||||
| throw new UserException.MissingReference("A reference file is required when using CRAM files."); | ||||||||
| } | ||||||||
| final String refPath = hasReference() ? referenceArguments.getReferenceFileName() : null; | ||||||||
| output = source.getParallelReads(input, refPath, traversalParameters, bamPartitionSplitSize); | ||||||||
| } | ||||||||
| return output; | ||||||||
| } | ||||||||
|
|
||||||||
| /** | ||||||||
|
|
@@ -334,7 +365,8 @@ public int getRecommendedNumReducers() { | |||||||
| if (numReducers != 0) { | ||||||||
| return numReducers; | ||||||||
| } | ||||||||
| int size = readInputs.keySet().stream().mapToInt(k -> (int) BucketUtils.dirSize(k)).sum(); | ||||||||
| return 1 + (size / getTargetPartitionSize()); | ||||||||
| } | ||||||||
|
|
||||||||
| /** | ||||||||
|
|
@@ -445,8 +477,18 @@ public Collection<Annotation> makeVariantAnnotations() { | |||||||
| /** | ||||||||
| * Returns the name of the source of reads data. It can be a file name or URL. | ||||||||
| */ | ||||||||
| protected List<String> getReadSourceName(){ | ||||||||
| if (readInputs.size() > 1) { | ||||||||
| throw new GATKException("Multiple ReadsDataSources specificed but a single source requested by the tool"); | ||||||||
|
There was a problem hiding this comment. not ideal maybe but I don't know what else to do about this method... maybe change it to getReadSourceNames and return a list? There was a problem hiding this comment. Right, I wanted to avoid returning a list because there are a bunch of places in the code where tools expect there to be only one read source kicking around and I didn't want to risk breaking something or having to uproot everything... I agree its pretty gross... Theoretically it shouldn't be a problem for most tools which don't accept multiple inputs anyway There was a problem hiding this comment. I think we should change it to a list, or even get rid of it. All of the existing consumers except one use it to generate an output name for saving metrics (relatively easy to change); the other one ( There was a problem hiding this comment. @cmnbroad Alright, I'll return a list. |
||||||||
| } | ||||||||
| return new ArrayList<>(readInputs.keySet()); | ||||||||
| } | ||||||||
|
|
||||||||
| /** | ||||||||
| * Returns a map of read input to header. | ||||||||
| */ | ||||||||
| protected LinkedHashMap<String, SAMFileHeader> getReadSouceHeaderMap(){ | ||||||||
| return readInputs; | ||||||||
| } | ||||||||
|
|
||||||||
| /** | ||||||||
|
|
@@ -489,15 +531,37 @@ private void initializeReads(final JavaSparkContext sparkContext) { | |||||||
| return; | ||||||||
| } | ||||||||
|
|
||||||||
| if (getReadInputMergingPolicy() == ReadInputMergingPolicy.doNotMerge && readArguments.getReadFilesNames().size() != 1 ) { | ||||||||
| throw new UserException("Sorry, we only support a single reads input for for this spark tool."); | ||||||||
| } | ||||||||
|
|
||||||||
| readInputs = new LinkedHashMap<>(); | ||||||||
| readsSource = new ReadsSparkSource(sparkContext, readArguments.getReadValidationStringency()); | ||||||||
| for (String input : readArguments.getReadFilesNames()) { | ||||||||
| readInputs.put(input, readsSource.getHeader( | ||||||||
| input, hasReference() ? referenceArguments.getReferenceFileName() : null)); | ||||||||
| } | ||||||||
| readsHeader = createHeaderMerger().getMergedHeader(); | ||||||||
| } | ||||||||
|
|
||||||||
| /** | ||||||||
| * Create a header merger from the individual SAM/BAM headers in our readers | ||||||||
| * | ||||||||
| * @return a header merger containing all individual headers in this data source | ||||||||
| */ | ||||||||
| private SamFileHeaderMerger createHeaderMerger() { | ||||||||
| return new SamFileHeaderMerger(identifySortOrder(readInputs.values()), readInputs.values(), true); | ||||||||
| } | ||||||||
|
There was a problem hiding this comment. There should be a newline here:
Suggested change
|
||||||||
| @VisibleForTesting | ||||||||
| static SAMFileHeader.SortOrder identifySortOrder(final Collection<SAMFileHeader> headers){ | ||||||||
| final Set<SAMFileHeader.SortOrder> sortOrders = headers.stream().map(SAMFileHeader::getSortOrder).collect(Collectors.toSet()); | ||||||||
|
|
||||||||
| final SAMFileHeader.SortOrder order; | ||||||||
| if (sortOrders.size() == 1) { | ||||||||
| order = sortOrders.iterator().next(); | ||||||||
| } else { | ||||||||
| order = SAMFileHeader.SortOrder.unsorted; | ||||||||
| } | ||||||||
| return order; | ||||||||
| } | ||||||||
|
|
||||||||
| /** | ||||||||
|
|
||||||||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Why not default to
concatMergeand support multiple inputs for all Spark tools?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I debated this with @lbergelson. It seems like an undesirable behavior for tools that tailor their behavior to the header sort order to union the RDDs of multiple inputs potentially invalidating any assumptions about input ordering.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I'm in favor of tools opting in.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
OK