You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by erikerlandson <gi...@git.apache.org> on 2016/11/29 16:14:34 UTC
[GitHub] spark pull request #16061: [SPARK-18278] [Scheduler] Support native submissi...
GitHub user erikerlandson opened a pull request:
https://github.com/apache/spark/pull/16061
[SPARK-18278] [Scheduler] Support native submission of spark jobs to a kubernetes cluster
## What changes were proposed in this pull request?
Add support for native submission of spark jobs to a kubernetes cluster, as a first-class scheduler back-end.
## How was this patch tested?
Currently, testing has been mostly informal. Integration tests will be added before completion.
## Notes
The initial branch on this PR is intended to serve as an "MVP" base for integrating various contributions to a "k8s-native" effort being conducted by several interested community members, including:
@foxish, @erikerlandson, @iyanuobidele
https://github.com/foxish/spark/tree/k8s-support/kubernetes
@mccheah
https://github.com/foxish/spark/pull/7
Goals for integrating features include:
- build of docker image (or images) for spark
- integration tests
- publishing of third-party resources
- resource constraints
- external shuffle service via daemon-sets
- possible hooks for control of executor scale by kubernetes control plane
## Instructions
Current instructions for building at the time of this submission are here:
https://github.com/foxish/spark/tree/k8s-support/kubernetes
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/erikerlandson/spark spark-18278-k8s-native
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/16061.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #16061
----
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16061: [SPARK-18278] [Scheduler] Support native submissi...
Posted by erikerlandson <gi...@git.apache.org>.
Github user erikerlandson commented on a diff in the pull request:
https://github.com/apache/spark/pull/16061#discussion_r90336438
--- Diff: kubernetes/src/main/scala/org/apache/spark/scheduler/cluster/kubernetes/KubernetesClusterSchedulerBackend.scala ---
@@ -0,0 +1,222 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.scheduler.cluster.kubernetes
+
+import collection.JavaConverters._
+import io.fabric8.kubernetes.api.model.PodBuilder
+import io.fabric8.kubernetes.api.model.extensions.JobBuilder
+import io.fabric8.kubernetes.client.{ConfigBuilder, DefaultKubernetesClient}
+import org.apache.spark.internal.config._
+import org.apache.spark.scheduler._
+import org.apache.spark.scheduler.cluster._
+import org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages._
+import org.apache.spark.{SparkConf, SparkContext, SparkException}
+import org.apache.spark.rpc.RpcEndpointAddress
+import org.apache.spark.scheduler.TaskSchedulerImpl
+import org.apache.spark.util.Utils
+
+import scala.collection.mutable
+import scala.util.Random
+import scala.concurrent.Future
+
+private[spark] class KubernetesClusterSchedulerBackend(
+ scheduler: TaskSchedulerImpl,
+ sc: SparkContext)
+ extends CoarseGrainedSchedulerBackend(scheduler, sc.env.rpcEnv) {
+
+ val client = new DefaultKubernetesClient()
+
+ val DEFAULT_NUMBER_EXECUTORS = 2
+ val sparkExecutorName = s"spark-executor-${Random.alphanumeric take 5 mkString("")}".toLowerCase()
+
+ // TODO: do these need mutex guarding?
+ // key is executor id, value is pod name
+ var executorToPod = mutable.Map.empty[String, String] // active executors
+ var shutdownToPod = mutable.Map.empty[String, String] // pending shutdown
+ var executorID = 0
+
+ val sparkImage = conf.get("spark.kubernetes.sparkImage")
+ val clientJarUri = conf.get("spark.executor.jar")
+ val ns = conf.get(
+ "spark.kubernetes.namespace",
+ KubernetesClusterScheduler.defaultNameSpace)
+ val dynamicExecutors = Utils.isDynamicAllocationEnabled(conf)
+
+ // executor back-ends take their configuration this way
+ if (dynamicExecutors) {
+ conf.setExecutorEnv("spark.dynamicAllocation.enabled", "true")
+ conf.setExecutorEnv("spark.shuffle.service.enabled", "true")
+ }
+
+ override def start(): Unit = {
+ super.start()
+ createExecutorPods(getInitialTargetExecutorNumber(sc.getConf))
+ }
+
+ override def stop(): Unit = {
+ // Kill all executor pods indiscriminately
+ killExecutorPods(executorToPod.toVector)
+ killExecutorPods(shutdownToPod.toVector)
+ super.stop()
+ }
+
+ // Dynamic allocation interfaces
+ override def doRequestTotalExecutors(requestedTotal: Int): Future[Boolean] = {
+ logInfo(s"Received doRequestTotalExecutors: $requestedTotal")
+ val n = executorToPod.size
+ val delta = requestedTotal - n
+ if (delta > 0) {
+ logInfo(s"Adding $delta new executors")
+ createExecutorPods(delta)
+ } else if (delta < 0) {
+ val d = -delta
+ val idle = executorToPod.toVector.filter { case (id, _) => !scheduler.isExecutorBusy(id) }
+ if (idle.length > 0) {
+ logInfo(s"Shutting down ${idle.length} idle executors")
+ shutdownExecutors(idle.take(d))
+ }
+ val r = math.max(0, d - idle.length)
+ if (r > 0) {
+ logInfo(s"Shutting down $r non-idle executors")
+ shutdownExecutors(executorToPod.toVector.slice(n - r, n))
+ }
+ }
+ // TODO: are there meaningful failure modes here?
+ Future.successful(true)
+ }
+
+ override def doKillExecutors(executorIds: Seq[String]): Future[Boolean] = {
+ logInfo(s"Received doKillExecutors")
+ killExecutorPods(executorIds.map { id => (id, executorToPod(id)) })
+ Future.successful(true)
+ }
+
+ private def createExecutorPods(n: Int) {
+ for (i <- 1 to n) {
+ executorID += 1
+ executorToPod += ((executorID.toString, createExecutorPod(executorID)))
+ }
+ }
+
+ def shutdownExecutors(idPodPairs: Seq[(String, String)]) {
+ val active = getExecutorIds.toSet
+
+ // Check for any finished shutting down and kill the pods
+ val shutdown = shutdownToPod.toVector.filter { case (e, _) => !active.contains(e) }
+ killExecutorPods(shutdown)
+
+ // Now request shutdown for the new ones.
+ // Move them from executor list to list pending shutdown
+ for ((id, pod) <- idPodPairs) {
+ try {
+ // TODO: 'ask' returns a future - can it be used to check eventual success?
+ Option(driverEndpoint).foreach(_.ask[Boolean](RemoveExecutor(id, ExecutorKilled)))
+ executorToPod -= id
+ shutdownToPod += ((id, pod))
+ } catch {
+ case e: Exception => logError(s"Error shutting down executor $id", e)
+ }
+ }
+ }
+
+ private def killExecutorPods(idPodPairs: Seq[(String, String)]) {
+ for ((id, pod) <- idPodPairs) {
+ try {
+ client.pods().inNamespace(ns).withName(pod).delete()
+ executorToPod -= id
+ shutdownToPod -= id
+ } catch {
+ case e: Exception => logError(s"Error killing executor pod $pod", e)
+ }
+ }
+ }
+
+ def getInitialTargetExecutorNumber(conf: SparkConf,
+ numExecutors: Int =
+ DEFAULT_NUMBER_EXECUTORS): Int = {
+ if (dynamicExecutors) {
+ val minNumExecutors = conf.get(DYN_ALLOCATION_MIN_EXECUTORS)
+ val initialNumExecutors =
+ Utils.getDynamicAllocationInitialExecutors(conf)
+ val maxNumExecutors = conf.get(DYN_ALLOCATION_MAX_EXECUTORS)
+ require(
+ initialNumExecutors >= minNumExecutors && initialNumExecutors <= maxNumExecutors,
+ s"initial executor number $initialNumExecutors must between min executor number " +
+ s"$minNumExecutors and max executor number $maxNumExecutors")
+
+ initialNumExecutors
+ } else {
+ conf.get(EXECUTOR_INSTANCES).getOrElse(numExecutors)
+ }
+ }
+
+ def createExecutorPod(executorNum: Int): String = {
+ // create a single k8s executor pod.
+ val labelMap = Map("type" -> "spark-executor")
+ val podName = s"$sparkExecutorName-$executorNum"
+
+ val submitArgs = mutable.ArrayBuffer.empty[String]
+
+ if (conf.getBoolean("spark.dynamicAllocation.enabled", false)) {
+ submitArgs ++= Vector(
+ "dynamic-executors")
+ }
+
+ submitArgs ++= Vector("org.apache.spark.executor.CoarseGrainedExecutorBackend",
+ "--driver-url", s"$driverURL",
+ "--executor-id", s"$executorNum",
+ "--hostname", "localhost",
+ "--app-id", "1", // TODO: change app-id per application and pass from driver.
+ "--cores", "1")
+
+ var pod = new PodBuilder()
+ .withNewMetadata()
+ .withLabels(labelMap.asJava)
+ .withName(podName)
+ .endMetadata()
+ .withNewSpec()
+ .withRestartPolicy("OnFailure")
+
+ .addNewContainer().withName("spark-executor").withImage(sparkImage)
+ .withImagePullPolicy("IfNotPresent")
+ .withCommand("/opt/executor.sh")
--- End diff --
I think if we can just execute the container and pass config via something along the lines of `SPARK_MASTER_OPTS` (is there `SPARK_DRIVER_OPTS` ?), that could also implicitly handle the case of running a custom user container. It might shift documentation of conventions to which env vars to expect, but if we can use standard spark env-vars that would be idiomatic.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #16061: [SPARK-18278] [Scheduler] Support native submission of s...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/16061
**[Test build #69335 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/69335/consoleFull)** for PR 16061 at commit [`8584913`](https://github.com/apache/spark/commit/8584913d51c1f68c8f9d3cdf142947cd35cdfcc2).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #16061: [SPARK-18278] [Scheduler] Support native submission of s...
Posted by foxish <gi...@git.apache.org>.
Github user foxish commented on the issue:
https://github.com/apache/spark/pull/16061
Cross posting link to [the proposal](https://issues.apache.org/jira/secure/attachment/12840988/SPARK-18278%20-%20Spark%20on%20Kubernetes%20Design%20Proposal.pdf) here.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #16061: [SPARK-18278] [Scheduler] Support native submission of s...
Posted by mccheah <gi...@git.apache.org>.
Github user mccheah commented on the issue:
https://github.com/apache/spark/pull/16061
Work on this feature has moved to https://github.com/apache-spark-on-k8s/spark. The exact diff we are working with is this: https://github.com/apache-spark-on-k8s/spark/pull/1. Feel free to provide feedback on the fork's PR.
@erikerlandson can you close this PR?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16061: [SPARK-18278] [Scheduler] Support native submissi...
Posted by erikerlandson <gi...@git.apache.org>.
Github user erikerlandson commented on a diff in the pull request:
https://github.com/apache/spark/pull/16061#discussion_r90287638
--- Diff: kubernetes/README.md ---
@@ -0,0 +1,21 @@
+# Pre-requisites
+* maven, JDK and all other pre-requisites for building Spark.
+
+# Steps to compile
+
+* Clone the fork of spark: https://github.com/foxish/spark/ and switch to the k8s-support branch.
+* Build the project
+ * ./build/mvn -Pkubernetes -Phadoop-2.4 -Dhadoop.version=2.4.0 -DskipTests package
+* Ensure that you are pointing to a k8s cluster (kubectl config current-context), which you want to use with spark.
+* Launch a spark-submit job:
+ * `./bin/spark-submit --deploy-mode cluster --class org.apache.spark.examples.SparkPi --master k8s://default --conf spark.executor.instances=5 --conf spark.kubernetes.sparkImage=manyangled/kube-spark:dynamic http://storage.googleapis.com/foxish-spark-distro/original-spark-examples_2.11-2.1.0-SNAPSHOT.jar 10000`
--- End diff --
There will need to be some official Apache Spark repository for images, which I presume will be up to Apache Spark to create. The exact nature of the images to be produced is still being discussed
I did create a "semi-official" org up on docker hub called "k8s4spark":
https://hub.docker.com/u/k8s4spark/dashboard/
I haven't actually pushed any images to it, but we could start using it as an interim repo if people think that is useful
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16061: [SPARK-18278] [Scheduler] Support native submissi...
Posted by mccheah <gi...@git.apache.org>.
Github user mccheah commented on a diff in the pull request:
https://github.com/apache/spark/pull/16061#discussion_r90320078
--- Diff: kubernetes/src/main/scala/org/apache/spark/scheduler/cluster/kubernetes/KubernetesClusterSchedulerBackend.scala ---
@@ -0,0 +1,222 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.scheduler.cluster.kubernetes
+
+import collection.JavaConverters._
+import io.fabric8.kubernetes.api.model.PodBuilder
+import io.fabric8.kubernetes.api.model.extensions.JobBuilder
+import io.fabric8.kubernetes.client.{ConfigBuilder, DefaultKubernetesClient}
+import org.apache.spark.internal.config._
+import org.apache.spark.scheduler._
+import org.apache.spark.scheduler.cluster._
+import org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages._
+import org.apache.spark.{SparkConf, SparkContext, SparkException}
+import org.apache.spark.rpc.RpcEndpointAddress
+import org.apache.spark.scheduler.TaskSchedulerImpl
+import org.apache.spark.util.Utils
+
+import scala.collection.mutable
+import scala.util.Random
+import scala.concurrent.Future
+
+private[spark] class KubernetesClusterSchedulerBackend(
+ scheduler: TaskSchedulerImpl,
+ sc: SparkContext)
+ extends CoarseGrainedSchedulerBackend(scheduler, sc.env.rpcEnv) {
+
+ val client = new DefaultKubernetesClient()
+
+ val DEFAULT_NUMBER_EXECUTORS = 2
+ val sparkExecutorName = s"spark-executor-${Random.alphanumeric take 5 mkString("")}".toLowerCase()
+
+ // TODO: do these need mutex guarding?
+ // key is executor id, value is pod name
+ var executorToPod = mutable.Map.empty[String, String] // active executors
+ var shutdownToPod = mutable.Map.empty[String, String] // pending shutdown
+ var executorID = 0
+
+ val sparkImage = conf.get("spark.kubernetes.sparkImage")
+ val clientJarUri = conf.get("spark.executor.jar")
+ val ns = conf.get(
+ "spark.kubernetes.namespace",
+ KubernetesClusterScheduler.defaultNameSpace)
+ val dynamicExecutors = Utils.isDynamicAllocationEnabled(conf)
+
+ // executor back-ends take their configuration this way
+ if (dynamicExecutors) {
+ conf.setExecutorEnv("spark.dynamicAllocation.enabled", "true")
+ conf.setExecutorEnv("spark.shuffle.service.enabled", "true")
+ }
+
+ override def start(): Unit = {
+ super.start()
+ createExecutorPods(getInitialTargetExecutorNumber(sc.getConf))
+ }
+
+ override def stop(): Unit = {
+ // Kill all executor pods indiscriminately
+ killExecutorPods(executorToPod.toVector)
+ killExecutorPods(shutdownToPod.toVector)
+ super.stop()
+ }
+
+ // Dynamic allocation interfaces
+ override def doRequestTotalExecutors(requestedTotal: Int): Future[Boolean] = {
+ logInfo(s"Received doRequestTotalExecutors: $requestedTotal")
+ val n = executorToPod.size
+ val delta = requestedTotal - n
+ if (delta > 0) {
+ logInfo(s"Adding $delta new executors")
+ createExecutorPods(delta)
+ } else if (delta < 0) {
+ val d = -delta
+ val idle = executorToPod.toVector.filter { case (id, _) => !scheduler.isExecutorBusy(id) }
+ if (idle.length > 0) {
+ logInfo(s"Shutting down ${idle.length} idle executors")
+ shutdownExecutors(idle.take(d))
+ }
+ val r = math.max(0, d - idle.length)
+ if (r > 0) {
+ logInfo(s"Shutting down $r non-idle executors")
+ shutdownExecutors(executorToPod.toVector.slice(n - r, n))
+ }
+ }
+ // TODO: are there meaningful failure modes here?
+ Future.successful(true)
+ }
+
+ override def doKillExecutors(executorIds: Seq[String]): Future[Boolean] = {
+ logInfo(s"Received doKillExecutors")
+ killExecutorPods(executorIds.map { id => (id, executorToPod(id)) })
+ Future.successful(true)
+ }
+
+ private def createExecutorPods(n: Int) {
+ for (i <- 1 to n) {
+ executorID += 1
+ executorToPod += ((executorID.toString, createExecutorPod(executorID)))
+ }
+ }
+
+ def shutdownExecutors(idPodPairs: Seq[(String, String)]) {
+ val active = getExecutorIds.toSet
+
+ // Check for any finished shutting down and kill the pods
+ val shutdown = shutdownToPod.toVector.filter { case (e, _) => !active.contains(e) }
+ killExecutorPods(shutdown)
+
+ // Now request shutdown for the new ones.
+ // Move them from executor list to list pending shutdown
+ for ((id, pod) <- idPodPairs) {
+ try {
+ // TODO: 'ask' returns a future - can it be used to check eventual success?
+ Option(driverEndpoint).foreach(_.ask[Boolean](RemoveExecutor(id, ExecutorKilled)))
+ executorToPod -= id
+ shutdownToPod += ((id, pod))
+ } catch {
+ case e: Exception => logError(s"Error shutting down executor $id", e)
+ }
+ }
+ }
+
+ private def killExecutorPods(idPodPairs: Seq[(String, String)]) {
+ for ((id, pod) <- idPodPairs) {
+ try {
+ client.pods().inNamespace(ns).withName(pod).delete()
+ executorToPod -= id
+ shutdownToPod -= id
+ } catch {
+ case e: Exception => logError(s"Error killing executor pod $pod", e)
+ }
+ }
+ }
+
+ def getInitialTargetExecutorNumber(conf: SparkConf,
+ numExecutors: Int =
+ DEFAULT_NUMBER_EXECUTORS): Int = {
+ if (dynamicExecutors) {
+ val minNumExecutors = conf.get(DYN_ALLOCATION_MIN_EXECUTORS)
+ val initialNumExecutors =
+ Utils.getDynamicAllocationInitialExecutors(conf)
+ val maxNumExecutors = conf.get(DYN_ALLOCATION_MAX_EXECUTORS)
+ require(
+ initialNumExecutors >= minNumExecutors && initialNumExecutors <= maxNumExecutors,
+ s"initial executor number $initialNumExecutors must between min executor number " +
+ s"$minNumExecutors and max executor number $maxNumExecutors")
+
+ initialNumExecutors
+ } else {
+ conf.get(EXECUTOR_INSTANCES).getOrElse(numExecutors)
+ }
+ }
+
+ def createExecutorPod(executorNum: Int): String = {
+ // create a single k8s executor pod.
+ val labelMap = Map("type" -> "spark-executor")
+ val podName = s"$sparkExecutorName-$executorNum"
+
+ val submitArgs = mutable.ArrayBuffer.empty[String]
+
+ if (conf.getBoolean("spark.dynamicAllocation.enabled", false)) {
+ submitArgs ++= Vector(
+ "dynamic-executors")
+ }
+
+ submitArgs ++= Vector("org.apache.spark.executor.CoarseGrainedExecutorBackend",
+ "--driver-url", s"$driverURL",
+ "--executor-id", s"$executorNum",
+ "--hostname", "localhost",
+ "--app-id", "1", // TODO: change app-id per application and pass from driver.
+ "--cores", "1")
+
+ var pod = new PodBuilder()
+ .withNewMetadata()
+ .withLabels(labelMap.asJava)
+ .withName(podName)
+ .endMetadata()
+ .withNewSpec()
+ .withRestartPolicy("OnFailure")
+
+ .addNewContainer().withName("spark-executor").withImage(sparkImage)
+ .withImagePullPolicy("IfNotPresent")
+ .withCommand("/opt/executor.sh")
--- End diff --
Another approach that we could take is to put the command in the Dockerfile and supply the container with environment variables to configure the runtime behavior. A few benefits to this:
- Transparency: Instead of needing to inspect the shim scripts to discover the specific execution behavior, the behavior is well-defined in the Dockerfile.
- Immutability: The run command does not need to be assembled by the client code every time.
- Decoupling execution from logic: The Docker containers are only configured with properties, as opposed to needing to specify both logic (running the shim script) and configuration.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16061: [SPARK-18278] [Scheduler] Support native submissi...
Posted by mccheah <gi...@git.apache.org>.
Github user mccheah commented on a diff in the pull request:
https://github.com/apache/spark/pull/16061#discussion_r90344349
--- Diff: kubernetes/src/main/scala/org/apache/spark/scheduler/cluster/kubernetes/KubernetesClusterSchedulerBackend.scala ---
@@ -0,0 +1,222 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.scheduler.cluster.kubernetes
+
+import collection.JavaConverters._
+import io.fabric8.kubernetes.api.model.PodBuilder
+import io.fabric8.kubernetes.api.model.extensions.JobBuilder
+import io.fabric8.kubernetes.client.{ConfigBuilder, DefaultKubernetesClient}
+import org.apache.spark.internal.config._
+import org.apache.spark.scheduler._
+import org.apache.spark.scheduler.cluster._
+import org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages._
+import org.apache.spark.{SparkConf, SparkContext, SparkException}
+import org.apache.spark.rpc.RpcEndpointAddress
+import org.apache.spark.scheduler.TaskSchedulerImpl
+import org.apache.spark.util.Utils
+
+import scala.collection.mutable
+import scala.util.Random
+import scala.concurrent.Future
+
+private[spark] class KubernetesClusterSchedulerBackend(
+ scheduler: TaskSchedulerImpl,
+ sc: SparkContext)
+ extends CoarseGrainedSchedulerBackend(scheduler, sc.env.rpcEnv) {
+
+ val client = new DefaultKubernetesClient()
+
+ val DEFAULT_NUMBER_EXECUTORS = 2
+ val sparkExecutorName = s"spark-executor-${Random.alphanumeric take 5 mkString("")}".toLowerCase()
+
+ // TODO: do these need mutex guarding?
+ // key is executor id, value is pod name
+ var executorToPod = mutable.Map.empty[String, String] // active executors
+ var shutdownToPod = mutable.Map.empty[String, String] // pending shutdown
+ var executorID = 0
+
+ val sparkImage = conf.get("spark.kubernetes.sparkImage")
+ val clientJarUri = conf.get("spark.executor.jar")
+ val ns = conf.get(
+ "spark.kubernetes.namespace",
+ KubernetesClusterScheduler.defaultNameSpace)
+ val dynamicExecutors = Utils.isDynamicAllocationEnabled(conf)
+
+ // executor back-ends take their configuration this way
+ if (dynamicExecutors) {
+ conf.setExecutorEnv("spark.dynamicAllocation.enabled", "true")
+ conf.setExecutorEnv("spark.shuffle.service.enabled", "true")
+ }
+
+ override def start(): Unit = {
+ super.start()
+ createExecutorPods(getInitialTargetExecutorNumber(sc.getConf))
+ }
+
+ override def stop(): Unit = {
+ // Kill all executor pods indiscriminately
+ killExecutorPods(executorToPod.toVector)
+ killExecutorPods(shutdownToPod.toVector)
+ super.stop()
+ }
+
+ // Dynamic allocation interfaces
+ override def doRequestTotalExecutors(requestedTotal: Int): Future[Boolean] = {
+ logInfo(s"Received doRequestTotalExecutors: $requestedTotal")
+ val n = executorToPod.size
+ val delta = requestedTotal - n
+ if (delta > 0) {
+ logInfo(s"Adding $delta new executors")
+ createExecutorPods(delta)
+ } else if (delta < 0) {
+ val d = -delta
+ val idle = executorToPod.toVector.filter { case (id, _) => !scheduler.isExecutorBusy(id) }
+ if (idle.length > 0) {
+ logInfo(s"Shutting down ${idle.length} idle executors")
+ shutdownExecutors(idle.take(d))
+ }
+ val r = math.max(0, d - idle.length)
+ if (r > 0) {
+ logInfo(s"Shutting down $r non-idle executors")
+ shutdownExecutors(executorToPod.toVector.slice(n - r, n))
+ }
+ }
+ // TODO: are there meaningful failure modes here?
+ Future.successful(true)
+ }
+
+ override def doKillExecutors(executorIds: Seq[String]): Future[Boolean] = {
+ logInfo(s"Received doKillExecutors")
+ killExecutorPods(executorIds.map { id => (id, executorToPod(id)) })
+ Future.successful(true)
+ }
+
+ private def createExecutorPods(n: Int) {
+ for (i <- 1 to n) {
+ executorID += 1
+ executorToPod += ((executorID.toString, createExecutorPod(executorID)))
+ }
+ }
+
+ def shutdownExecutors(idPodPairs: Seq[(String, String)]) {
+ val active = getExecutorIds.toSet
+
+ // Check for any finished shutting down and kill the pods
+ val shutdown = shutdownToPod.toVector.filter { case (e, _) => !active.contains(e) }
+ killExecutorPods(shutdown)
+
+ // Now request shutdown for the new ones.
+ // Move them from executor list to list pending shutdown
+ for ((id, pod) <- idPodPairs) {
+ try {
+ // TODO: 'ask' returns a future - can it be used to check eventual success?
+ Option(driverEndpoint).foreach(_.ask[Boolean](RemoveExecutor(id, ExecutorKilled)))
+ executorToPod -= id
+ shutdownToPod += ((id, pod))
+ } catch {
+ case e: Exception => logError(s"Error shutting down executor $id", e)
+ }
+ }
+ }
+
+ private def killExecutorPods(idPodPairs: Seq[(String, String)]) {
+ for ((id, pod) <- idPodPairs) {
+ try {
+ client.pods().inNamespace(ns).withName(pod).delete()
+ executorToPod -= id
+ shutdownToPod -= id
+ } catch {
+ case e: Exception => logError(s"Error killing executor pod $pod", e)
+ }
+ }
+ }
+
+ def getInitialTargetExecutorNumber(conf: SparkConf,
+ numExecutors: Int =
+ DEFAULT_NUMBER_EXECUTORS): Int = {
+ if (dynamicExecutors) {
+ val minNumExecutors = conf.get(DYN_ALLOCATION_MIN_EXECUTORS)
+ val initialNumExecutors =
+ Utils.getDynamicAllocationInitialExecutors(conf)
+ val maxNumExecutors = conf.get(DYN_ALLOCATION_MAX_EXECUTORS)
+ require(
+ initialNumExecutors >= minNumExecutors && initialNumExecutors <= maxNumExecutors,
+ s"initial executor number $initialNumExecutors must between min executor number " +
+ s"$minNumExecutors and max executor number $maxNumExecutors")
+
+ initialNumExecutors
+ } else {
+ conf.get(EXECUTOR_INSTANCES).getOrElse(numExecutors)
+ }
+ }
+
+ def createExecutorPod(executorNum: Int): String = {
+ // create a single k8s executor pod.
+ val labelMap = Map("type" -> "spark-executor")
+ val podName = s"$sparkExecutorName-$executorNum"
+
+ val submitArgs = mutable.ArrayBuffer.empty[String]
+
+ if (conf.getBoolean("spark.dynamicAllocation.enabled", false)) {
+ submitArgs ++= Vector(
+ "dynamic-executors")
+ }
+
+ submitArgs ++= Vector("org.apache.spark.executor.CoarseGrainedExecutorBackend",
+ "--driver-url", s"$driverURL",
+ "--executor-id", s"$executorNum",
+ "--hostname", "localhost",
+ "--app-id", "1", // TODO: change app-id per application and pass from driver.
+ "--cores", "1")
+
+ var pod = new PodBuilder()
+ .withNewMetadata()
+ .withLabels(labelMap.asJava)
+ .withName(podName)
+ .endMetadata()
+ .withNewSpec()
+ .withRestartPolicy("OnFailure")
+
+ .addNewContainer().withName("spark-executor").withImage(sparkImage)
+ .withImagePullPolicy("IfNotPresent")
+ .withCommand("/opt/executor.sh")
--- End diff --
`spark-class` wasn't built to handle starting executors so it won't pick up memory flags for them. We have to execute the Java binary directly. `spark-class` I believe was built for running long-lived daemons like the history server and standalone components.
As for the driver - we can run `spark-submit` for now but when we move to supporting uploading files from the submitter's local disk we're going to need an in-between process in the pod to upload the local resources to it.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16061: [SPARK-18278] [Scheduler] Support native submissi...
Posted by mccheah <gi...@git.apache.org>.
Github user mccheah commented on a diff in the pull request:
https://github.com/apache/spark/pull/16061#discussion_r90323622
--- Diff: kubernetes/src/main/scala/org/apache/spark/scheduler/cluster/kubernetes/KubernetesClusterSchedulerBackend.scala ---
@@ -0,0 +1,222 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.scheduler.cluster.kubernetes
+
+import collection.JavaConverters._
+import io.fabric8.kubernetes.api.model.PodBuilder
+import io.fabric8.kubernetes.api.model.extensions.JobBuilder
+import io.fabric8.kubernetes.client.{ConfigBuilder, DefaultKubernetesClient}
+import org.apache.spark.internal.config._
+import org.apache.spark.scheduler._
+import org.apache.spark.scheduler.cluster._
+import org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages._
+import org.apache.spark.{SparkConf, SparkContext, SparkException}
+import org.apache.spark.rpc.RpcEndpointAddress
+import org.apache.spark.scheduler.TaskSchedulerImpl
+import org.apache.spark.util.Utils
+
+import scala.collection.mutable
+import scala.util.Random
+import scala.concurrent.Future
+
+private[spark] class KubernetesClusterSchedulerBackend(
+ scheduler: TaskSchedulerImpl,
+ sc: SparkContext)
+ extends CoarseGrainedSchedulerBackend(scheduler, sc.env.rpcEnv) {
+
+ val client = new DefaultKubernetesClient()
+
+ val DEFAULT_NUMBER_EXECUTORS = 2
+ val sparkExecutorName = s"spark-executor-${Random.alphanumeric take 5 mkString("")}".toLowerCase()
+
+ // TODO: do these need mutex guarding?
+ // key is executor id, value is pod name
+ var executorToPod = mutable.Map.empty[String, String] // active executors
+ var shutdownToPod = mutable.Map.empty[String, String] // pending shutdown
+ var executorID = 0
+
+ val sparkImage = conf.get("spark.kubernetes.sparkImage")
+ val clientJarUri = conf.get("spark.executor.jar")
--- End diff --
Is this used anywhere in this class?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16061: [SPARK-18278] [Scheduler] Support native submissi...
Posted by mccheah <gi...@git.apache.org>.
Github user mccheah commented on a diff in the pull request:
https://github.com/apache/spark/pull/16061#discussion_r90320239
--- Diff: kubernetes/src/main/scala/org/apache/spark/scheduler/cluster/kubernetes/KubernetesClusterSchedulerBackend.scala ---
@@ -0,0 +1,222 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.scheduler.cluster.kubernetes
+
+import collection.JavaConverters._
+import io.fabric8.kubernetes.api.model.PodBuilder
+import io.fabric8.kubernetes.api.model.extensions.JobBuilder
+import io.fabric8.kubernetes.client.{ConfigBuilder, DefaultKubernetesClient}
+import org.apache.spark.internal.config._
+import org.apache.spark.scheduler._
+import org.apache.spark.scheduler.cluster._
+import org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages._
+import org.apache.spark.{SparkConf, SparkContext, SparkException}
+import org.apache.spark.rpc.RpcEndpointAddress
+import org.apache.spark.scheduler.TaskSchedulerImpl
+import org.apache.spark.util.Utils
+
+import scala.collection.mutable
+import scala.util.Random
+import scala.concurrent.Future
+
+private[spark] class KubernetesClusterSchedulerBackend(
+ scheduler: TaskSchedulerImpl,
+ sc: SparkContext)
+ extends CoarseGrainedSchedulerBackend(scheduler, sc.env.rpcEnv) {
+
+ val client = new DefaultKubernetesClient()
+
+ val DEFAULT_NUMBER_EXECUTORS = 2
+ val sparkExecutorName = s"spark-executor-${Random.alphanumeric take 5 mkString("")}".toLowerCase()
+
+ // TODO: do these need mutex guarding?
+ // key is executor id, value is pod name
+ var executorToPod = mutable.Map.empty[String, String] // active executors
+ var shutdownToPod = mutable.Map.empty[String, String] // pending shutdown
+ var executorID = 0
+
+ val sparkImage = conf.get("spark.kubernetes.sparkImage")
+ val clientJarUri = conf.get("spark.executor.jar")
+ val ns = conf.get(
+ "spark.kubernetes.namespace",
+ KubernetesClusterScheduler.defaultNameSpace)
+ val dynamicExecutors = Utils.isDynamicAllocationEnabled(conf)
+
+ // executor back-ends take their configuration this way
+ if (dynamicExecutors) {
+ conf.setExecutorEnv("spark.dynamicAllocation.enabled", "true")
+ conf.setExecutorEnv("spark.shuffle.service.enabled", "true")
+ }
+
+ override def start(): Unit = {
+ super.start()
+ createExecutorPods(getInitialTargetExecutorNumber(sc.getConf))
+ }
+
+ override def stop(): Unit = {
+ // Kill all executor pods indiscriminately
+ killExecutorPods(executorToPod.toVector)
+ killExecutorPods(shutdownToPod.toVector)
+ super.stop()
+ }
+
+ // Dynamic allocation interfaces
+ override def doRequestTotalExecutors(requestedTotal: Int): Future[Boolean] = {
+ logInfo(s"Received doRequestTotalExecutors: $requestedTotal")
+ val n = executorToPod.size
+ val delta = requestedTotal - n
+ if (delta > 0) {
+ logInfo(s"Adding $delta new executors")
+ createExecutorPods(delta)
+ } else if (delta < 0) {
+ val d = -delta
+ val idle = executorToPod.toVector.filter { case (id, _) => !scheduler.isExecutorBusy(id) }
+ if (idle.length > 0) {
+ logInfo(s"Shutting down ${idle.length} idle executors")
+ shutdownExecutors(idle.take(d))
+ }
+ val r = math.max(0, d - idle.length)
+ if (r > 0) {
+ logInfo(s"Shutting down $r non-idle executors")
+ shutdownExecutors(executorToPod.toVector.slice(n - r, n))
+ }
+ }
+ // TODO: are there meaningful failure modes here?
+ Future.successful(true)
+ }
+
+ override def doKillExecutors(executorIds: Seq[String]): Future[Boolean] = {
+ logInfo(s"Received doKillExecutors")
+ killExecutorPods(executorIds.map { id => (id, executorToPod(id)) })
+ Future.successful(true)
+ }
+
+ private def createExecutorPods(n: Int) {
+ for (i <- 1 to n) {
+ executorID += 1
+ executorToPod += ((executorID.toString, createExecutorPod(executorID)))
+ }
+ }
+
+ def shutdownExecutors(idPodPairs: Seq[(String, String)]) {
+ val active = getExecutorIds.toSet
+
+ // Check for any finished shutting down and kill the pods
+ val shutdown = shutdownToPod.toVector.filter { case (e, _) => !active.contains(e) }
+ killExecutorPods(shutdown)
+
+ // Now request shutdown for the new ones.
+ // Move them from executor list to list pending shutdown
+ for ((id, pod) <- idPodPairs) {
+ try {
+ // TODO: 'ask' returns a future - can it be used to check eventual success?
+ Option(driverEndpoint).foreach(_.ask[Boolean](RemoveExecutor(id, ExecutorKilled)))
+ executorToPod -= id
+ shutdownToPod += ((id, pod))
+ } catch {
+ case e: Exception => logError(s"Error shutting down executor $id", e)
+ }
+ }
+ }
+
+ private def killExecutorPods(idPodPairs: Seq[(String, String)]) {
+ for ((id, pod) <- idPodPairs) {
+ try {
+ client.pods().inNamespace(ns).withName(pod).delete()
+ executorToPod -= id
+ shutdownToPod -= id
+ } catch {
+ case e: Exception => logError(s"Error killing executor pod $pod", e)
+ }
+ }
+ }
+
+ def getInitialTargetExecutorNumber(conf: SparkConf,
+ numExecutors: Int =
+ DEFAULT_NUMBER_EXECUTORS): Int = {
+ if (dynamicExecutors) {
+ val minNumExecutors = conf.get(DYN_ALLOCATION_MIN_EXECUTORS)
+ val initialNumExecutors =
+ Utils.getDynamicAllocationInitialExecutors(conf)
+ val maxNumExecutors = conf.get(DYN_ALLOCATION_MAX_EXECUTORS)
+ require(
+ initialNumExecutors >= minNumExecutors && initialNumExecutors <= maxNumExecutors,
+ s"initial executor number $initialNumExecutors must between min executor number " +
+ s"$minNumExecutors and max executor number $maxNumExecutors")
+
+ initialNumExecutors
+ } else {
+ conf.get(EXECUTOR_INSTANCES).getOrElse(numExecutors)
+ }
+ }
+
+ def createExecutorPod(executorNum: Int): String = {
+ // create a single k8s executor pod.
+ val labelMap = Map("type" -> "spark-executor")
+ val podName = s"$sparkExecutorName-$executorNum"
+
+ val submitArgs = mutable.ArrayBuffer.empty[String]
+
+ if (conf.getBoolean("spark.dynamicAllocation.enabled", false)) {
+ submitArgs ++= Vector(
+ "dynamic-executors")
+ }
+
+ submitArgs ++= Vector("org.apache.spark.executor.CoarseGrainedExecutorBackend",
+ "--driver-url", s"$driverURL",
+ "--executor-id", s"$executorNum",
+ "--hostname", "localhost",
--- End diff --
Should we use the HOSTNAME environment variable here?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16061: [SPARK-18278] [Scheduler] Support native submissi...
Posted by mccheah <gi...@git.apache.org>.
Github user mccheah commented on a diff in the pull request:
https://github.com/apache/spark/pull/16061#discussion_r90365964
--- Diff: kubernetes/src/main/scala/org/apache/spark/scheduler/cluster/kubernetes/KubernetesClusterSchedulerBackend.scala ---
@@ -0,0 +1,222 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.scheduler.cluster.kubernetes
+
+import collection.JavaConverters._
+import io.fabric8.kubernetes.api.model.PodBuilder
+import io.fabric8.kubernetes.api.model.extensions.JobBuilder
+import io.fabric8.kubernetes.client.{ConfigBuilder, DefaultKubernetesClient}
+import org.apache.spark.internal.config._
+import org.apache.spark.scheduler._
+import org.apache.spark.scheduler.cluster._
+import org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages._
+import org.apache.spark.{SparkConf, SparkContext, SparkException}
+import org.apache.spark.rpc.RpcEndpointAddress
+import org.apache.spark.scheduler.TaskSchedulerImpl
+import org.apache.spark.util.Utils
+
+import scala.collection.mutable
+import scala.util.Random
+import scala.concurrent.Future
+
+private[spark] class KubernetesClusterSchedulerBackend(
+ scheduler: TaskSchedulerImpl,
+ sc: SparkContext)
+ extends CoarseGrainedSchedulerBackend(scheduler, sc.env.rpcEnv) {
+
+ val client = new DefaultKubernetesClient()
+
+ val DEFAULT_NUMBER_EXECUTORS = 2
+ val sparkExecutorName = s"spark-executor-${Random.alphanumeric take 5 mkString("")}".toLowerCase()
+
+ // TODO: do these need mutex guarding?
+ // key is executor id, value is pod name
+ var executorToPod = mutable.Map.empty[String, String] // active executors
+ var shutdownToPod = mutable.Map.empty[String, String] // pending shutdown
+ var executorID = 0
+
+ val sparkImage = conf.get("spark.kubernetes.sparkImage")
+ val clientJarUri = conf.get("spark.executor.jar")
+ val ns = conf.get(
+ "spark.kubernetes.namespace",
+ KubernetesClusterScheduler.defaultNameSpace)
+ val dynamicExecutors = Utils.isDynamicAllocationEnabled(conf)
+
+ // executor back-ends take their configuration this way
+ if (dynamicExecutors) {
+ conf.setExecutorEnv("spark.dynamicAllocation.enabled", "true")
+ conf.setExecutorEnv("spark.shuffle.service.enabled", "true")
+ }
+
+ override def start(): Unit = {
+ super.start()
+ createExecutorPods(getInitialTargetExecutorNumber(sc.getConf))
+ }
+
+ override def stop(): Unit = {
+ // Kill all executor pods indiscriminately
+ killExecutorPods(executorToPod.toVector)
+ killExecutorPods(shutdownToPod.toVector)
+ super.stop()
+ }
+
+ // Dynamic allocation interfaces
+ override def doRequestTotalExecutors(requestedTotal: Int): Future[Boolean] = {
+ logInfo(s"Received doRequestTotalExecutors: $requestedTotal")
+ val n = executorToPod.size
+ val delta = requestedTotal - n
+ if (delta > 0) {
+ logInfo(s"Adding $delta new executors")
+ createExecutorPods(delta)
+ } else if (delta < 0) {
+ val d = -delta
+ val idle = executorToPod.toVector.filter { case (id, _) => !scheduler.isExecutorBusy(id) }
+ if (idle.length > 0) {
+ logInfo(s"Shutting down ${idle.length} idle executors")
+ shutdownExecutors(idle.take(d))
+ }
+ val r = math.max(0, d - idle.length)
+ if (r > 0) {
+ logInfo(s"Shutting down $r non-idle executors")
+ shutdownExecutors(executorToPod.toVector.slice(n - r, n))
+ }
+ }
+ // TODO: are there meaningful failure modes here?
+ Future.successful(true)
+ }
+
+ override def doKillExecutors(executorIds: Seq[String]): Future[Boolean] = {
+ logInfo(s"Received doKillExecutors")
+ killExecutorPods(executorIds.map { id => (id, executorToPod(id)) })
+ Future.successful(true)
+ }
+
+ private def createExecutorPods(n: Int) {
+ for (i <- 1 to n) {
+ executorID += 1
+ executorToPod += ((executorID.toString, createExecutorPod(executorID)))
+ }
+ }
+
+ def shutdownExecutors(idPodPairs: Seq[(String, String)]) {
+ val active = getExecutorIds.toSet
+
+ // Check for any finished shutting down and kill the pods
+ val shutdown = shutdownToPod.toVector.filter { case (e, _) => !active.contains(e) }
+ killExecutorPods(shutdown)
+
+ // Now request shutdown for the new ones.
+ // Move them from executor list to list pending shutdown
+ for ((id, pod) <- idPodPairs) {
+ try {
+ // TODO: 'ask' returns a future - can it be used to check eventual success?
+ Option(driverEndpoint).foreach(_.ask[Boolean](RemoveExecutor(id, ExecutorKilled)))
+ executorToPod -= id
+ shutdownToPod += ((id, pod))
+ } catch {
+ case e: Exception => logError(s"Error shutting down executor $id", e)
+ }
+ }
+ }
+
+ private def killExecutorPods(idPodPairs: Seq[(String, String)]) {
+ for ((id, pod) <- idPodPairs) {
+ try {
+ client.pods().inNamespace(ns).withName(pod).delete()
+ executorToPod -= id
+ shutdownToPod -= id
+ } catch {
+ case e: Exception => logError(s"Error killing executor pod $pod", e)
+ }
+ }
+ }
+
+ def getInitialTargetExecutorNumber(conf: SparkConf,
+ numExecutors: Int =
+ DEFAULT_NUMBER_EXECUTORS): Int = {
+ if (dynamicExecutors) {
+ val minNumExecutors = conf.get(DYN_ALLOCATION_MIN_EXECUTORS)
+ val initialNumExecutors =
+ Utils.getDynamicAllocationInitialExecutors(conf)
+ val maxNumExecutors = conf.get(DYN_ALLOCATION_MAX_EXECUTORS)
+ require(
+ initialNumExecutors >= minNumExecutors && initialNumExecutors <= maxNumExecutors,
+ s"initial executor number $initialNumExecutors must between min executor number " +
+ s"$minNumExecutors and max executor number $maxNumExecutors")
+
+ initialNumExecutors
+ } else {
+ conf.get(EXECUTOR_INSTANCES).getOrElse(numExecutors)
+ }
+ }
+
+ def createExecutorPod(executorNum: Int): String = {
+ // create a single k8s executor pod.
+ val labelMap = Map("type" -> "spark-executor")
+ val podName = s"$sparkExecutorName-$executorNum"
+
+ val submitArgs = mutable.ArrayBuffer.empty[String]
+
+ if (conf.getBoolean("spark.dynamicAllocation.enabled", false)) {
+ submitArgs ++= Vector(
+ "dynamic-executors")
+ }
+
+ submitArgs ++= Vector("org.apache.spark.executor.CoarseGrainedExecutorBackend",
+ "--driver-url", s"$driverURL",
+ "--executor-id", s"$executorNum",
+ "--hostname", "localhost",
+ "--app-id", "1", // TODO: change app-id per application and pass from driver.
+ "--cores", "1")
+
+ var pod = new PodBuilder()
+ .withNewMetadata()
+ .withLabels(labelMap.asJava)
+ .withName(podName)
+ .endMetadata()
+ .withNewSpec()
+ .withRestartPolicy("OnFailure")
+
+ .addNewContainer().withName("spark-executor").withImage(sparkImage)
+ .withImagePullPolicy("IfNotPresent")
+ .withCommand("/opt/executor.sh")
--- End diff --
Actually I think I was mistaken here - `spark-class` eventually calls to [this class](https://github.com/apache/spark/blob/master/launcher/src/main/java/org/apache/spark/launcher/SparkClassCommandBuilder.java#L65) which then forks the actual process running the passed-in class. We can probably then indeed take advantage of `SPARK_JAVA_OPTS` and `SPARK_EXECUTOR_MEMORY`, etc. Arguments like executor-id which are passed as command line arguments and not JVM options should probably still be set as environment variables on the container though and the Dockerfile handles formatting the command line arguments sent to `spark-class`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16061: [SPARK-18278] [Scheduler] Support native submissi...
Posted by tnachen <gi...@git.apache.org>.
Github user tnachen commented on a diff in the pull request:
https://github.com/apache/spark/pull/16061#discussion_r90063528
--- Diff: kubernetes/src/main/scala/org/apache/spark/scheduler/cluster/kubernetes/KubernetesClusterScheduler.scala ---
@@ -0,0 +1,236 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.scheduler.cluster.kubernetes
+
+import java.io.File
+import java.util.Date
+import java.util.concurrent.atomic.AtomicLong
+
+import io.fabric8.kubernetes.client.{ConfigBuilder, DefaultKubernetesClient, KubernetesClient}
+import io.fabric8.kubernetes.api.model.{PodBuilder, ServiceBuilder}
+import io.fabric8.kubernetes.client.dsl.LogWatch
+import org.apache.spark.deploy.Command
+import org.apache.spark.deploy.kubernetes.ClientArguments
+import org.apache.spark.{io, _}
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.config._
+
+import collection.JavaConverters._
+import org.apache.spark.util.Utils
+
+import scala.util.Random
+
+private[spark] object KubernetesClusterScheduler {
+ def defaultNameSpace = "default"
+ def defaultServiceAccountName = "default"
+}
+
+/**
+ * This is a simple extension to ClusterScheduler
+ * */
+private[spark] class KubernetesClusterScheduler(conf: SparkConf)
+ extends Logging {
+ private val DEFAULT_SUPERVISE = false
+ private val DEFAULT_MEMORY = Utils.DEFAULT_DRIVER_MEM_MB // mb
+ private val DEFAULT_CORES = 1.0
+
+ logInfo("Created KubernetesClusterScheduler instance")
+
+ var client = setupKubernetesClient()
+ val driverName = s"spark-driver-${Random.alphanumeric take 5 mkString("")}".toLowerCase()
+ val svcName = s"spark-svc-${Random.alphanumeric take 5 mkString("")}".toLowerCase()
+ val nameSpace = conf.get(
+ "spark.kubernetes.namespace",
+ KubernetesClusterScheduler.defaultNameSpace)
+ val serviceAccountName = conf.get(
+ "spark.kubernetes.serviceAccountName",
+ KubernetesClusterScheduler.defaultServiceAccountName)
+
+ // Anything that should either not be passed to driver config in the cluster, or
+ // that is going to be explicitly managed as command argument to the driver pod
+ val confBlackList = scala.collection.Set(
+ "spark.master",
+ "spark.app.name",
+ "spark.submit.deployMode",
+ "spark.executor.jar",
+ "spark.dynamicAllocation.enabled",
+ "spark.shuffle.service.enabled")
+
+ def start(args: ClientArguments): Unit = {
+ startDriver(client, args)
+ }
+
+ def stop(): Unit = {
+ client.pods().inNamespace(nameSpace).withName(driverName).delete()
+ client
+ .services()
+ .inNamespace(nameSpace)
+ .withName(svcName)
+ .delete()
+ }
+
+ def startDriver(client: KubernetesClient,
+ args: ClientArguments): Unit = {
+ logInfo("Starting spark driver on kubernetes cluster")
+ val driverDescription = buildDriverDescription(args)
+
+ // image needs to support shim scripts "/opt/driver.sh" and "/opt/executor.sh"
+ val sparkImage = conf.getOption("spark.kubernetes.sparkImage").getOrElse {
+ // TODO: this needs to default to some standard Apache Spark image
+ throw new SparkException("Spark image not set. Please configure spark.kubernetes.sparkImage")
+ }
+
+ // This is the URL of the client jar.
+ val clientJarUri = args.userJar
+
+ // This is the kubernetes master we're launching on.
+ val kubernetesHost = "k8s://" + client.getMasterUrl().getHost()
+ logInfo("Using as kubernetes-master: " + kubernetesHost.toString())
+
+ val submitArgs = scala.collection.mutable.ArrayBuffer.empty[String]
+ submitArgs ++= Vector(
+ clientJarUri,
+ s"--class=${args.userClass}",
+ s"--master=$kubernetesHost",
+ s"--executor-memory=${driverDescription.mem}",
+ s"--conf spark.executor.jar=$clientJarUri")
+
+ submitArgs ++= conf.getAll.filter { case (name, _) => !confBlackList.contains(name) }
+ .map { case (name, value) => s"--conf ${name}=${value}" }
+
+ if (conf.getBoolean("spark.dynamicAllocation.enabled", false)) {
+ submitArgs ++= Vector(
+ "--conf spark.dynamicAllocation.enabled=true",
+ "--conf spark.shuffle.service.enabled=true")
+ }
+
+ // these have to come at end of arg list
+ submitArgs ++= Vector("/opt/spark/kubernetes/client.jar",
+ args.userArgs.mkString(" "))
+
+ val labelMap = Map("type" -> "spark-driver")
+ val pod = new PodBuilder()
+ .withNewMetadata()
+ .withLabels(labelMap.asJava)
+ .withName(driverName)
+ .endMetadata()
+ .withNewSpec()
+ .withRestartPolicy("OnFailure")
+ .withServiceAccount(serviceAccountName)
+ .addNewContainer()
+ .withName("spark-driver")
+ .withImage(sparkImage)
+ .withImagePullPolicy("Always")
+ .withCommand(s"/opt/driver.sh")
+ .withArgs(submitArgs :_*)
+ .endContainer()
+ .endSpec()
+ .build()
+ client.pods().inNamespace(nameSpace).withName(driverName).create(pod)
+
+ var svc = new ServiceBuilder()
+ .withNewMetadata()
+ .withLabels(labelMap.asJava)
+ .withName(svcName)
+ .endMetadata()
+ .withNewSpec()
+ .addNewPort()
+ .withPort(4040)
+ .withNewTargetPort()
+ .withIntVal(4040)
+ .endTargetPort()
+ .endPort()
+ .withSelector(labelMap.asJava)
+ .withType("LoadBalancer")
+ .endSpec()
+ .build()
+
+ client
+ .services()
+ .inNamespace(nameSpace)
+ .withName(svcName)
+ .create(svc)
+
+// try {
--- End diff --
Remove
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16061: [SPARK-18278] [Scheduler] Support native submissi...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/16061#discussion_r90168600
--- Diff: kubernetes/README.md ---
@@ -0,0 +1,21 @@
+# Pre-requisites
+* maven, JDK and all other pre-requisites for building Spark.
+
+# Steps to compile
+
+* Clone the fork of spark: https://github.com/foxish/spark/ and switch to the k8s-support branch.
--- End diff --
I think this is not correct now?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16061: [SPARK-18278] [Scheduler] Support native submissi...
Posted by erikerlandson <gi...@git.apache.org>.
Github user erikerlandson commented on a diff in the pull request:
https://github.com/apache/spark/pull/16061#discussion_r90678500
--- Diff: kubernetes/src/main/scala/org/apache/spark/scheduler/cluster/kubernetes/KubernetesClusterSchedulerBackend.scala ---
@@ -0,0 +1,222 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.scheduler.cluster.kubernetes
+
+import collection.JavaConverters._
+import io.fabric8.kubernetes.api.model.PodBuilder
+import io.fabric8.kubernetes.api.model.extensions.JobBuilder
+import io.fabric8.kubernetes.client.{ConfigBuilder, DefaultKubernetesClient}
+import org.apache.spark.internal.config._
+import org.apache.spark.scheduler._
+import org.apache.spark.scheduler.cluster._
+import org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages._
+import org.apache.spark.{SparkConf, SparkContext, SparkException}
+import org.apache.spark.rpc.RpcEndpointAddress
+import org.apache.spark.scheduler.TaskSchedulerImpl
+import org.apache.spark.util.Utils
+
+import scala.collection.mutable
+import scala.util.Random
+import scala.concurrent.Future
+
+private[spark] class KubernetesClusterSchedulerBackend(
+ scheduler: TaskSchedulerImpl,
+ sc: SparkContext)
+ extends CoarseGrainedSchedulerBackend(scheduler, sc.env.rpcEnv) {
+
+ val client = new DefaultKubernetesClient()
+
+ val DEFAULT_NUMBER_EXECUTORS = 2
+ val sparkExecutorName = s"spark-executor-${Random.alphanumeric take 5 mkString("")}".toLowerCase()
+
+ // TODO: do these need mutex guarding?
+ // key is executor id, value is pod name
+ var executorToPod = mutable.Map.empty[String, String] // active executors
+ var shutdownToPod = mutable.Map.empty[String, String] // pending shutdown
+ var executorID = 0
+
+ val sparkImage = conf.get("spark.kubernetes.sparkImage")
+ val clientJarUri = conf.get("spark.executor.jar")
+ val ns = conf.get(
+ "spark.kubernetes.namespace",
+ KubernetesClusterScheduler.defaultNameSpace)
+ val dynamicExecutors = Utils.isDynamicAllocationEnabled(conf)
+
+ // executor back-ends take their configuration this way
+ if (dynamicExecutors) {
+ conf.setExecutorEnv("spark.dynamicAllocation.enabled", "true")
+ conf.setExecutorEnv("spark.shuffle.service.enabled", "true")
+ }
+
+ override def start(): Unit = {
+ super.start()
+ createExecutorPods(getInitialTargetExecutorNumber(sc.getConf))
+ }
+
+ override def stop(): Unit = {
+ // Kill all executor pods indiscriminately
+ killExecutorPods(executorToPod.toVector)
+ killExecutorPods(shutdownToPod.toVector)
+ super.stop()
+ }
+
+ // Dynamic allocation interfaces
+ override def doRequestTotalExecutors(requestedTotal: Int): Future[Boolean] = {
+ logInfo(s"Received doRequestTotalExecutors: $requestedTotal")
+ val n = executorToPod.size
+ val delta = requestedTotal - n
+ if (delta > 0) {
+ logInfo(s"Adding $delta new executors")
+ createExecutorPods(delta)
+ } else if (delta < 0) {
+ val d = -delta
+ val idle = executorToPod.toVector.filter { case (id, _) => !scheduler.isExecutorBusy(id) }
+ if (idle.length > 0) {
+ logInfo(s"Shutting down ${idle.length} idle executors")
+ shutdownExecutors(idle.take(d))
+ }
+ val r = math.max(0, d - idle.length)
+ if (r > 0) {
+ logInfo(s"Shutting down $r non-idle executors")
+ shutdownExecutors(executorToPod.toVector.slice(n - r, n))
+ }
+ }
+ // TODO: are there meaningful failure modes here?
+ Future.successful(true)
+ }
+
+ override def doKillExecutors(executorIds: Seq[String]): Future[Boolean] = {
+ logInfo(s"Received doKillExecutors")
+ killExecutorPods(executorIds.map { id => (id, executorToPod(id)) })
+ Future.successful(true)
+ }
+
+ private def createExecutorPods(n: Int) {
+ for (i <- 1 to n) {
+ executorID += 1
+ executorToPod += ((executorID.toString, createExecutorPod(executorID)))
+ }
+ }
+
+ def shutdownExecutors(idPodPairs: Seq[(String, String)]) {
+ val active = getExecutorIds.toSet
+
+ // Check for any finished shutting down and kill the pods
+ val shutdown = shutdownToPod.toVector.filter { case (e, _) => !active.contains(e) }
+ killExecutorPods(shutdown)
+
+ // Now request shutdown for the new ones.
+ // Move them from executor list to list pending shutdown
+ for ((id, pod) <- idPodPairs) {
+ try {
+ // TODO: 'ask' returns a future - can it be used to check eventual success?
+ Option(driverEndpoint).foreach(_.ask[Boolean](RemoveExecutor(id, ExecutorKilled)))
+ executorToPod -= id
+ shutdownToPod += ((id, pod))
+ } catch {
+ case e: Exception => logError(s"Error shutting down executor $id", e)
+ }
+ }
+ }
+
+ private def killExecutorPods(idPodPairs: Seq[(String, String)]) {
+ for ((id, pod) <- idPodPairs) {
+ try {
+ client.pods().inNamespace(ns).withName(pod).delete()
+ executorToPod -= id
+ shutdownToPod -= id
+ } catch {
+ case e: Exception => logError(s"Error killing executor pod $pod", e)
+ }
+ }
+ }
+
+ def getInitialTargetExecutorNumber(conf: SparkConf,
+ numExecutors: Int =
+ DEFAULT_NUMBER_EXECUTORS): Int = {
+ if (dynamicExecutors) {
+ val minNumExecutors = conf.get(DYN_ALLOCATION_MIN_EXECUTORS)
+ val initialNumExecutors =
+ Utils.getDynamicAllocationInitialExecutors(conf)
+ val maxNumExecutors = conf.get(DYN_ALLOCATION_MAX_EXECUTORS)
+ require(
+ initialNumExecutors >= minNumExecutors && initialNumExecutors <= maxNumExecutors,
+ s"initial executor number $initialNumExecutors must between min executor number " +
+ s"$minNumExecutors and max executor number $maxNumExecutors")
+
+ initialNumExecutors
+ } else {
+ conf.get(EXECUTOR_INSTANCES).getOrElse(numExecutors)
+ }
+ }
+
+ def createExecutorPod(executorNum: Int): String = {
+ // create a single k8s executor pod.
+ val labelMap = Map("type" -> "spark-executor")
+ val podName = s"$sparkExecutorName-$executorNum"
+
+ val submitArgs = mutable.ArrayBuffer.empty[String]
+
+ if (conf.getBoolean("spark.dynamicAllocation.enabled", false)) {
+ submitArgs ++= Vector(
+ "dynamic-executors")
+ }
+
+ submitArgs ++= Vector("org.apache.spark.executor.CoarseGrainedExecutorBackend",
+ "--driver-url", s"$driverURL",
+ "--executor-id", s"$executorNum",
+ "--hostname", "localhost",
--- End diff --
I'm not sure how that plays out in a pod context. "localhost" has been working, but its worth testing alternatives.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16061: [SPARK-18278] [Scheduler] Support native submissi...
Posted by mccheah <gi...@git.apache.org>.
Github user mccheah commented on a diff in the pull request:
https://github.com/apache/spark/pull/16061#discussion_r90321324
--- Diff: core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala ---
@@ -596,6 +599,26 @@ object SparkSubmit extends CommandLineUtils {
}
}
+ if (isKubernetesCluster) {
--- End diff --
We should throw an error for now if in client mode with Kubernetes. There's still open questions about how we can manage the networking there, so we can revisit the support later.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #16061: [SPARK-18278] [Scheduler] Support native submission of s...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on the issue:
https://github.com/apache/spark/pull/16061
We should acknowledge people that have done this in the merged commit, but I don't think it'd make a lot of sense to do a merge rather than squash. The initial Spark SQL commit was a single commit even though it was a much larger project https://github.com/apache/spark/pull/146
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #16061: [SPARK-18278] [Scheduler] Support native submission of s...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/16061
Merged build finished. Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16061: [SPARK-18278] [Scheduler] Support native submissi...
Posted by mogthesprog <gi...@git.apache.org>.
Github user mogthesprog commented on a diff in the pull request:
https://github.com/apache/spark/pull/16061#discussion_r102437781
--- Diff: core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala ---
@@ -596,6 +599,26 @@ object SparkSubmit extends CommandLineUtils {
}
}
+ if (isKubernetesCluster) {
--- End diff --
Hey Guys, really excited about this work by the way. :)
Just wondering if client mode from within a kubernetes cluster will be supported? Not looking to add work, just curious.
I'm wondering whether we could check if the client IP is within the kubernetes container CIDR (this info should be available via the kubernetes API), instead of just blocking client mode altogether. This would support Zeppelin/Jupyter instances running within kube connecting to spark. Which is a big usecase in our case.
We could take the [nodelist](https://kubernetes.io/docs/api-reference/v1/definitions/#_v1_nodelist) and check if the client IP is in any of the [v1.NodeSpec.PodCIDR](https://kubernetes.io/docs/api-reference/v1/definitions/#_v1_nodespec) of any node. Although as k8s clusters get large (nodes>1000), this may become a costly operation.
This would allow for client mode support, provided the client is on one of the kube nodes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #16061: [SPARK-18278] [Scheduler] Support native submission of s...
Posted by liancheng <gi...@git.apache.org>.
Github user liancheng commented on the issue:
https://github.com/apache/spark/pull/16061
@erikerlandson For the RAT failure, you may either add Apache license header to newly added files or add the file to `dev/.rat-excludes`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16061: [SPARK-18278] [Scheduler] Support native submissi...
Posted by tnachen <gi...@git.apache.org>.
Github user tnachen commented on a diff in the pull request:
https://github.com/apache/spark/pull/16061#discussion_r90062695
--- Diff: dev/make-distribution.sh ---
@@ -154,7 +154,9 @@ export MAVEN_OPTS="${MAVEN_OPTS:--Xmx2g -XX:MaxPermSize=512M -XX:ReservedCodeCac
# Store the command as an array because $MVN variable might have spaces in it.
# Normal quoting tricks don't work.
# See: http://mywiki.wooledge.org/BashFAQ/050
-BUILD_COMMAND=("$MVN" -T 1C clean package -DskipTests $@)
+# BUILD_COMMAND=("$MVN" -T 1C clean package -DskipTests $@)
+
+BUILD_COMMAND=("$MVN" -T 2C package -DskipTests $@)
--- End diff --
?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #16061: [SPARK-18278] [Scheduler] Support native submission of s...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/16061
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/69335/
Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16061: [SPARK-18278] [Scheduler] Support native submissi...
Posted by erikerlandson <gi...@git.apache.org>.
Github user erikerlandson commented on a diff in the pull request:
https://github.com/apache/spark/pull/16061#discussion_r90342705
--- Diff: kubernetes/src/main/scala/org/apache/spark/scheduler/cluster/kubernetes/KubernetesClusterSchedulerBackend.scala ---
@@ -0,0 +1,222 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.scheduler.cluster.kubernetes
+
+import collection.JavaConverters._
+import io.fabric8.kubernetes.api.model.PodBuilder
+import io.fabric8.kubernetes.api.model.extensions.JobBuilder
+import io.fabric8.kubernetes.client.{ConfigBuilder, DefaultKubernetesClient}
+import org.apache.spark.internal.config._
+import org.apache.spark.scheduler._
+import org.apache.spark.scheduler.cluster._
+import org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages._
+import org.apache.spark.{SparkConf, SparkContext, SparkException}
+import org.apache.spark.rpc.RpcEndpointAddress
+import org.apache.spark.scheduler.TaskSchedulerImpl
+import org.apache.spark.util.Utils
+
+import scala.collection.mutable
+import scala.util.Random
+import scala.concurrent.Future
+
+private[spark] class KubernetesClusterSchedulerBackend(
+ scheduler: TaskSchedulerImpl,
+ sc: SparkContext)
+ extends CoarseGrainedSchedulerBackend(scheduler, sc.env.rpcEnv) {
+
+ val client = new DefaultKubernetesClient()
+
+ val DEFAULT_NUMBER_EXECUTORS = 2
+ val sparkExecutorName = s"spark-executor-${Random.alphanumeric take 5 mkString("")}".toLowerCase()
+
+ // TODO: do these need mutex guarding?
+ // key is executor id, value is pod name
+ var executorToPod = mutable.Map.empty[String, String] // active executors
+ var shutdownToPod = mutable.Map.empty[String, String] // pending shutdown
+ var executorID = 0
+
+ val sparkImage = conf.get("spark.kubernetes.sparkImage")
+ val clientJarUri = conf.get("spark.executor.jar")
+ val ns = conf.get(
+ "spark.kubernetes.namespace",
+ KubernetesClusterScheduler.defaultNameSpace)
+ val dynamicExecutors = Utils.isDynamicAllocationEnabled(conf)
+
+ // executor back-ends take their configuration this way
+ if (dynamicExecutors) {
+ conf.setExecutorEnv("spark.dynamicAllocation.enabled", "true")
+ conf.setExecutorEnv("spark.shuffle.service.enabled", "true")
+ }
+
+ override def start(): Unit = {
+ super.start()
+ createExecutorPods(getInitialTargetExecutorNumber(sc.getConf))
+ }
+
+ override def stop(): Unit = {
+ // Kill all executor pods indiscriminately
+ killExecutorPods(executorToPod.toVector)
+ killExecutorPods(shutdownToPod.toVector)
+ super.stop()
+ }
+
+ // Dynamic allocation interfaces
+ override def doRequestTotalExecutors(requestedTotal: Int): Future[Boolean] = {
+ logInfo(s"Received doRequestTotalExecutors: $requestedTotal")
+ val n = executorToPod.size
+ val delta = requestedTotal - n
+ if (delta > 0) {
+ logInfo(s"Adding $delta new executors")
+ createExecutorPods(delta)
+ } else if (delta < 0) {
+ val d = -delta
--- End diff --
I had started in on keeping track of new pods until they stop saying "pending", however I think this may be a good use case for a `PodWatcher`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16061: [SPARK-18278] [Scheduler] Support native submissi...
Posted by tnachen <gi...@git.apache.org>.
Github user tnachen commented on a diff in the pull request:
https://github.com/apache/spark/pull/16061#discussion_r90062639
--- Diff: core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala ---
@@ -596,6 +599,26 @@ object SparkSubmit extends CommandLineUtils {
}
}
+ if (isKubernetesCluster) {
--- End diff --
What if in Kubernetes and client mode?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #16061: [SPARK-18278] [Scheduler] Support native submission of s...
Posted by foxish <gi...@git.apache.org>.
Github user foxish commented on the issue:
https://github.com/apache/spark/pull/16061
+1 I think we should close this PR to avoid confusion.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #16061: [SPARK-18278] [Scheduler] Support native submission of s...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/16061
**[Test build #69335 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/69335/consoleFull)** for PR 16061 at commit [`8584913`](https://github.com/apache/spark/commit/8584913d51c1f68c8f9d3cdf142947cd35cdfcc2).
* This patch **fails RAT tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #16061: [SPARK-18278] [Scheduler] Support native submission of s...
Posted by erikerlandson <gi...@git.apache.org>.
Github user erikerlandson commented on the issue:
https://github.com/apache/spark/pull/16061
@iyanuobidele the current head of this branch should be equivalent, but it can't hurt to rebase just in case.
@foxish yesterday occurred to me that it would've been sensible set up a GH organization for this little consortium, and run the PR from there. Presumably we still could. The drawback to making any change to it now is losing the dialog that's happened here already.
The problem I'm interested in solving even more is ensuring that everybody who contributes to this PR gets logged somehow in the upstream commit history when this PR eventually gets merged. I don't think that would happen in the case of a typical squash-merge, although I assume it _would_ using a rebase merge. @rxin do you have any thoughts on that?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #16061: [SPARK-18278] [Scheduler] Support native submission of s...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on the issue:
https://github.com/apache/spark/pull/16061
This is pretty cool.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16061: [SPARK-18278] [Scheduler] Support native submissi...
Posted by erikerlandson <gi...@git.apache.org>.
Github user erikerlandson commented on a diff in the pull request:
https://github.com/apache/spark/pull/16061#discussion_r90092345
--- Diff: kubernetes/src/main/scala/org/apache/spark/scheduler/cluster/kubernetes/KubernetesClusterSchedulerBackend.scala ---
@@ -0,0 +1,222 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.scheduler.cluster.kubernetes
+
+import collection.JavaConverters._
+import io.fabric8.kubernetes.api.model.PodBuilder
+import io.fabric8.kubernetes.api.model.extensions.JobBuilder
+import io.fabric8.kubernetes.client.{ConfigBuilder, DefaultKubernetesClient}
+import org.apache.spark.internal.config._
+import org.apache.spark.scheduler._
+import org.apache.spark.scheduler.cluster._
+import org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages._
+import org.apache.spark.{SparkConf, SparkContext, SparkException}
+import org.apache.spark.rpc.RpcEndpointAddress
+import org.apache.spark.scheduler.TaskSchedulerImpl
+import org.apache.spark.util.Utils
+
+import scala.collection.mutable
+import scala.util.Random
+import scala.concurrent.Future
+
+private[spark] class KubernetesClusterSchedulerBackend(
+ scheduler: TaskSchedulerImpl,
+ sc: SparkContext)
+ extends CoarseGrainedSchedulerBackend(scheduler, sc.env.rpcEnv) {
+
+ val client = new DefaultKubernetesClient()
+
+ val DEFAULT_NUMBER_EXECUTORS = 2
+ val sparkExecutorName = s"spark-executor-${Random.alphanumeric take 5 mkString("")}".toLowerCase()
+
+ // TODO: do these need mutex guarding?
+ // key is executor id, value is pod name
+ var executorToPod = mutable.Map.empty[String, String] // active executors
+ var shutdownToPod = mutable.Map.empty[String, String] // pending shutdown
+ var executorID = 0
+
+ val sparkImage = conf.get("spark.kubernetes.sparkImage")
+ val clientJarUri = conf.get("spark.executor.jar")
+ val ns = conf.get(
+ "spark.kubernetes.namespace",
+ KubernetesClusterScheduler.defaultNameSpace)
+ val dynamicExecutors = Utils.isDynamicAllocationEnabled(conf)
+
+ // executor back-ends take their configuration this way
+ if (dynamicExecutors) {
+ conf.setExecutorEnv("spark.dynamicAllocation.enabled", "true")
+ conf.setExecutorEnv("spark.shuffle.service.enabled", "true")
+ }
+
+ override def start(): Unit = {
+ super.start()
+ createExecutorPods(getInitialTargetExecutorNumber(sc.getConf))
+ }
+
+ override def stop(): Unit = {
+ // Kill all executor pods indiscriminately
+ killExecutorPods(executorToPod.toVector)
+ killExecutorPods(shutdownToPod.toVector)
+ super.stop()
+ }
+
+ // Dynamic allocation interfaces
+ override def doRequestTotalExecutors(requestedTotal: Int): Future[Boolean] = {
+ logInfo(s"Received doRequestTotalExecutors: $requestedTotal")
+ val n = executorToPod.size
+ val delta = requestedTotal - n
+ if (delta > 0) {
+ logInfo(s"Adding $delta new executors")
+ createExecutorPods(delta)
+ } else if (delta < 0) {
+ val d = -delta
--- End diff --
Delta (as currently computed) can become negative, whenever the requested total starts to decrease.
However, this logic warrants a total overhaul, since it was designed around the assumption that `doRequestTotalExecutors` is symmetric, in the sense of having to handle a decreasing total analogously to an increasing total. But responsibilities for executor shutdown vs spin-up aren't symmetric. I might file a small doc PR to clarify that point.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #16061: [SPARK-18278] [Scheduler] Support native submission of s...
Posted by mccheah <gi...@git.apache.org>.
Github user mccheah commented on the issue:
https://github.com/apache/spark/pull/16061
The diff showing our progress so far has moved to https://github.com/apache-spark-on-k8s/spark/pull/200 for anyone following along.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #16061: [SPARK-18278] [Scheduler] Support native submission of s...
Posted by erikerlandson <gi...@git.apache.org>.
Github user erikerlandson commented on the issue:
https://github.com/apache/spark/pull/16061
@rxin, when you say "move all resource managers" does that mean "move scheduler back-ends for mesos, yarn, etc, into some `resource-managers` sub-project" ?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16061: [SPARK-18278] [Scheduler] Support native submissi...
Posted by tnachen <gi...@git.apache.org>.
Github user tnachen commented on a diff in the pull request:
https://github.com/apache/spark/pull/16061#discussion_r90063380
--- Diff: kubernetes/src/main/scala/org/apache/spark/scheduler/cluster/kubernetes/KubernetesClusterSchedulerBackend.scala ---
@@ -0,0 +1,222 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.scheduler.cluster.kubernetes
+
+import collection.JavaConverters._
+import io.fabric8.kubernetes.api.model.PodBuilder
+import io.fabric8.kubernetes.api.model.extensions.JobBuilder
+import io.fabric8.kubernetes.client.{ConfigBuilder, DefaultKubernetesClient}
+import org.apache.spark.internal.config._
+import org.apache.spark.scheduler._
+import org.apache.spark.scheduler.cluster._
+import org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages._
+import org.apache.spark.{SparkConf, SparkContext, SparkException}
+import org.apache.spark.rpc.RpcEndpointAddress
+import org.apache.spark.scheduler.TaskSchedulerImpl
+import org.apache.spark.util.Utils
+
+import scala.collection.mutable
+import scala.util.Random
+import scala.concurrent.Future
+
+private[spark] class KubernetesClusterSchedulerBackend(
+ scheduler: TaskSchedulerImpl,
+ sc: SparkContext)
+ extends CoarseGrainedSchedulerBackend(scheduler, sc.env.rpcEnv) {
+
+ val client = new DefaultKubernetesClient()
+
+ val DEFAULT_NUMBER_EXECUTORS = 2
+ val sparkExecutorName = s"spark-executor-${Random.alphanumeric take 5 mkString("")}".toLowerCase()
+
+ // TODO: do these need mutex guarding?
+ // key is executor id, value is pod name
+ var executorToPod = mutable.Map.empty[String, String] // active executors
+ var shutdownToPod = mutable.Map.empty[String, String] // pending shutdown
+ var executorID = 0
+
+ val sparkImage = conf.get("spark.kubernetes.sparkImage")
+ val clientJarUri = conf.get("spark.executor.jar")
+ val ns = conf.get(
+ "spark.kubernetes.namespace",
+ KubernetesClusterScheduler.defaultNameSpace)
+ val dynamicExecutors = Utils.isDynamicAllocationEnabled(conf)
+
+ // executor back-ends take their configuration this way
+ if (dynamicExecutors) {
+ conf.setExecutorEnv("spark.dynamicAllocation.enabled", "true")
+ conf.setExecutorEnv("spark.shuffle.service.enabled", "true")
+ }
+
+ override def start(): Unit = {
+ super.start()
+ createExecutorPods(getInitialTargetExecutorNumber(sc.getConf))
+ }
+
+ override def stop(): Unit = {
+ // Kill all executor pods indiscriminately
+ killExecutorPods(executorToPod.toVector)
+ killExecutorPods(shutdownToPod.toVector)
+ super.stop()
+ }
+
+ // Dynamic allocation interfaces
+ override def doRequestTotalExecutors(requestedTotal: Int): Future[Boolean] = {
+ logInfo(s"Received doRequestTotalExecutors: $requestedTotal")
+ val n = executorToPod.size
+ val delta = requestedTotal - n
+ if (delta > 0) {
+ logInfo(s"Adding $delta new executors")
+ createExecutorPods(delta)
+ } else if (delta < 0) {
+ val d = -delta
--- End diff --
This shouldn't happen, assert instead
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #16061: [SPARK-18278] [Scheduler] Support native submission of s...
Posted by iyanuobidele <gi...@git.apache.org>.
Github user iyanuobidele commented on the issue:
https://github.com/apache/spark/pull/16061
@erikerlandson are there any new commits to this branch ? Or it's at the same level as @foxish's k8s-support. In case I need to rebase....
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16061: [SPARK-18278] [Scheduler] Support native submissi...
Posted by mccheah <gi...@git.apache.org>.
Github user mccheah commented on a diff in the pull request:
https://github.com/apache/spark/pull/16061#discussion_r90321063
--- Diff: dev/deps/spark-deps-hadoop-2.4 ---
@@ -98,7 +98,7 @@ jersey-client-2.22.2.jar
jersey-common-2.22.2.jar
jersey-container-servlet-2.22.2.jar
jersey-container-servlet-core-2.22.2.jar
-jersey-guava-2.22.2.jar
+jersey-guava-2.22.2.jarshaded-proto
--- End diff --
Any particular reason why this changed? Would be curious to trace how this dependency changed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #16061: [SPARK-18278] [Scheduler] Support native submission of s...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on the issue:
https://github.com/apache/spark/pull/16061
One thing - can we submit a separate pr to move all resource managers into
resource-managers/yarn
resource-managers/mesos
?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #16061: [SPARK-18278] [Scheduler] Support native submission of s...
Posted by erikerlandson <gi...@git.apache.org>.
Github user erikerlandson commented on the issue:
https://github.com/apache/spark/pull/16061
At the risk of redundant announcements, I am going to present this spark-on-kube topic at the [Kubernetes SIG-Apps](https://github.com/kubernetes/community/tree/master/sig-apps) next Monday (Dec 5). It will be similar to my earlier [OpenShift briefing](https://blog.openshift.com/openshift-commons-big-data-sig-2-running-apache-spark-natively-on-kubernetes-with-openshift/) but the focus will be on kube specifically, and I'll update it to cover all the latest developments.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16061: [SPARK-18278] [Scheduler] Support native submissi...
Posted by foxish <gi...@git.apache.org>.
Github user foxish commented on a diff in the pull request:
https://github.com/apache/spark/pull/16061#discussion_r90298908
--- Diff: dev/make-distribution.sh ---
@@ -154,7 +154,9 @@ export MAVEN_OPTS="${MAVEN_OPTS:--Xmx2g -XX:MaxPermSize=512M -XX:ReservedCodeCac
# Store the command as an array because $MVN variable might have spaces in it.
# Normal quoting tricks don't work.
# See: http://mywiki.wooledge.org/BashFAQ/050
-BUILD_COMMAND=("$MVN" -T 1C clean package -DskipTests $@)
+# BUILD_COMMAND=("$MVN" -T 1C clean package -DskipTests $@)
+
+BUILD_COMMAND=("$MVN" -T 2C package -DskipTests $@)
--- End diff --
This should be reverted.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #16061: [SPARK-18278] [Scheduler] Support native submission of s...
Posted by tnachen <gi...@git.apache.org>.
Github user tnachen commented on the issue:
https://github.com/apache/spark/pull/16061
@rxin Makes sense, @srowen also talked about starting the discussion of having a better support for external cluster managers as well.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16061: [SPARK-18278] [Scheduler] Support native submissi...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/spark/pull/16061
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16061: [SPARK-18278] [Scheduler] Support native submissi...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/16061#discussion_r90168566
--- Diff: kubernetes/README.md ---
@@ -0,0 +1,21 @@
+# Pre-requisites
+* maven, JDK and all other pre-requisites for building Spark.
+
+# Steps to compile
+
+* Clone the fork of spark: https://github.com/foxish/spark/ and switch to the k8s-support branch.
+* Build the project
+ * ./build/mvn -Pkubernetes -Phadoop-2.4 -Dhadoop.version=2.4.0 -DskipTests package
+* Ensure that you are pointing to a k8s cluster (kubectl config current-context), which you want to use with spark.
+* Launch a spark-submit job:
+ * `./bin/spark-submit --deploy-mode cluster --class org.apache.spark.examples.SparkPi --master k8s://default --conf spark.executor.instances=5 --conf spark.kubernetes.sparkImage=manyangled/kube-spark:dynamic http://storage.googleapis.com/foxish-spark-distro/original-spark-examples_2.11-2.1.0-SNAPSHOT.jar 10000`
--- End diff --
Do we need to prepare an official image for this?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16061: [SPARK-18278] [Scheduler] Support native submissi...
Posted by mccheah <gi...@git.apache.org>.
Github user mccheah commented on a diff in the pull request:
https://github.com/apache/spark/pull/16061#discussion_r90338702
--- Diff: kubernetes/src/main/scala/org/apache/spark/scheduler/cluster/kubernetes/KubernetesClusterSchedulerBackend.scala ---
@@ -0,0 +1,222 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.scheduler.cluster.kubernetes
+
+import collection.JavaConverters._
+import io.fabric8.kubernetes.api.model.PodBuilder
+import io.fabric8.kubernetes.api.model.extensions.JobBuilder
+import io.fabric8.kubernetes.client.{ConfigBuilder, DefaultKubernetesClient}
+import org.apache.spark.internal.config._
+import org.apache.spark.scheduler._
+import org.apache.spark.scheduler.cluster._
+import org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages._
+import org.apache.spark.{SparkConf, SparkContext, SparkException}
+import org.apache.spark.rpc.RpcEndpointAddress
+import org.apache.spark.scheduler.TaskSchedulerImpl
+import org.apache.spark.util.Utils
+
+import scala.collection.mutable
+import scala.util.Random
+import scala.concurrent.Future
+
+private[spark] class KubernetesClusterSchedulerBackend(
+ scheduler: TaskSchedulerImpl,
+ sc: SparkContext)
+ extends CoarseGrainedSchedulerBackend(scheduler, sc.env.rpcEnv) {
+
+ val client = new DefaultKubernetesClient()
+
+ val DEFAULT_NUMBER_EXECUTORS = 2
+ val sparkExecutorName = s"spark-executor-${Random.alphanumeric take 5 mkString("")}".toLowerCase()
+
+ // TODO: do these need mutex guarding?
+ // key is executor id, value is pod name
+ var executorToPod = mutable.Map.empty[String, String] // active executors
+ var shutdownToPod = mutable.Map.empty[String, String] // pending shutdown
+ var executorID = 0
+
+ val sparkImage = conf.get("spark.kubernetes.sparkImage")
+ val clientJarUri = conf.get("spark.executor.jar")
+ val ns = conf.get(
+ "spark.kubernetes.namespace",
+ KubernetesClusterScheduler.defaultNameSpace)
+ val dynamicExecutors = Utils.isDynamicAllocationEnabled(conf)
+
+ // executor back-ends take their configuration this way
+ if (dynamicExecutors) {
+ conf.setExecutorEnv("spark.dynamicAllocation.enabled", "true")
+ conf.setExecutorEnv("spark.shuffle.service.enabled", "true")
+ }
+
+ override def start(): Unit = {
+ super.start()
+ createExecutorPods(getInitialTargetExecutorNumber(sc.getConf))
+ }
+
+ override def stop(): Unit = {
+ // Kill all executor pods indiscriminately
+ killExecutorPods(executorToPod.toVector)
+ killExecutorPods(shutdownToPod.toVector)
+ super.stop()
+ }
+
+ // Dynamic allocation interfaces
+ override def doRequestTotalExecutors(requestedTotal: Int): Future[Boolean] = {
+ logInfo(s"Received doRequestTotalExecutors: $requestedTotal")
+ val n = executorToPod.size
+ val delta = requestedTotal - n
+ if (delta > 0) {
+ logInfo(s"Adding $delta new executors")
+ createExecutorPods(delta)
+ } else if (delta < 0) {
+ val d = -delta
+ val idle = executorToPod.toVector.filter { case (id, _) => !scheduler.isExecutorBusy(id) }
+ if (idle.length > 0) {
+ logInfo(s"Shutting down ${idle.length} idle executors")
+ shutdownExecutors(idle.take(d))
+ }
+ val r = math.max(0, d - idle.length)
+ if (r > 0) {
+ logInfo(s"Shutting down $r non-idle executors")
+ shutdownExecutors(executorToPod.toVector.slice(n - r, n))
+ }
+ }
+ // TODO: are there meaningful failure modes here?
+ Future.successful(true)
+ }
+
+ override def doKillExecutors(executorIds: Seq[String]): Future[Boolean] = {
+ logInfo(s"Received doKillExecutors")
+ killExecutorPods(executorIds.map { id => (id, executorToPod(id)) })
+ Future.successful(true)
+ }
+
+ private def createExecutorPods(n: Int) {
+ for (i <- 1 to n) {
+ executorID += 1
+ executorToPod += ((executorID.toString, createExecutorPod(executorID)))
+ }
+ }
+
+ def shutdownExecutors(idPodPairs: Seq[(String, String)]) {
+ val active = getExecutorIds.toSet
+
+ // Check for any finished shutting down and kill the pods
+ val shutdown = shutdownToPod.toVector.filter { case (e, _) => !active.contains(e) }
+ killExecutorPods(shutdown)
+
+ // Now request shutdown for the new ones.
+ // Move them from executor list to list pending shutdown
+ for ((id, pod) <- idPodPairs) {
+ try {
+ // TODO: 'ask' returns a future - can it be used to check eventual success?
+ Option(driverEndpoint).foreach(_.ask[Boolean](RemoveExecutor(id, ExecutorKilled)))
+ executorToPod -= id
+ shutdownToPod += ((id, pod))
+ } catch {
+ case e: Exception => logError(s"Error shutting down executor $id", e)
+ }
+ }
+ }
+
+ private def killExecutorPods(idPodPairs: Seq[(String, String)]) {
+ for ((id, pod) <- idPodPairs) {
+ try {
+ client.pods().inNamespace(ns).withName(pod).delete()
+ executorToPod -= id
+ shutdownToPod -= id
+ } catch {
+ case e: Exception => logError(s"Error killing executor pod $pod", e)
+ }
+ }
+ }
+
+ def getInitialTargetExecutorNumber(conf: SparkConf,
+ numExecutors: Int =
+ DEFAULT_NUMBER_EXECUTORS): Int = {
+ if (dynamicExecutors) {
+ val minNumExecutors = conf.get(DYN_ALLOCATION_MIN_EXECUTORS)
+ val initialNumExecutors =
+ Utils.getDynamicAllocationInitialExecutors(conf)
+ val maxNumExecutors = conf.get(DYN_ALLOCATION_MAX_EXECUTORS)
+ require(
+ initialNumExecutors >= minNumExecutors && initialNumExecutors <= maxNumExecutors,
+ s"initial executor number $initialNumExecutors must between min executor number " +
+ s"$minNumExecutors and max executor number $maxNumExecutors")
+
+ initialNumExecutors
+ } else {
+ conf.get(EXECUTOR_INSTANCES).getOrElse(numExecutors)
+ }
+ }
+
+ def createExecutorPod(executorNum: Int): String = {
+ // create a single k8s executor pod.
+ val labelMap = Map("type" -> "spark-executor")
+ val podName = s"$sparkExecutorName-$executorNum"
+
+ val submitArgs = mutable.ArrayBuffer.empty[String]
+
+ if (conf.getBoolean("spark.dynamicAllocation.enabled", false)) {
+ submitArgs ++= Vector(
+ "dynamic-executors")
+ }
+
+ submitArgs ++= Vector("org.apache.spark.executor.CoarseGrainedExecutorBackend",
+ "--driver-url", s"$driverURL",
+ "--executor-id", s"$executorNum",
+ "--hostname", "localhost",
+ "--app-id", "1", // TODO: change app-id per application and pass from driver.
+ "--cores", "1")
+
+ var pod = new PodBuilder()
+ .withNewMetadata()
+ .withLabels(labelMap.asJava)
+ .withName(podName)
+ .endMetadata()
+ .withNewSpec()
+ .withRestartPolicy("OnFailure")
+
+ .addNewContainer().withName("spark-executor").withImage(sparkImage)
+ .withImagePullPolicy("IfNotPresent")
+ .withCommand("/opt/executor.sh")
--- End diff --
That makes sense - essentially the Dockerfile has this line:
`CMD exec ${JAVA_HOME}/bin/java -Xmx$SPARK_EXECUTOR_MEMORY org.apache.spark... --executor-id $SPARK_EXECUTOR_ID ...` etc.
And then we set these environment variables on the container.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #16061: [SPARK-18278] [Scheduler] Support native submission of s...
Posted by foxish <gi...@git.apache.org>.
Github user foxish commented on the issue:
https://github.com/apache/spark/pull/16061
I would think that it would be easier if we continue to keep PRs and issues in foxish/spark since we have the previous PRs and issues over there. It should be easy enough to pick those commits up and update this branch once we review and merge them. It doesn't matter where it lives really, except that it seems like most of it is already on one place.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16061: [SPARK-18278] [Scheduler] Support native submissi...
Posted by jaceklaskowski <gi...@git.apache.org>.
Github user jaceklaskowski commented on a diff in the pull request:
https://github.com/apache/spark/pull/16061#discussion_r104206163
--- Diff: kubernetes/pom.xml ---
@@ -0,0 +1,54 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ Licensed to the Apache Software Foundation (ASF) under one or more
+ ~ contributor license agreements. See the NOTICE file distributed with
+ ~ this work for additional information regarding copyright ownership.
+ ~ The ASF licenses this file to You under the Apache License, Version 2.0
+ ~ (the "License"); you may not use this file except in compliance with
+ ~ the License. You may obtain a copy of the License at
+ ~
+ ~ http://www.apache.org/licenses/LICENSE-2.0
+ ~
+ ~ Unless required by applicable law or agreed to in writing, software
+ ~ distributed under the License is distributed on an "AS IS" BASIS,
+ ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ ~ See the License for the specific language governing permissions and
+ ~ limitations under the License.
+ -->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-parent_2.11</artifactId>
+ <version>2.1.0-SNAPSHOT</version>
--- End diff --
`2.1.0-SNAPSHOT`???
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16061: [SPARK-18278] [Scheduler] Support native submissi...
Posted by tnachen <gi...@git.apache.org>.
Github user tnachen commented on a diff in the pull request:
https://github.com/apache/spark/pull/16061#discussion_r90063456
--- Diff: kubernetes/src/main/scala/org/apache/spark/scheduler/cluster/kubernetes/KubernetesClusterSchedulerBackend.scala ---
@@ -0,0 +1,222 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.scheduler.cluster.kubernetes
+
+import collection.JavaConverters._
+import io.fabric8.kubernetes.api.model.PodBuilder
+import io.fabric8.kubernetes.api.model.extensions.JobBuilder
+import io.fabric8.kubernetes.client.{ConfigBuilder, DefaultKubernetesClient}
+import org.apache.spark.internal.config._
+import org.apache.spark.scheduler._
+import org.apache.spark.scheduler.cluster._
+import org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages._
+import org.apache.spark.{SparkConf, SparkContext, SparkException}
+import org.apache.spark.rpc.RpcEndpointAddress
+import org.apache.spark.scheduler.TaskSchedulerImpl
+import org.apache.spark.util.Utils
+
+import scala.collection.mutable
+import scala.util.Random
+import scala.concurrent.Future
+
+private[spark] class KubernetesClusterSchedulerBackend(
+ scheduler: TaskSchedulerImpl,
--- End diff --
Fix the formatting to conform with Spark style
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16061: [SPARK-18278] [Scheduler] Support native submissi...
Posted by jaceklaskowski <gi...@git.apache.org>.
Github user jaceklaskowski commented on a diff in the pull request:
https://github.com/apache/spark/pull/16061#discussion_r104206250
--- Diff: kubernetes/pom.xml ---
@@ -0,0 +1,54 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ Licensed to the Apache Software Foundation (ASF) under one or more
+ ~ contributor license agreements. See the NOTICE file distributed with
+ ~ this work for additional information regarding copyright ownership.
+ ~ The ASF licenses this file to You under the Apache License, Version 2.0
+ ~ (the "License"); you may not use this file except in compliance with
+ ~ the License. You may obtain a copy of the License at
+ ~
+ ~ http://www.apache.org/licenses/LICENSE-2.0
+ ~
+ ~ Unless required by applicable law or agreed to in writing, software
+ ~ distributed under the License is distributed on an "AS IS" BASIS,
+ ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ ~ See the License for the specific language governing permissions and
+ ~ limitations under the License.
+ -->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-parent_2.11</artifactId>
+ <version>2.1.0-SNAPSHOT</version>
+ <relativePath>../pom.xml</relativePath>
+ </parent>
+
+ <artifactId>spark-kubernetes_2.11</artifactId>
--- End diff --
`${scala.binary.version}`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16061: [SPARK-18278] [Scheduler] Support native submissi...
Posted by jaceklaskowski <gi...@git.apache.org>.
Github user jaceklaskowski commented on a diff in the pull request:
https://github.com/apache/spark/pull/16061#discussion_r104205883
--- Diff: kubernetes/README.md ---
@@ -0,0 +1,21 @@
+# Pre-requisites
+* maven, JDK and all other pre-requisites for building Spark.
+
+# Steps to compile
+
+* Clone the fork of spark: https://github.com/foxish/spark/ and switch to the k8s-support branch.
+* Build the project
+ * ./build/mvn -Pkubernetes -Phadoop-2.4 -Dhadoop.version=2.4.0 -DskipTests package
--- End diff --
I think `hadoop-2.4` is gone in master.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #16061: [SPARK-18278] [Scheduler] Support native submission of s...
Posted by erikerlandson <gi...@git.apache.org>.
Github user erikerlandson commented on the issue:
https://github.com/apache/spark/pull/16061
Since we have at least four people with interest in working on the code, I have a workflow proposal, which is that we all submit candidate updates to #16061 in the form of PRs against [erikerlandson:spark-18278-k8s-native](https://github.com/erikerlandson/spark/tree/spark-18278-k8s-native) (recursive PRs against a PR!)
I did this with Anirudh against his `k8s-support` branch, and it seemed to work fine. It should help keep asynchronous contributions sane, and provides the usual durable public forum for discussing the updates as they evolve.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #16061: [SPARK-18278] [Scheduler] Support native submission of s...
Posted by ash211 <gi...@git.apache.org>.
Github user ash211 commented on the issue:
https://github.com/apache/spark/pull/16061
Another external scheduler backend I'm aware of is Two Sigma's scheduler backend for the system they've created called [Cook](https://github.com/twosigma/Cook). See [CoarseCookSchedulerBackend.scala](https://github.com/twosigma/spark/blob/2.0.2-cook/core/src/main/scala/org/apache/spark/scheduler/CoarseCookSchedulerBackend.scala)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org