You are viewing a plain text version of this content. The canonical link for it is here.

Posted to reviews@spark.apache.org by steveloughran <gi...@git.apache.org> on 2015/10/20 17:01:27 UTC

[GitHub] spark pull request: SPARK-1537: pure service API and test service

GitHub user steveloughran opened a pull request:

    https://github.com/apache/spark/pull/9182

    SPARK-1537: pure service API and test service

    This is purely the yarn/src/main and yarn/src/test bits of the YARN ATS integration: the extension model to load and run implementations of `SchedulerExtensionService` in the yarn cluster scheduler process —and to stop them afterwards.
    
    There's duplication between the two schedulers, yarn-client and yarn-cluster, at least in terms of setting everything up, because the common superclass, `YarnSchedulerBackend` is in spark-core, and the extension services need the YARN app/attempt IDs.
    
    If you look at how the the extension services are loaded, the case class `SchedulerExtensionServiceBinding` is used to pass in config info -currently just the spark context and the yarn IDs, of which one, the attemptID, will be null when running client-side. I'm passing in a case class to ensure that it would be possible in future to add extra arguments to the binding class, yet, as the method signature will not have changed, still be able to load existing services.
    
    There's no functional extension service here, just one for testing. The real tests come in the bigger pull requests. At the same time, there's no restriction of this extension service purely to the ATS history publisher. Anything else that wants to listen to the spark context and publish events could use this, and I'd also consider writing one for the YARN-913 registry service, so that the URLs of the web UI would be locatable through that (low priority; would make more sense if integrated with a REST client).
    
    There's no minicluster test. Given the test execution overhead of setting up minicluster tests, it'd  probably be better to add an extension service into one of the existing tests.

You can merge this pull request into a Git repository by running:

    $ git pull https://github.com/steveloughran/spark stevel/feature/SPARK-1537-service

Alternatively you can review and apply these changes as the patch at:

    https://github.com/apache/spark/pull/9182.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 #9182
    
----
commit e796bb33d3128587f87529284fe45b37e1205de5
Author: Steve Loughran <st...@hortonworks.com>
Date:   2015-10-20T14:35:40Z

    SPARK-1537: pure service API and test service

----


---
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: [SPARK-11314] [YARN] add service API and test ...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the pull request:

    https://github.com/apache/spark/pull/9182#issuecomment-151877402
  
    Test PASSed.
    Refer to this link for build results (access rights to CI server needed): 
    https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/44524/
    Test PASSed.


---
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: [SPARK-11314] [YARN] add service API and test ...

Posted by vanzin <gi...@git.apache.org>.

Github user vanzin commented on a diff in the pull request:

    https://github.com/apache/spark/pull/9182#discussion_r43251937
  
    --- Diff: yarn/src/main/scala/org/apache/spark/scheduler/cluster/SchedulerExtensionService.scala ---
    @@ -0,0 +1,136 @@
    +/*
    + * 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
    +
    +import java.util.concurrent.atomic.AtomicBoolean
    +
    +import org.apache.hadoop.yarn.api.records.{ApplicationAttemptId, ApplicationId}
    +
    +import org.apache.spark.util.Utils
    +import org.apache.spark.{Logging, SparkContext}
    --- End diff --
    
    nit: this goes before the previous import


---
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: [SPARK-11314] [YARN] add service API and test ...

Posted by vanzin <gi...@git.apache.org>.

Github user vanzin commented on a diff in the pull request:

    https://github.com/apache/spark/pull/9182#discussion_r43254504
  
    --- Diff: yarn/src/test/scala/org/apache/spark/scheduler/cluster/StubApplicationAttemptId.scala ---
    @@ -0,0 +1,50 @@
    +/*
    + * 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
    +
    +import org.apache.hadoop.yarn.api.records.{ApplicationAttemptId, ApplicationId}
    +
    +
    --- End diff --
    
    nit: nuke the extra empty lines.


---
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: [SPARK-11314] [YARN] add service API and test ...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the pull request:

    https://github.com/apache/spark/pull/9182#issuecomment-152164469
  
    **[Test build #44593 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/44593/consoleFull)** for PR 9182 at commit [`8a6a1f1`](https://github.com/apache/spark/commit/8a6a1f13235fd00dcc58c4106b0314098f961e67).
     * This patch **fails Scala style 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 pull request: [SPARK-11314] [YARN] add service API and test ...

Posted by vanzin <gi...@git.apache.org>.

Github user vanzin commented on a diff in the pull request:

    https://github.com/apache/spark/pull/9182#discussion_r43284929
  
    --- Diff: yarn/src/main/scala/org/apache/spark/scheduler/cluster/YarnSchedulerBackend.scala ---
    @@ -51,6 +51,38 @@ private[spark] abstract class YarnSchedulerBackend(
     
       private implicit val askTimeout = RpcUtils.askRpcTimeout(sc.conf)
     
    +  /** Application ID. Must be set by a subclass before starting the service */
    +  private var appId: ApplicationId = null
    +
    +  /** Attempt ID. This is unset for client-side schedulers */
    +  private var attemptId: Option[ApplicationAttemptId] = None
    +
    +  /** Scheduler extension services */
    +  private val services: SchedulerExtensionServices = new SchedulerExtensionServices()
    +
    +  /**
    +    * Bind to YARN. This *must* be done before calling [[start()]].
    +    *
    +    * @param appId YARN application ID
    +    * @param attemptId Optional YARN attempt ID
    +    */
    +  protected def bindToYarn(appId: ApplicationId, attemptId: Option[ApplicationAttemptId]): Unit = {
    +    this.appId = appId
    +    this.attemptId = attemptId
    +  }
    +
    +  override def start() {
    +    require(appId != null, "application ID unset")
    +    val binding = SchedulerExtensionServiceBinding(sc, appId, attemptId)
    +    services.start(binding)
    +    super.start()
    +  }
    +
    +  override def stop(): Unit = {
    +    super.stop()
    --- End diff --
    
    Good question about what to do with exceptions. I think the best approach is to have `stop()` be called from a method that catches and logs exceptions. That way you kill two problems: you'll not replace any exceptions caused by `super.stop()`, and you'll make sure `stop()` is called for all existing services (which the current code doesn't do when exceptions are thrown).


---
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: [SPARK-11314] [YARN] add service API and test ...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the pull request:

    https://github.com/apache/spark/pull/9182#issuecomment-151979760
  
    Test PASSed.
    Refer to this link for build results (access rights to CI server needed): 
    https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/44536/
    Test PASSed.


---
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: [SPARK-11314] [YARN] add service API and test ...

Posted by steveloughran <gi...@git.apache.org>.

Github user steveloughran commented on a diff in the pull request:

    https://github.com/apache/spark/pull/9182#discussion_r43373740
  
    --- Diff: yarn/src/main/scala/org/apache/spark/scheduler/cluster/SchedulerExtensionService.scala ---
    @@ -0,0 +1,155 @@
    +/*
    + * 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
    +
    +import java.util.concurrent.atomic.AtomicBoolean
    +
    +import org.apache.hadoop.yarn.api.records.{ApplicationAttemptId, ApplicationId}
    +
    +import org.apache.spark.util.Utils
    +import org.apache.spark.{Logging, SparkContext}
    +
    +/**
    + * An extension service that can be loaded into a Spark YARN scheduler.
    + * A Service that can be started and stopped.
    + *
    + * 1. For implementations to be loadable by [[SchedulerExtensionServices]],
    + * they must provide an empty constructor.
    + * 2. The `stop()` operation MUST be idempotent, and succeed even if `start()` was
    + * never invoked.
    + */
    +trait SchedulerExtensionService {
    +
    +  /**
    +   * Start the extension service. This should be a no-op if
    +   * called more than once.
    +   * @param binding binding to the spark application and YARN
    +   */
    +  def start(binding: SchedulerExtensionServiceBinding): Unit
    +
    +  /**
    +   * Stop the service
    +   * The `stop()` operation MUST be idempotent, and succeed even if `start()` was
    +   * never invoked.
    +   */
    +  def stop(): Unit
    +}
    +
    +/**
    + * Binding information for a [[SchedulerExtensionService]].
    + *
    + * The attempt ID will be set if the service is started within a YARN application master;
    + * there is then a different attempt ID for every time that AM is restarted.
    + * When the service binding is instantiated on a client, there's no attempt ID, as it lacks
    + * this information.
    + * @param sparkContext current spark context
    + * @param applicationId YARN application ID
    + * @param attemptId YARN attemptID -if known.
    + */
    +case class SchedulerExtensionServiceBinding(
    +    sparkContext: SparkContext,
    +    applicationId: ApplicationId,
    +    attemptId: Option[ApplicationAttemptId] = None)
    +
    +/**
    + * Container for [[SchedulerExtensionService]] instances.
    + *
    + * Loads Extension Services from the configuration property
    + * `"spark.yarn.services"`, instantiates and starts them.
    + * When stopped, it stops all child entries.
    + *
    + * The order in which child extension services are started and stopped
    + * is undefined.
    + *
    + */
    +private[spark] class SchedulerExtensionServices extends SchedulerExtensionService
    +    with Logging {
    +  private var services: List[SchedulerExtensionService] = Nil
    +  private val started = new AtomicBoolean(false)
    +  private var binding: SchedulerExtensionServiceBinding = _
    +
    +  /**
    +   * Binding operation will load the named services and call bind on them too; the
    +   * entire set of services are then ready for `init()` and `start()` calls.
    +   *
    +   * @param binding binding to the spark application and YARN
    +   */
    +  def start(binding: SchedulerExtensionServiceBinding): Unit = {
    +    if (started.getAndSet(true)) {
    +      logWarning("Ignoring re-entrant start operation")
    +      return
    +    }
    +    require(binding.sparkContext != null, "Null context parameter")
    +    require(binding.applicationId != null, "Null appId parameter")
    +    this.binding = binding
    +    val sparkContext = binding.sparkContext
    +    val appId = binding.applicationId
    +    val attemptId = binding.attemptId
    +    logInfo(s"Starting Yarn extension services with app ${binding.applicationId}" +
    +      s" and attemptId $attemptId")
    +
    +    services = sparkContext.getConf.getOption(SchedulerExtensionServices.SPARK_YARN_SERVICES)
    +      .map { s =>
    +        s.split(",").map(_.trim()).filter(!_.isEmpty)
    +          .map { sClass =>
    +            val instance = Utils.classForName(sClass)
    +              .newInstance()
    +              .asInstanceOf[SchedulerExtensionService]
    +            // bind this service
    +            instance.start(binding)
    +            logInfo(s"Service $sClass started")
    +            instance
    +          }.toList
    +      }.getOrElse(Nil)
    +  }
    +
    +  /**
    +   * Get the list of services.
    +   *
    +   * @return a list of services; Nil until the service is started
    +   */
    +  def getServices: List[SchedulerExtensionService] = {
    --- End diff --
    
    OK


---
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: [SPARK-11314] [YARN] add service API and test ...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the pull request:

    https://github.com/apache/spark/pull/9182#issuecomment-151914677
  
     Build triggered.


---
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