Merged hops code

fixes Merged hops code fixes
logicalclocks · Jan 8, 2019 · 939ae8c · 939ae8c
1 parent 075447b
commit 939ae8c
Show file tree

Hide file tree

Showing 20 changed files with 344 additions and 41 deletions.
diff --git a/common/network-yarn/pom.xml b/common/network-yarn/pom.xml
@@ -64,7 +64,7 @@
 
     <!-- Provided dependencies -->
     <dependency>
-      <groupId>org.apache.hadoop</groupId>
+      <groupId>io.hops</groupId>
       <artifactId>hadoop-client</artifactId>
     </dependency>
     <dependency>

diff --git a/core/pom.xml b/core/pom.xml
@@ -59,7 +59,7 @@
       <artifactId>xbean-asm6-shaded</artifactId>
     </dependency>
     <dependency>
-      <groupId>org.apache.hadoop</groupId>
+      <groupId>io.hops</groupId>
       <artifactId>hadoop-client</artifactId>
     </dependency>
     <dependency>

diff --git a/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala b/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala
@@ -541,6 +541,8 @@ private[spark] class SparkSubmit extends Logging {
       OptionAssigner(args.keytab, YARN, ALL_DEPLOY_MODES, confKey = "spark.yarn.keytab"),
 
       // Other options
+      OptionAssigner(args.executorGPUs, STANDALONE | YARN, ALL_DEPLOY_MODES,
+        confKey = "spark.executor.gpus"),
       OptionAssigner(args.executorCores, STANDALONE | YARN | KUBERNETES, ALL_DEPLOY_MODES,
         confKey = "spark.executor.cores"),
       OptionAssigner(args.executorMemory, STANDALONE | MESOS | YARN | KUBERNETES, ALL_DEPLOY_MODES,

diff --git a/core/src/main/scala/org/apache/spark/deploy/SparkSubmitArguments.scala b/core/src/main/scala/org/apache/spark/deploy/SparkSubmitArguments.scala
@@ -46,6 +46,7 @@ private[deploy] class SparkSubmitArguments(args: Seq[String], env: Map[String, S
   var deployMode: String = null
   var executorMemory: String = null
   var executorCores: String = null
+  var executorGPUs: String = null
   var totalExecutorCores: String = null
   var propertiesFile: String = null
   var driverMemory: String = null
@@ -70,6 +71,8 @@ private[deploy] class SparkSubmitArguments(args: Seq[String], env: Map[String, S
   var isPython: Boolean = false
   var pyFiles: String = null
   var isR: Boolean = false
+  var isTensorFlow: String = null
+  var tensorflowNumPs: String = null
   var action: SparkSubmitAction = null
   val sparkProperties: HashMap[String, String] = new HashMap[String, String]()
   var proxyUser: String = null
@@ -181,6 +184,10 @@ private[deploy] class SparkSubmitArguments(args: Seq[String], env: Map[String, S
     totalExecutorCores = Option(totalExecutorCores)
       .orElse(sparkProperties.get("spark.cores.max"))
       .orNull
+    executorGPUs = Option(executorGPUs)
+      .orElse(sparkProperties.get("spark.executor.gpus"))
+      .orElse(env.get("SPARK_EXECUTOR_GPUS"))
+      .orNull
     name = Option(name).orElse(sparkProperties.get("spark.app.name")).orNull
     jars = Option(jars).orElse(sparkProperties.get("spark.jars")).orNull
     files = Option(files).orElse(sparkProperties.get("spark.files")).orNull
@@ -198,6 +205,14 @@ private[deploy] class SparkSubmitArguments(args: Seq[String], env: Map[String, S
       .orNull
     numExecutors = Option(numExecutors)
       .getOrElse(sparkProperties.get("spark.executor.instances").orNull)
+    isTensorFlow = Option(isTensorFlow)
+      .orElse(sparkProperties.get("spark.tensorflow.application"))
+      .orElse(env.get("SPARK_TENSORFLOW_APPLICATION"))
+      .getOrElse("false")
+    tensorflowNumPs = Option(tensorflowNumPs)
+      .orElse(sparkProperties.get("spark.tensorflow.num.ps"))
+      .orElse(env.get("SPARK_TENSORFLOW_NUM_PS"))
+      .getOrElse("0")
     queue = Option(queue).orElse(sparkProperties.get("spark.yarn.queue")).orNull
     keytab = Option(keytab).orElse(sparkProperties.get("spark.yarn.keytab")).orNull
     principal = Option(principal).orElse(sparkProperties.get("spark.yarn.principal")).orNull
@@ -326,6 +341,7 @@ private[deploy] class SparkSubmitArguments(args: Seq[String], env: Map[String, S
     |  deployMode              $deployMode
     |  executorMemory          $executorMemory
     |  executorCores           $executorCores
+    |  executorGPUs            $executorGPUs
     |  totalExecutorCores      $totalExecutorCores
     |  propertiesFile          $propertiesFile
     |  driverMemory            $driverMemory
@@ -335,6 +351,8 @@ private[deploy] class SparkSubmitArguments(args: Seq[String], env: Map[String, S
     |  driverExtraJavaOptions  $driverExtraJavaOptions
     |  supervise               $supervise
     |  queue                   $queue
+    |  isTensorFlow            $isTensorFlow
+    |  tensorflowNumPs         $tensorflowNumPs
     |  numExecutors            $numExecutors
     |  files                   $files
     |  pyFiles                 $pyFiles
@@ -379,6 +397,9 @@ private[deploy] class SparkSubmitArguments(args: Seq[String], env: Map[String, S
       case TOTAL_EXECUTOR_CORES =>
         totalExecutorCores = value
 
+      case EXECUTOR_GPUS =>
+        executorGPUs = value
+
       case EXECUTOR_CORES =>
         executorCores = value
 
@@ -423,6 +444,12 @@ private[deploy] class SparkSubmitArguments(args: Seq[String], env: Map[String, S
       case QUEUE =>
         queue = value
 
+      case IS_TENSORFLOW =>
+        isTensorFlow = value
+
+      case NUM_TENSORFLOW_PS =>
+        tensorflowNumPs = value
+
       case FILES =>
         files = Utils.resolveURIs(value)
 

diff --git a/core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala b/core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala
@@ -235,6 +235,7 @@ private[spark] object CoarseGrainedExecutorBackend extends Logging {
     var executorId: String = null
     var hostname: String = null
     var cores: Int = 0
+    var gpus: Int = 0
     var appId: String = null
     var workerUrl: Option[String] = None
     val userClassPath = new mutable.ListBuffer[URL]()
@@ -254,6 +255,9 @@ private[spark] object CoarseGrainedExecutorBackend extends Logging {
         case ("--cores") :: value :: tail =>
           cores = value.toInt
           argv = tail
+        case ("--gpus") :: value :: tail =>
+          gpus = value.toInt
+          argv = tail
         case ("--app-id") :: value :: tail =>
           appId = value
           argv = tail

diff --git a/external/kafka-0-10-assembly/pom.xml b/external/kafka-0-10-assembly/pom.xml
@@ -70,7 +70,7 @@
       <scope>provided</scope>
     </dependency>
     <dependency>
-      <groupId>org.apache.hadoop</groupId>
+      <groupId>io.hops</groupId>
       <artifactId>hadoop-client</artifactId>
       <scope>provided</scope>
     </dependency>

diff --git a/external/kinesis-asl-assembly/pom.xml b/external/kinesis-asl-assembly/pom.xml
@@ -90,7 +90,7 @@
       <scope>provided</scope>
     </dependency>
     <dependency>
-      <groupId>org.apache.hadoop</groupId>
+      <groupId>io.hops</groupId>
       <artifactId>hadoop-client</artifactId>
       <scope>provided</scope>
     </dependency>

diff --git a/hadoop-cloud/pom.xml b/hadoop-cloud/pom.xml
@@ -59,7 +59,7 @@
       <scope>test</scope>
     </dependency>
     <dependency>
-      <groupId>org.apache.hadoop</groupId>
+      <groupId>io.hops</groupId>
       <artifactId>hadoop-client</artifactId>
       <version>${hadoop.version}</version>
       <scope>provided</scope>
@@ -69,13 +69,13 @@
       intra-jackson-module version problems.
       -->
     <dependency>
-      <groupId>org.apache.hadoop</groupId>
+      <groupId>io.hops</groupId>
       <artifactId>hadoop-aws</artifactId>
       <version>${hadoop.version}</version>
       <scope>${hadoop.deps.scope}</scope>
       <exclusions>
         <exclusion>
-          <groupId>org.apache.hadoop</groupId>
+          <groupId>io.hops</groupId>
           <artifactId>hadoop-common</artifactId>
         </exclusion>
         <exclusion>
@@ -105,13 +105,13 @@
       </exclusions>
     </dependency>
     <dependency>
-      <groupId>org.apache.hadoop</groupId>
+      <groupId>io.hops</groupId>
       <artifactId>hadoop-openstack</artifactId>
       <version>${hadoop.version}</version>
       <scope>${hadoop.deps.scope}</scope>
       <exclusions>
         <exclusion>
-          <groupId>org.apache.hadoop</groupId>
+          <groupId>io.hops</groupId>
           <artifactId>hadoop-common</artifactId>
         </exclusion>
         <exclusion>
@@ -179,13 +179,13 @@
         Hadoop WASB client only arrived in Hadoop 2.7
         -->
         <dependency>
-          <groupId>org.apache.hadoop</groupId>
+          <groupId>io.hops</groupId>
           <artifactId>hadoop-azure</artifactId>
           <version>${hadoop.version}</version>
           <scope>${hadoop.deps.scope}</scope>
           <exclusions>
             <exclusion>
-              <groupId>org.apache.hadoop</groupId>
+              <groupId>io.hops</groupId>
               <artifactId>hadoop-common</artifactId>
             </exclusion>
             <exclusion>

diff --git a/launcher/pom.xml b/launcher/pom.xml
@@ -80,7 +80,7 @@
 
     <!-- Not needed by the test code, but referenced by SparkSubmit which is used by the tests. -->
     <dependency>
-      <groupId>org.apache.hadoop</groupId>
+      <groupId>io.hops</groupId>
       <artifactId>hadoop-client</artifactId>
       <scope>test</scope>
     </dependency>

diff --git a/launcher/src/main/java/org/apache/spark/launcher/SparkLauncher.java b/launcher/src/main/java/org/apache/spark/launcher/SparkLauncher.java
@@ -63,6 +63,13 @@ public class SparkLauncher extends AbstractLauncher<SparkLauncher> {
   public static final String EXECUTOR_EXTRA_LIBRARY_PATH = "spark.executor.extraLibraryPath";
   /** Configuration key for the number of executor CPU cores. */
   public static final String EXECUTOR_CORES = "spark.executor.cores";
+
+  /** Configuration key for the number of executor CPU cores. */
+  public static final String EXECUTOR_GPUS = "spark.executor.gpus";
+
+  static final String IS_TENSORFLOW = "spark.tensorflow.application";
+
+  static final String NUM_TENSORFLOW_PS = "spark.tensorflow.num.ps";
 
   static final String PYSPARK_DRIVER_PYTHON = "spark.pyspark.driver.python";
 

diff --git a/launcher/src/main/java/org/apache/spark/launcher/SparkSubmitOptionParser.java b/launcher/src/main/java/org/apache/spark/launcher/SparkSubmitOptionParser.java
@@ -71,11 +71,14 @@ class SparkSubmitOptionParser {
   // YARN-only options.
   protected final String ARCHIVES = "--archives";
   protected final String EXECUTOR_CORES = "--executor-cores";
+  protected final String EXECUTOR_GPUS = "--executor-gpus";
   protected final String KEYTAB = "--keytab";
   protected final String NUM_EXECUTORS = "--num-executors";
   protected final String PRINCIPAL = "--principal";
   protected final String QUEUE = "--queue";
-
+  protected final String IS_TENSORFLOW = "--is-tensorflow";
+  protected final String NUM_TENSORFLOW_PS = "--num-tensorflow-ps";
+
   /**
    * This is the canonical list of spark-submit options. Each entry in the array contains the
    * different aliases for the same option; the first element of each entry is the "official"
@@ -96,6 +99,7 @@ class SparkSubmitOptionParser {
     { DRIVER_JAVA_OPTIONS },
     { DRIVER_LIBRARY_PATH },
     { DRIVER_MEMORY },
+    { EXECUTOR_GPUS },
     { EXECUTOR_CORES },
     { EXECUTOR_MEMORY },
     { FILES },
@@ -113,6 +117,8 @@ class SparkSubmitOptionParser {
     { PY_FILES },
     { QUEUE },
     { REPOSITORIES },
+    { IS_TENSORFLOW },
+    { NUM_TENSORFLOW_PS },
     { STATUS },
     { TOTAL_EXECUTOR_CORES },
   };