-
Notifications
You must be signed in to change notification settings - Fork 586
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
19 changed files
with
398 additions
and
120 deletions.
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
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
59 changes: 59 additions & 0 deletions
59
src/main/java/org/broadinstitute/hellbender/engine/spark/BroadcastJoinReadsWithRefBases.java
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 |
---|---|---|
@@ -0,0 +1,59 @@ | ||
package org.broadinstitute.hellbender.engine.spark; | ||
|
||
import com.google.cloud.dataflow.sdk.transforms.SerializableFunction; | ||
import org.apache.spark.SparkContext; | ||
import org.apache.spark.api.java.JavaPairRDD; | ||
import org.apache.spark.api.java.JavaRDD; | ||
import org.apache.spark.api.java.JavaSparkContext; | ||
import org.apache.spark.broadcast.Broadcast; | ||
import org.broadinstitute.hellbender.engine.dataflow.datasources.ReferenceDataflowSource; | ||
import org.broadinstitute.hellbender.engine.spark.datasources.ReferenceTwoBitSource; | ||
import org.broadinstitute.hellbender.utils.SimpleInterval; | ||
import org.broadinstitute.hellbender.utils.read.GATKRead; | ||
import org.broadinstitute.hellbender.utils.reference.ReferenceBases; | ||
import scala.Tuple2; | ||
|
||
/** | ||
* Joins an RDD of GATKReads to reference data using a broadcast strategy. | ||
* | ||
* The ReferenceDataflowSource is broadcast using Spark's Broadcast variable mechanism. The reads are then mapped | ||
* over and a reference query is executed on each read. This makes sense for ReferenceDataflowSource implementations | ||
* that contain the reference data in memory (e.g., ReferenceTwoBitSource), but will likely be much slower for | ||
* implementations that have to query other resources for the reference sequences. | ||
*/ | ||
public class BroadcastJoinReadsWithRefBases { | ||
|
||
/** | ||
* Joins each read of an RDD<GATKRead> with that read's corresponding reference sequence. | ||
* | ||
* @param referenceDataflowSource The source of the reference sequence information | ||
* @param reads The reads for which to extract reference sequence information | ||
* @return The JavaPairRDD that contains each read along with the corresponding ReferenceBases object | ||
*/ | ||
public static JavaPairRDD<GATKRead, ReferenceBases> addBases(final ReferenceDataflowSource referenceDataflowSource, | ||
final JavaRDD<GATKRead> reads) { | ||
JavaSparkContext ctx = new JavaSparkContext(reads.context()); | ||
Broadcast<ReferenceDataflowSource> bReferenceSource = ctx.broadcast(referenceDataflowSource); | ||
return reads.mapToPair(read -> { | ||
SimpleInterval interval = bReferenceSource.getValue().getReferenceWindowFunction().apply(read); | ||
return new Tuple2<>(read, bReferenceSource.getValue().getReferenceBases(null, interval)); | ||
}); | ||
} | ||
|
||
/** | ||
* Joins each read of an RDD<GATKRead, T> with key's corresponding reference sequence. | ||
* | ||
* @param referenceDataflowSource The source of the reference sequence information | ||
* @param keyedByRead The read-keyed RDD for which to extract reference sequence information | ||
* @return The JavaPairRDD that contains each read along with the corresponding ReferenceBases object and the value | ||
*/ | ||
public static <T> JavaPairRDD<GATKRead, Tuple2<T, ReferenceBases>> addBases(final ReferenceDataflowSource referenceDataflowSource, | ||
final JavaPairRDD<GATKRead, T> keyedByRead) { | ||
JavaSparkContext ctx = new JavaSparkContext(keyedByRead.context()); | ||
Broadcast<ReferenceDataflowSource> bReferenceSource = ctx.broadcast(referenceDataflowSource); | ||
return keyedByRead.mapToPair(pair -> { | ||
SimpleInterval interval = bReferenceSource.getValue().getReferenceWindowFunction().apply(pair._1()); | ||
return new Tuple2<>(pair._1(), new Tuple2<>(pair._2(), bReferenceSource.getValue().getReferenceBases(null, interval))); | ||
}); | ||
} | ||
} |
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
16 changes: 16 additions & 0 deletions
16
src/main/java/org/broadinstitute/hellbender/engine/spark/JoinStrategy.java
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 |
---|---|---|
@@ -0,0 +1,16 @@ | ||
package org.broadinstitute.hellbender.engine.spark; | ||
|
||
/** | ||
* Possible join strategies when using Spark | ||
*/ | ||
public enum JoinStrategy { | ||
/** | ||
* Use a broadcast join strategy, where one side of the join is collected into memory and broadcast to all workers. | ||
*/ | ||
BROADCAST, | ||
|
||
/** | ||
* Use a shuffle join strategy, where both sides of join are shuffled across the workers. | ||
*/ | ||
SHUFFLE | ||
} |
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
24 changes: 24 additions & 0 deletions
24
...org/broadinstitute/hellbender/engine/spark/datasources/DirectFullByteArrayByteAccess.java
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 |
---|---|---|
@@ -0,0 +1,24 @@ | ||
package org.broadinstitute.hellbender.engine.spark.datasources; | ||
|
||
import org.bdgenomics.utils.io.ByteArrayByteAccess; | ||
|
||
/** | ||
* A version of org.bdgenomics.utils.io.ByteArrayByteAccess that makes no copies of byte array used for initialization. | ||
* However DirectFullByteArrayByteAccess.readFully can only return a reference to the full underlying byte array. | ||
* Therefore, the user should exercise caution that the underlying data does not get mutated. | ||
*/ | ||
class DirectFullByteArrayByteAccess extends ByteArrayByteAccess { | ||
private static final long serialVersionUID = 1L; | ||
|
||
DirectFullByteArrayByteAccess(byte[] bytes) { | ||
super(bytes); | ||
} | ||
|
||
@Override | ||
public byte[] readFully(long offset, int length) { | ||
if ((offset != 0) || (length != this.length())) { | ||
throw new IllegalArgumentException("readFully can only return a reference to the full underlying byte array"); | ||
} | ||
return this.bytes(); | ||
} | ||
} |
Oops, something went wrong.