[SPARK-23429][CORE] Add executor memory metrics to heartbeat and expose in executors REST API #21221
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,71 @@ | ||
| /* | ||
| * 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 | ||
|
|
||
| import java.util.concurrent.TimeUnit | ||
|
|
||
| import org.apache.spark.executor.ExecutorMetrics | ||
| import org.apache.spark.internal.Logging | ||
| import org.apache.spark.memory.MemoryManager | ||
| import org.apache.spark.metrics.ExecutorMetricType | ||
| import org.apache.spark.util.{ThreadUtils, Utils} | ||
|
|
||
| /** | ||
| * Creates a heartbeat thread which will call the specified reportHeartbeat function at | ||
| * intervals of intervalMs. | ||
| * | ||
| * @param memoryManager the memory manager for execution and storage memory. | ||
| * @param reportHeartbeat the heartbeat reporting function to call. | ||
| * @param name the thread name for the heartbeater. | ||
| * @param intervalMs the interval between heartbeats. | ||
| */ | ||
| private[spark] class Heartbeater( | ||
| memoryManager: MemoryManager, | ||
| reportHeartbeat: () => Unit, | ||
| name: String, | ||
| intervalMs: Long) extends Logging { | ||
| // Executor for the heartbeat task | ||
| private val heartbeater = ThreadUtils.newDaemonSingleThreadScheduledExecutor(name) | ||
|
|
||
| /** Schedules a task to report a heartbeat. */ | ||
| def start(): Unit = { | ||
| // Wait a random interval so the heartbeats don't end up in sync | ||
| val initialDelay = intervalMs + (math.random * intervalMs).asInstanceOf[Int] | ||
|
|
||
| val heartbeatTask = new Runnable() { | ||
| override def run(): Unit = Utils.logUncaughtExceptions(reportHeartbeat()) | ||
| } | ||
| heartbeater.scheduleAtFixedRate(heartbeatTask, initialDelay, intervalMs, TimeUnit.MILLISECONDS) | ||
| } | ||
|
|
||
| /** Stops the heartbeat thread. */ | ||
| def stop(): Unit = { | ||
| heartbeater.shutdown() | ||
| heartbeater.awaitTermination(10, TimeUnit.SECONDS) | ||
| } | ||
|
|
||
| /** | ||
| * Get the current executor level metrics. These are returned as an array, with the index | ||
| * determined by MetricGetter.values | ||
| */ | ||
| def getCurrentMetrics(): ExecutorMetrics = { | ||
| val metrics = ExecutorMetricType.values.map(_.getMetricValue(memoryManager)).toArray | ||
| new ExecutorMetrics(metrics) | ||
| } | ||
| } | ||
|
|
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,81 @@ | ||
| /* | ||
| * 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.executor | ||
|
|
||
| import org.apache.spark.annotation.DeveloperApi | ||
| import org.apache.spark.metrics.ExecutorMetricType | ||
|
|
||
| /** | ||
| * :: DeveloperApi :: | ||
| * Metrics tracked for executors and the driver. | ||
| * | ||
| * Executor-level metrics are sent from each executor to the driver as part of the Heartbeat. | ||
| */ | ||
| @DeveloperApi | ||
| class ExecutorMetrics private[spark] extends Serializable { | ||
|
|
||
| // Metrics are indexed by MetricGetter.values | ||
| private val metrics = new Array[Long](ExecutorMetricType.values.length) | ||
|
Contributor
There was a problem hiding this comment. Out of curiosity, why are we using an array here with index-based fetching? We could use a struct / case class to represent these metrics. But I suppose the size of the payload we send is smaller if we use an Array, and we don't want to pay serialization costs?
Contributor
There was a problem hiding this comment. I suggested this earlier in the reviews. Most of the operations for dealing with this data want to iterate over all the fields. its much easier this way vs. having a bazillion if (x.fizz > y.fizz) {
y.fizz = x.fizz
}
if (x.buzz > y.buzz) {
y.buzz = x.buzz
}
...
Contributor
There was a problem hiding this comment. Yup that's fine - I did some googling, unfortunately there isn't a great way to iterate over fields of a case class. You could create a thin wrapper object around the array instead though, if we really think the nicer API is worthwhile: Or even this: The latter which would be better because you'd be guaranteed to create the struct with the right number of metrics. Though such abstractions are not necessary by any means.
Contributor
Author
There was a problem hiding this comment. Is it likely that users would want to access the individual fields, rather than iterating through all? The 1st option would be a bit nicer if so.
Contributor
There was a problem hiding this comment. Unclear - if we expose these metrics to some external consumer via an API for example, then we almost certainly want to have a schema labelling these fields for consumption by e.g. dashboards. I think what we have here is fine for now. |
||
|
|
||
| // the first element is initialized to -1, indicating that the values for the array | ||
| // haven't been set yet. | ||
| metrics(0) = -1 | ||
|
|
||
| /** Returns the value for the specified metricType. */ | ||
| def getMetricValue(metricType: ExecutorMetricType): Long = { | ||
| metrics(ExecutorMetricType.metricIdxMap(metricType)) | ||
| } | ||
|
|
||
| /** Returns true if the values for the metrics have been set, false otherwise. */ | ||
| def isSet(): Boolean = metrics(0) > -1 | ||
|
|
||
| private[spark] def this(metrics: Array[Long]) { | ||
| this() | ||
| Array.copy(metrics, 0, this.metrics, 0, Math.min(metrics.size, this.metrics.size)) | ||
| } | ||
|
|
||
| /** | ||
| * Constructor: create the ExecutorMetrics with the values specified. | ||
| * | ||
| * @param executorMetrics map of executor metric name to value | ||
| */ | ||
| private[spark] def this(executorMetrics: Map[String, Long]) { | ||
| this() | ||
| (0 until ExecutorMetricType.values.length).foreach { idx => | ||
| metrics(idx) = executorMetrics.getOrElse(ExecutorMetricType.values(idx).name, 0L) | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Compare the specified executor metrics values with the current executor metric values, | ||
| * and update the value for any metrics where the new value for the metric is larger. | ||
| * | ||
| * @param executorMetrics the executor metrics to compare | ||
| * @return if there is a new peak value for any metric | ||
| */ | ||
| private[spark] def compareAndUpdatePeakValues(executorMetrics: ExecutorMetrics): Boolean = { | ||
| var updated: Boolean = false | ||
|
||
|
|
||
| (0 until ExecutorMetricType.values.length).foreach { idx => | ||
| if ( executorMetrics.metrics(idx) > metrics(idx)) { | ||
|
||
| updated = true | ||
| metrics(idx) = executorMetrics.metrics(idx) | ||
| } | ||
| } | ||
| updated | ||
| } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
nit: remove extra spaces for better indent