Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[DISQ-29] Use a separate instance of VCFCodec for each partition #30

Merged
merged 5 commits into from
Sep 18, 2018
Merged

[DISQ-29] Use a separate instance of VCFCodec for each partition #30

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
d8a8d0c
Upgrade to htsjdk 2.16.1
tomwhite Sep 10, 2018
9a4eef8
Use all available cores for testing
tomwhite Sep 10, 2018
8f6dfc1
VCFCodec is not thread-safe, so use a separate instance for each part…
tomwhite Sep 4, 2018
a62a09f
Add comments
tomwhite Sep 10, 2018
d3370a1
Make VCFCodec final
tomwhite Sep 13, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -246,7 +246,7 @@ The following tests all the [GATK 'large' files](https://github.com/broadinstitu
```
mvn verify \
-Ddisq.test.real.world.files.dir=/home/gatk/src/test/resources/large \
-Ddisq.test.real.world.files.ref=/homegatk/src/test/resources/large/human_g1k_v37.20.21.fasta \
-Ddisq.test.real.world.files.ref=/home/gatk/src/test/resources/large/human_g1k_v37.20.21.fasta \
-Ddisq.samtools.bin=/path/to/bin/samtools \
-Ddisq.bcftools.bin=/path/to/bin/bcftools
```
2 changes: 1 addition & 1 deletion pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@
<properties>
<maven.compiler.target>1.8</maven.compiler.target>
<maven.compiler.source>1.8</maven.compiler.source>
<htsjdk.version>2.16.0</htsjdk.version>
<htsjdk.version>2.16.1</htsjdk.version>
<hadoop.version>2.7.0</hadoop.version>
<spark.version>2.2.2</spark.version>
</properties>
Expand Down
3 changes: 2 additions & 1 deletion src/main/java/org/disq_bio/disq/impl/formats/bam/HeaderlessBamOutputFormat.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
import htsjdk.samtools.SAMRecord;
import htsjdk.samtools.util.BinaryCodec;
import htsjdk.samtools.util.BlockCompressedOutputStream;
import java.io.File;
import java.io.IOException;
import java.io.OutputStream;
import org.apache.hadoop.conf.Configuration;
Expand All @@ -31,7 +32,7 @@ static class BamRecordWriter extends RecordWriter<Void, SAMRecord> {

public BamRecordWriter(Configuration conf, Path file, SAMFileHeader header) throws IOException {
this.out = file.getFileSystem(conf).create(file);
BlockCompressedOutputStream compressedOut = new BlockCompressedOutputStream(out, null);
BlockCompressedOutputStream compressedOut = new BlockCompressedOutputStream(out, (File) null);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I hate htsjdk.

Anyone object to making an assertion in this library that all method parameters should be nonnull by default?

@ParametersAreNonnullByDefault
package foo;

import javax.annotation.ParametersAreNonnullByDefault;

https://github.com/google/guava/wiki/UsingAndAvoidingNullExplained#using-and-avoiding-null

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@heuermh I'm not very familiar with the various nullness annotations. What does adding that do? Is it just a statement of principle or is enforced somehow?

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

As far as I know, a statement of principle, although there may be tools that use that annotation in static code analysis.

Throughout the guava library method parameters are checked via Preconditions to fail fast on nulls. Adding a dependency on guava comes with its own set of problems though.

I'd be happy to simply say we're not going to design our API to accept nulls as method parameters.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

+1 to not allowing nulls in the API. I'd rather not have a Guava dependency though.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

+1 also for non-null (I am trying the same in htsjdk-next). Maybe it is worth to check the Bean Validation project for constraints in the library (I am planning to propose this also in htsjdk-next).

binaryCodec = new BinaryCodec(compressedOut);
bamRecordCodec = new BAMRecordCodec(header);
bamRecordCodec.setOutputStream(compressedOut);
Expand Down
5 changes: 3 additions & 2 deletions src/main/java/org/disq_bio/disq/impl/formats/bgzf/BGZFCompressionOutputStream.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
package org.disq_bio.disq.impl.formats.bgzf;

import htsjdk.samtools.util.BlockCompressedOutputStream;
import java.io.File;
import java.io.IOException;
import java.io.OutputStream;
import org.apache.hadoop.io.compress.CompressionOutputStream;
Expand All @@ -18,7 +19,7 @@ public class BGZFCompressionOutputStream extends CompressionOutputStream {

public BGZFCompressionOutputStream(OutputStream out) throws IOException {
super(out);
this.output = new BlockCompressedOutputStream(out, null);
this.output = new BlockCompressedOutputStream(out, (File) null);
}

public void write(int b) throws IOException {
Expand All @@ -35,7 +36,7 @@ public void finish() throws IOException {

public void resetState() throws IOException {
output.flush();
output = new BlockCompressedOutputStream(out, null);
output = new BlockCompressedOutputStream(out, (File) null);
}

public void close() throws IOException {
Expand Down
3 changes: 2 additions & 1 deletion src/main/java/org/disq_bio/disq/impl/formats/vcf/VcfOutputFormat.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
import htsjdk.variant.variantcontext.writer.VariantContextWriter;
import htsjdk.variant.variantcontext.writer.VariantContextWriterBuilder;
import htsjdk.variant.vcf.VCFHeader;
import java.io.File;
import java.io.IOException;
import java.io.OutputStream;
import org.apache.hadoop.conf.Configuration;
Expand Down Expand Up @@ -33,7 +34,7 @@ public VcfRecordWriter(Configuration conf, Path file, VCFHeader header, String e
boolean compressed =
extension.endsWith(BGZFCodec.DEFAULT_EXTENSION) || extension.endsWith(".gz");
if (compressed) {
out = new BlockCompressedOutputStream(out, null);
out = new BlockCompressedOutputStream(out, (File) null);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

How much I miss in htsjdk a method to set the filename/source (as a String) to bgzip streams...

}
variantContextWriter =
new VariantContextWriterBuilder().clearOptions().setOutputVCFStream(out).build();
Expand Down
4 changes: 3 additions & 1 deletion src/main/java/org/disq_bio/disq/impl/formats/vcf/VcfSink.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
import htsjdk.variant.variantcontext.writer.VariantContextWriterBuilder;
import htsjdk.variant.vcf.VCFEncoder;
import htsjdk.variant.vcf.VCFHeader;
import java.io.File;
import java.io.IOException;
import java.io.OutputStream;
import java.util.Iterator;
Expand Down Expand Up @@ -50,7 +51,8 @@ public void save(
String headerFile =
tempPartsDirectory + "/header" + (compressed ? BGZFCodec.DEFAULT_EXTENSION : "");
try (OutputStream headerOut = fileSystemWrapper.create(jsc.hadoopConfiguration(), headerFile)) {
OutputStream out = compressed ? new BlockCompressedOutputStream(headerOut, null) : headerOut;
OutputStream out =
compressed ? new BlockCompressedOutputStream(headerOut, (File) null) : headerOut;
VariantContextWriter writer =
new VariantContextWriterBuilder().clearOptions().setOutputVCFStream(out).build();
writer.writeHeader(vcfHeader);
Expand Down
14 changes: 11 additions & 3 deletions src/main/java/org/disq_bio/disq/impl/formats/vcf/VcfSource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@
import htsjdk.variant.variantcontext.VariantContext;
import htsjdk.variant.vcf.VCFCodec;
import htsjdk.variant.vcf.VCFHeader;
import htsjdk.variant.vcf.VCFHeaderVersion;
import java.io.BufferedInputStream;
import java.io.IOException;
import java.io.InputStream;
Expand Down Expand Up @@ -96,14 +97,21 @@ public <T extends Locatable> JavaRDD<VariantContext> getVariants(
}
enableBGZFCodecs(conf);

Broadcast<VCFCodec> vcfCodecBroadcast = jsc.broadcast(getVCFCodec(jsc, path));
Broadcast<List<T>> intervalsBroadcast = intervals == null ? null : jsc.broadcast(intervals);
final VCFCodec vcfCodec = getVCFCodec(jsc, path);
final VCFHeader header = vcfCodec.getHeader();
// get the version separately since htsjdk doesn't provide a way to get it from the header
final VCFHeaderVersion version = vcfCodec.getVersion();
final Broadcast<VCFHeader> headerBroadcast = jsc.broadcast(header);
final Broadcast<List<T>> intervalsBroadcast =
intervals == null ? null : jsc.broadcast(intervals);

return textFile(jsc, conf, path, intervals)
.mapPartitions(
(FlatMapFunction<Iterator<String>, VariantContext>)
lines -> {
VCFCodec codec = vcfCodecBroadcast.getValue();
// VCFCodec is not threadsafe, so create a new one for each task
final VCFCodec codec = new VCFCodec();
codec.setVCFHeader(headerBroadcast.getValue(), version);
final OverlapDetector<T> overlapDetector =
intervalsBroadcast == null
? null
Expand Down
2 changes: 1 addition & 1 deletion src/test/java/org/disq_bio/disq/BaseTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@ public abstract class BaseTest {
public static void setup() {
SparkConf sparkConf = new SparkConf();
sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");
jsc = new JavaSparkContext("local", "myapp", sparkConf);
jsc = new JavaSparkContext("local[*]", "myapp", sparkConf);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could it be possible to have both kind of tests?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It would need some more work. I think it's most useful to test with multiple cores as it's most likely to expose concurrency issues (as it did in this case).

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

But having consistency between both versions might be also beneficial, right? Maybe we can open an issue if someone wants to look at refactoring to test both...

}

@AfterClass
Expand Down