Merge pull request #2220 from broadinstitute/tw_spark2

Upgrade to Spark 2

Author: tomwhite

build.gradle

@@ -55,7 +55,7 @@ repositories {
 }
 
 final htsjdkVersion = System.getProperty('htsjdk.version','2.6.1')
-final hadoopBamVersion = System.getProperty('hadoopBam.version','7.7.0')
+final hadoopBamVersion = System.getProperty('hadoopBam.version','7.7.1')
 
 configurations.all {
     resolutionStrategy {
@@ -99,7 +99,7 @@ dependencies {
     // Using the shaded version to avoid conflicts between its protobuf dependency
     // and that of Hadoop/Spark (either the one we reference explicitly, or the one
     // provided by dataproc).
-    compile 'com.google.cloud:gcloud-java-nio:0.2.8:shaded'
+    compile 'com.google.cloud:google-cloud-nio:0.5.1:shaded'
     compile 'com.google.cloud.genomics:google-genomics-dataflow:v1beta2-0.15'
     compile 'com.google.cloud.genomics:gatk-tools-java:1.1'
     compile 'org.apache.logging.log4j:log4j-api:2.3'
@@ -117,15 +117,15 @@ dependencies {
     compile 'com.google.cloud.genomics:google-genomics-utils:v1beta2-0.30'
 
    compile 'org.ojalgo:ojalgo:39.0'
-    compile ('org.apache.spark:spark-mllib_2.10:1.6.1') {
+    compile ('org.apache.spark:spark-mllib_2.10:2.0.0') {
         // JUL is used by Google Dataflow as the backend logger, so exclude jul-to-slf4j to avoid a loop
         exclude module: 'jul-to-slf4j'
         exclude module: 'javax.servlet'
         exclude module: 'servlet-api'
     }
 
     compile 'org.bdgenomics.bdg-formats:bdg-formats:0.5.0'
-    compile('org.bdgenomics.adam:adam-core_2.10:0.18.0') {
+    compile('org.bdgenomics.adam:adam-core_2.10:0.20.0') {
         exclude group: 'org.slf4j'
         exclude group: 'org.apache.hadoop'
         exclude group: 'org.scala-lang'
@@ -142,7 +142,7 @@ dependencies {
         exclude module: 'htsjdk'
     }
     compile('org.apache.hadoop:hadoop-client:2.7.2') // should be a 'provided' dependency
-    compile('com.github.jsr203hadoop:jsr203hadoop:1.0.2')
+    compile('com.github.jsr203hadoop:jsr203hadoop:1.0.3')
 
     compile('de.javakaffee:kryo-serializers:0.37') {
         exclude module: 'kryo' // use Spark's version

gatk-launch

@@ -86,6 +86,7 @@ def main(args):
             print("     LOCAL:   run using the in-memory spark runner")
             print("     SPARK:   run using spark-submit on an existing cluster ")
             print("              --sparkMaster must be specified")
+            print("              --sparkSubmitCommand may be specified to control the Spark submit command")
             print("              arguments to spark-submit may optionally be specified after -- ")
             print("     GCS:     run using Google cloud dataproc")
             print("              commands after the -- will be passed to dataproc")
@@ -109,6 +110,12 @@ def main(args):
             del args[i] #remove spark target
             del args[i] #and its parameter
 
+        sparkSubmitCommand = getValueForArgument(args, "--sparkSubmitCommand")
+        if sparkSubmitCommand is not None:
+            i = args.index("--sparkSubmitCommand")
+            del args[i] #remove sparkSubmitCommand target
+            del args[i] #and its parameter
+
         (gatkArgs, sparkArgs) = getSplitArgs(args)
 
         sparkMaster = getValueForArgument(sparkArgs, "--sparkMaster")
@@ -118,19 +125,22 @@ def main(args):
             del sparkArgs[i] #and its parameter
             gatkArgs += ["--sparkMaster", sparkMaster]
 
-        runGATK(sparkRunner, dryRun, gatkArgs, sparkArgs)
+        runGATK(sparkRunner, sparkSubmitCommand, dryRun, gatkArgs, sparkArgs)
     except GATKLaunchException as e:
         sys.stderr.write(str(e)+"\n")
         sys.exit(3)
     except CalledProcessError as e:
         sys.exit(e.returncode)
 
-def getSparkSubmit():
-    sparkhome = os.environ.get("SPARK_HOME")
-    if sparkhome is not None:
-        return sparkhome +"/bin/spark-submit"
+def getSparkSubmitCommand(sparkSubmitCommand):
+    if sparkSubmitCommand is None:
+        sparkhome = os.environ.get("SPARK_HOME")
+        if sparkhome is not None:
+            return sparkhome +"/bin/spark-submit"
+        else:
+            return "spark-submit"
     else:
-        return "spark-submit"
+        return sparkSubmitCommand
 
 def getLocalGatkRunCommand():
     localJarFromEnv = getJarFromEnv(GATK_LOCAL_JAR_ENV_VARIABLE)
@@ -251,12 +261,13 @@ def cacheJarOnGCS(jar, dryRun):
             return jar
 
 
-def runGATK(sparkRunner, dryrun, gatkArgs, sparkArgs):
+def runGATK(sparkRunner, suppliedSparkSubmitCommand, dryrun, gatkArgs, sparkArgs):
     if sparkRunner is None or sparkRunner == "LOCAL":
         cmd = getLocalGatkRunCommand() + gatkArgs + sparkArgs
         runCommand(cmd, dryrun)
     elif sparkRunner == "SPARK":
-        cmd = [ getSparkSubmit(),
+        sparkSubmitCmd = getSparkSubmitCommand(suppliedSparkSubmitCommand)
+        cmd = [ sparkSubmitCmd,
           "--master", getSparkMasterSpecified(gatkArgs)] \
               + DEFAULT_SPARK_ARGS \
               + sparkArgs \
@@ -265,7 +276,7 @@ def runGATK(sparkRunner, dryrun, gatkArgs, sparkArgs):
         try:
             runCommand(cmd, dryrun)
         except OSError:
-            raise GATKLaunchException("Tried to run spark-submit but failed.\nMake sure spark-submit is available in your path")
+            raise GATKLaunchException("Tried to run %s but failed.\nMake sure %s is available in your path" % (sparkSubmitCmd, sparkSubmitCmd))
     elif sparkRunner == "GCS":
         jarPath = cacheJarOnGCS(getSparkJar(), dryrun)
         dataprocargs = convertSparkSubmitToDataprocArgs(DEFAULT_SPARK_ARGS + sparkArgs)

src/main/java/org/broadinstitute/hellbender/cmdline/CommandLineProgram.java

@@ -17,7 +17,6 @@
 import org.broadinstitute.hellbender.utils.Utils;
 
 import java.io.File;
-import java.io.IOException;
 import java.io.PrintStream;
 import java.net.InetAddress;
 import java.text.DecimalFormat;
@@ -45,13 +44,7 @@
  *
  */
 public abstract class CommandLineProgram {
-    protected transient Logger logger = LogManager.getLogger(this.getClass());
-
-    private void readObject(java.io.ObjectInputStream in)
-            throws IOException, ClassNotFoundException {
-        in.defaultReadObject();
-        logger = LogManager.getLogger(this.getClass()); // Logger is not serializable (even by Kryo)
-    }
+    protected final Logger logger = LogManager.getLogger(this.getClass());
 
     @Argument(common=true, optional=true)
     public List<File> TMP_DIR = new ArrayList<>();

src/main/java/org/broadinstitute/hellbender/engine/FeatureManager.java

@@ -77,10 +77,10 @@ public final class FeatureManager implements AutoCloseable {
     }
 
     /**
-     * The tool instance containing the FeatureInput argument values that will form the basis of our
+     * The simple class name of the tool instance containing the FeatureInput argument values that will form the basis of our
      * pool of FeatureDataSources
      */
-    private final CommandLineProgram toolInstance;
+    private final String toolInstanceSimpleClassName;
 
     /**
      * Mapping from FeatureInput argument to query-able FeatureDataSource for that source of Features
@@ -110,10 +110,10 @@ public FeatureManager( final CommandLineProgram toolInstance ) {
      *                              the end of query intervals in anticipation of future queries (>= 0).
      */
     public FeatureManager( final CommandLineProgram toolInstance, final int featureQueryLookahead ) {
-        this.toolInstance = toolInstance;
+        this.toolInstanceSimpleClassName = toolInstance.getClass().getSimpleName();
         featureSources = new LinkedHashMap<>();
 
-        initializeFeatureSources(featureQueryLookahead);
+        initializeFeatureSources(featureQueryLookahead, toolInstance);
     }
 
     /**
@@ -124,7 +124,7 @@ public FeatureManager( final CommandLineProgram toolInstance, final int featureQ
      *                              the end of query intervals in anticipation of future queries (>= 0).
      */
     @SuppressWarnings({"unchecked", "rawtypes"})
-    private void initializeFeatureSources( final int featureQueryLookahead ) {
+    private void initializeFeatureSources( final int featureQueryLookahead, final CommandLineProgram toolInstance ) {
 
         // Discover all arguments of type FeatureInput (or Collections thereof) in our tool's class hierarchy
         // (and associated ArgumentCollections). Arguments not specified by the user on the command line will
@@ -316,7 +316,7 @@ private <T extends Feature> FeatureDataSource<T> lookupDataSource( final Feature
                                                   "In order to be detected, FeatureInputs must be declared in the tool class " +
                                                   "itself, a superclass of the tool class, or an @ArgumentCollection declared " +
                                                   "in the tool class or a superclass. They must also be annotated as an @Argument.",
-                                                  featureDescriptor.getName(), toolInstance.getClass().getSimpleName()));
+                                                  featureDescriptor.getName(), toolInstanceSimpleClassName));
         }
 
         return dataSource;

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

@@ -111,7 +111,7 @@ private static JavaPairRDD<GATKRead, ReadContextData> addUsingOverlapsPartitioni
         return shardedReads.flatMapToPair(new PairFlatMapFunction<Shard<GATKRead>, GATKRead, ReadContextData>() {
             private static final long serialVersionUID = 1L;
             @Override
-            public Iterable<Tuple2<GATKRead, ReadContextData>> call(Shard<GATKRead> shard) throws Exception {
+            public Iterator<Tuple2<GATKRead, ReadContextData>> call(Shard<GATKRead> shard) throws Exception {
                 // get reference bases for this shard (padded)
                 SimpleInterval paddedInterval = shard.getInterval().expandWithinContig(shardPadding, sequenceDictionary);
                 ReferenceBases referenceBases = bReferenceSource.getValue().getReferenceBases(null, paddedInterval);
@@ -132,7 +132,7 @@ public Tuple2<GATKRead, ReadContextData> apply(@Nullable GATKRead r) {
                     }
                 });
                 // only include reads that start in the shard
-                return () -> Iterators.filter(transform, r -> r._1().getStart() >= shard.getStart()
+                return Iterators.filter(transform, r -> r._1().getStart() >= shard.getStart()
                         && r._1().getStart() <= shard.getEnd());
             }
         });

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

@@ -108,17 +108,8 @@ public static FlatMapFunction<ContextShard,ContextShard> subdivideAndFillReads(S
             return new FlatMapFunction<ContextShard, ContextShard>() {
                 private static final long serialVersionUID = 1L;
                 @Override
-                public Iterable<ContextShard> call(ContextShard contextShard) throws Exception {
-                    return new Iterable<ContextShard>() {
-                        @Override
-                        public Iterator<ContextShard> iterator() {
-                            try {
-                                return new SubdivideAndFillReadsIterator(bam, auth, outputShardSize, margin, optFilter, contextShard);
-                            } catch (Exception x) {
-                                throw new RuntimeException(x);
-                            }
-                        }
-                    };
+                public Iterator<ContextShard> call(ContextShard contextShard) throws Exception {
+                    return new SubdivideAndFillReadsIterator(bam, auth, outputShardSize, margin, optFilter, contextShard);
                 }
             };
     }

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

@@ -130,7 +130,7 @@ public Tuple3<AlignmentContext, ReferenceContext, FeatureContext> apply(@Nullabl
                     return new Tuple3<>(alignmentContext, new ReferenceContext(reference, alignmentInterval), new FeatureContext(fm, alignmentInterval));
                 }
             });
-            return () -> transform;
+            return transform;
         };
     }
 }