Added an argument to specify the parts directory for ReadsSparkSink #4666

jamesemery · 2018-04-17T19:26:07Z

This should allow for the .parts whiteout to live elsewhere which is needed for using LOCAL disks with spark.

codecov-io · 2018-04-17T20:21:13Z

Codecov Report

Merging #4666 into master will increase coverage by 0.547%.
The diff coverage is 66.667%.

@@              Coverage Diff               @@
##              master    #4666       +/-   ##
==============================================
+ Coverage     79.823%   80.37%   +0.547%     
- Complexity     17328    18189      +861     
==============================================
  Files           1075     1083        +8     
  Lines          62917    66551     +3634     
  Branches       10181    11208     +1027     
==============================================
+ Hits           50222    53487     +3265     
- Misses          8714     8924      +210     
- Partials        3981     4140      +159

Impacted Files	Coverage Δ	Complexity Δ
...lbender/tools/spark/pathseq/PathSeqScoreSpark.java	`57.407% <ø> (ø)`	`7 <0> (ø)`	⬇️
...nder/tools/spark/pathseq/PathSeqPipelineSpark.java	`81.25% <ø> (ø)`	`7 <0> (ø)`	⬇️
...stitute/hellbender/engine/spark/GATKSparkTool.java	`83.495% <100%> (+0.162%)`	`56 <0> (ø)`	⬇️
...k/pipelines/BwaAndMarkDuplicatesPipelineSpark.java	`92.593% <100%> (+13.645%)`	`7 <0> (+3)`	⬆️
...ellbender/tools/spark/pathseq/PathSeqBwaSpark.java	`67.391% <100%> (ø)`	`7 <0> (ø)`	⬇️
...ender/engine/spark/datasources/ReadsSparkSink.java	`73.043% <50%> (-2.411%)`	`26 <2> (+2)`
...der/tools/walkers/mutect/M2ArgumentCollection.java	`96.875% <0%> (-3.125%)`	`3% <0%> (+2%)`
...broadinstitute/hellbender/utils/test/BaseTest.java	`62.921% <0%> (-1.523%)`	`49% <0%> (+16%)`
...hellbender/utils/read/markduplicates/ReadsKey.java	`93.103% <0%> (-1.341%)`	`15% <0%> (+6%)`
...itute/hellbender/tools/walkers/bqsr/ApplyBQSR.java	`90.909% <0%> (-0.758%)`	`11% <0%> (+5%)`
... and 56 more

lbergelson · 2018-04-24T20:28:48Z

src/main/java/org/broadinstitute/hellbender/engine/spark/datasources/ReadsSparkSink.java


        final JavaRDD<SAMRecord> sortedReads = SparkUtils.sortReads(reads, header, numReducers);
-        final String outputPartsDirectory = outputFile + ".parts/";
+        final String outputPartsDirectory = outputPartsDir==null?outputFile + ".parts/":outputPartsDir;


spaces please

lbergelson · 2018-04-24T20:34:27Z

...test/java/org/broadinstitute/hellbender/engine/spark/datasources/ReadsSparkSinkUnitTest.java

+        BufferedWriter testfile = new BufferedWriter(new FileWriter(defaultDir.getAbsolutePath() + "/test.txt"));
+        testfile.write("test");
+        testfile.close();
+        Runtime.getRuntime().exec("chmod a-w -R " + defaultDir.getAbsolutePath()+"/");


why do you have to shell out? can't you just use defaultDir.setWriteable(false)?

IntelliJ makes a big fuss about how defaultDir.setWriteable(false) will not be respected in the test suite, and i found that it wasn't resulting in failure when we tried to write to it.

lbergelson · 2018-04-24T20:36:46Z

...test/java/org/broadinstitute/hellbender/engine/spark/datasources/ReadsSparkSinkUnitTest.java

+
+        // Make a directory with unusable permissions in place of where the default file will live
+        final File defaultDir = new File(outputFile.getAbsolutePath() + ".parts/");
+        defaultDir.mkdir();


you should assert that this returned true

lbergelson · 2018-04-24T20:36:58Z

...test/java/org/broadinstitute/hellbender/engine/spark/datasources/ReadsSparkSinkUnitTest.java

+        assertSingleShardedWritingWorks(inputBam, referenceFile, outputFile.getAbsolutePath(), outputDir.getAbsolutePath());
+
+        // Test that the file wasn't deleted when spark cleared its temp directory
+        Assert.assertEquals(new File(defaultDir.getAbsolutePath() + "/test.txt").exists(), true);


assertTrue is probably better here

lbergelson · 2018-04-24T20:37:44Z

...test/java/org/broadinstitute/hellbender/engine/spark/datasources/ReadsSparkSinkUnitTest.java

+        final File outputDir = createTempDir(outputFileName + ".someOtherPlace");
+
+        // Make a directory with unusable permissions in place of where the default file will live
+        final File defaultDir = new File(outputFile.getAbsolutePath() + ".parts/");


extract a constant for this value in the ReadSparkSink and refer to it, or even a method that takes a file and returns the default parts folder

lbergelson · 2018-04-24T20:55:40Z

src/main/java/org/broadinstitute/hellbender/engine/spark/GATKSparkTool.java

+            fullName = OUTPUT_SHARD_DIR_LONG_NAME,
+            optional = true)
+    protected String shardedPartsDir = null;
+
    @Argument(doc="For tools that shuffle data or write an output, sets the number of reducers. Defaults to 0, which gives one partition per 10MB of input.",


We probably want a check that makes sure you're not setting this if shardedOutput == true, because it will be confusingly ignored.

I added a check in the writer which will throw.

lbergelson · 2018-04-24T21:04:14Z

src/main/java/org/broadinstitute/hellbender/tools/spark/pathseq/PathSeqBwaSpark.java

@@ -180,7 +180,7 @@ private void writeBam(final JavaRDD<GATKRead> reads, final String inputBamPath,
        try {
            ReadsSparkSink.writeReads(ctx, outputPath, bwaArgs.referencePath, reads, header,
                    shardedOutput ? ReadsWriteFormat.SHARDED : ReadsWriteFormat.SINGLE,
-                    PSUtils.pathseqGetRecommendedNumReducers(inputBamPath, numReducers, getTargetPartitionSize()));
+                    PSUtils.pathseqGetRecommendedNumReducers(inputBamPath, numReducers, getTargetPartitionSize()), null);


I think these nulls should all be replaced with references to shardedPartsDir

lbergelson · 2018-04-24T21:04:19Z

src/main/java/org/broadinstitute/hellbender/tools/spark/pathseq/PathSeqPipelineSpark.java

@@ -289,7 +289,7 @@ protected void runTool(final JavaSparkContext ctx) {
                final int numPartitions = Math.max(1, (int) (numTotalReads / readsPerPartitionOutput));
                final JavaRDD<GATKRead> readsFinalRepartitioned = readsFinal.coalesce(numPartitions, false);
                ReadsSparkSink.writeReads(ctx, outputPath, null, readsFinalRepartitioned, header,
-                        shardedOutput ? ReadsWriteFormat.SHARDED : ReadsWriteFormat.SINGLE, numPartitions);
+                        shardedOutput ? ReadsWriteFormat.SHARDED : ReadsWriteFormat.SINGLE, numPartitions, null);


lbergelson · 2018-04-24T21:04:25Z

src/main/java/org/broadinstitute/hellbender/tools/spark/pathseq/PathSeqScoreSpark.java

@@ -221,7 +221,7 @@ protected void runTool(final JavaSparkContext ctx) {
        if (outputPath != null) {
            try {
                ReadsSparkSink.writeReads(ctx, outputPath, null, readsFinal, header,
-                        shardedOutput ? ReadsWriteFormat.SHARDED : ReadsWriteFormat.SINGLE, recommendedNumReducers);
+                        shardedOutput ? ReadsWriteFormat.SHARDED : ReadsWriteFormat.SINGLE, recommendedNumReducers, null);


lbergelson · 2018-04-24T21:04:42Z

...a/org/broadinstitute/hellbender/tools/spark/pipelines/BwaAndMarkDuplicatesPipelineSpark.java

@@ -66,7 +66,7 @@ protected void runTool(final JavaSparkContext ctx) {
                        referenceArguments.getReferencePath().toAbsolutePath().toUri().toString(),
                        markedReads, bwaEngine.getHeader(),
                        shardedOutput ? ReadsWriteFormat.SHARDED : ReadsWriteFormat.SINGLE,
-                        getRecommendedNumReducers());
+                        getRecommendedNumReducers(), null);


jamesemery · 2018-04-25T16:14:08Z

Responded to a round of comments

lbergelson · 2018-04-27T16:06:28Z

...test/java/org/broadinstitute/hellbender/engine/spark/datasources/ReadsSparkSinkUnitTest.java

+    }
+
+    @Test(dataProvider = "loadReadsBAM", groups = "spark")
+    public void outputDirectoryTest(String inputBam, String outputFileName, String referenceFile, String outputFileExtension) throws IOException {


I rewrote this:

@Test(dataProvider = "loadReadsBAM", groups = "spark") public void testSpecifyPartsDir(String inputBam, String outputFileName, String referenceFile, String outputFileExtension) throws IOException { final File outputFile = createTempFile(outputFileName, outputFileExtension); final File nonDefaultShardsDir = createTempDir(outputFileName + ".someOtherPlace"); final java.nio.file.Path defaultPartsDir = IOUtils.getPath(ReadsSparkSink.getDefaultPartsDirectory(outputFile.getAbsolutePath())); final java.nio.file.Path subpath = defaultPartsDir.resolve("subpath"); // Make a directory with unusable permissions in place of where the default parts file will live try { final Set<PosixFilePermission> readOnly = EnumSet.of(PosixFilePermission.OWNER_READ); final FileAttribute<Set<PosixFilePermission>> readOnlyPermissions = PosixFilePermissions.asFileAttribute(readOnly); Files.createDirectory(defaultPartsDir); // An empty directory seems to be able to be deleted even with write permissions disabled, so put a file in it Files.createFile(subpath, readOnlyPermissions); Files.setPosixFilePermissions(defaultPartsDir, readOnly); //assert it fails when writing to the default path Assert.assertThrows(() -> assertSingleShardedWritingWorks(inputBam, referenceFile, outputFile.getAbsolutePath(), null)); //show this succeeds when specifying a different path for the parts directory assertSingleShardedWritingWorks(inputBam, referenceFile, outputFile.getAbsolutePath(), nonDefaultShardsDir.getAbsolutePath()); // Test that the file wasn't deleted when spark cleared its temp directory Assert.assertTrue(Files.exists(defaultPartsDir)); } finally { try { final EnumSet<PosixFilePermission> readWriteExecute = EnumSet.of(PosixFilePermission.OWNER_READ, PosixFilePermission.OWNER_WRITE, PosixFilePermission.OWNER_EXECUTE); Files.setPosixFilePermissions(defaultPartsDir, readWriteExecute); Files.setPosixFilePermissions(subpath, readWriteExecute); Files.deleteIfExists(subpath); Files.deleteIfExists(defaultPartsDir); } catch (IOException e) { System.out.print("Failed to delete test file"); e.printStackTrace(); } } }

@jamesemery I changed it to be more assertive and also to not shell out.

lbergelson · 2018-05-07T15:11:40Z

@jamesemery Apparently my test doesn't work on travis. Gaaah.

jamesemery · 2018-05-09T17:36:24Z

@lbergelson Returned to shelling out.

lbergelson · 2018-05-09T17:38:11Z

so sad

Added an argument to specify the parts directory for ReadsSparkSink

45a1acc

droazen requested a review from lbergelson April 17, 2018 19:42

droazen assigned lbergelson Apr 17, 2018

lbergelson requested changes Apr 24, 2018

View reviewed changes

lbergelson assigned jamesemery and unassigned lbergelson Apr 24, 2018

responded to a round of comments

0f1932b

jamesemery assigned lbergelson and unassigned jamesemery Apr 25, 2018

lbergelson reviewed Apr 27, 2018

View reviewed changes

lbergelson assigned jamesemery and unassigned lbergelson Apr 27, 2018

added louis' test

6f75d93

reverted to shelling out for file permissions

2b25bad

jamesemery assigned lbergelson and unassigned jamesemery May 9, 2018

lbergelson approved these changes May 9, 2018

View reviewed changes

throwing in the towel

8d11b4f

jamesemery merged commit 3bf60ef into master May 9, 2018

davidbenjamin deleted the je_configureSparkOutDir branch February 24, 2021 15:06

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Added an argument to specify the parts directory for ReadsSparkSink #4666

Added an argument to specify the parts directory for ReadsSparkSink #4666

jamesemery commented Apr 17, 2018

codecov-io commented Apr 17, 2018 •

edited

Loading

lbergelson Apr 24, 2018

jamesemery Apr 25, 2018

lbergelson Apr 24, 2018

jamesemery Apr 25, 2018

lbergelson Apr 24, 2018

jamesemery Apr 25, 2018

lbergelson Apr 24, 2018

jamesemery Apr 25, 2018

lbergelson Apr 24, 2018

jamesemery Apr 25, 2018

lbergelson Apr 24, 2018

jamesemery Apr 25, 2018

lbergelson Apr 24, 2018

lbergelson Apr 24, 2018

lbergelson Apr 24, 2018

lbergelson Apr 24, 2018

jamesemery commented Apr 25, 2018

lbergelson Apr 27, 2018

lbergelson Apr 27, 2018

lbergelson commented May 7, 2018

jamesemery commented May 9, 2018

lbergelson commented May 9, 2018

Added an argument to specify the parts directory for ReadsSparkSink #4666

Added an argument to specify the parts directory for ReadsSparkSink #4666

Conversation

jamesemery commented Apr 17, 2018

codecov-io commented Apr 17, 2018 • edited Loading

Codecov Report

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jamesemery commented Apr 25, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lbergelson commented May 7, 2018

jamesemery commented May 9, 2018

lbergelson commented May 9, 2018

codecov-io commented Apr 17, 2018 •

edited

Loading