Skip to content

[SPARK-9104][SPARK-9105][SPARK-9106][SPARK-9107][CORE] Netty network layer memory usage on webUI #7753

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
liyezhang556520 wants to merge 40 commits into from
Closed

[SPARK-9104][SPARK-9105][SPARK-9106][SPARK-9107][CORE] Netty network layer memory usage on webUI #7753

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
40 commits
Select commit Hold shift + click to select a range
39ba441
spark-9104 first draft version
liyezhang556520 Aug 17, 2015
ecc1044
Merge remote-tracking branch 'apache/master' into netMem-9104
liyezhang556520 Aug 17, 2015
2101538
show N/A for nio
liyezhang556520 Aug 17, 2015
9ccaf88
handle executor add and remove event for memotyTab
liyezhang556520 Aug 18, 2015
13c17fb
show removed executors info on page
liyezhang556520 Aug 19, 2015
c9b44b1
add stage memory trace
liyezhang556520 Aug 19, 2015
984feaf
add history support for heartbeat event
liyezhang556520 Aug 20, 2015
2501c82
limit history event log frequency
liyezhang556520 Aug 20, 2015
e0ae855
add some comments for EventLoggingListener
liyezhang556520 Aug 20, 2015
7491279
Merge remote-tracking branch 'apache/master' into netMem-9104
liyezhang556520 Aug 20, 2015
424c172
scala style fix
liyezhang556520 Aug 20, 2015
f21a804
remove executor port and fix test failure
liyezhang556520 Aug 21, 2015
2f3d30b
merge spache/master after master updated
liyezhang556520 Sep 25, 2015
7b846a2
work with JavaConverters
liyezhang556520 Oct 9, 2015
41874aa
Merge remote-tracking branch 'apache/master' into netMem-9104
liyezhang556520 Oct 9, 2015
0531d0f
Merge remote-tracking branch 'apache/master' into netMem-9104
liyezhang556520 Oct 29, 2015
27b7da1
refine the code according to Imran's comments and the design doc
liyezhang556520 Nov 2, 2015
a8fcf74
Merge remote-tracking branch 'apache/master' into netMem-9104
liyezhang556520 Nov 2, 2015
f2f0e64
fix scala style test
liyezhang556520 Nov 3, 2015
5f7a999
capitalize class name
liyezhang556520 Nov 3, 2015
5ad7a6a
change task metrics json format back to origin
liyezhang556520 Nov 3, 2015
c836fb9
Merge remote-tracking branch 'apache/master' into netMem-9104
liyezhang556520 Nov 3, 2015
b5aa4da
Merge remote-tracking branch 'apache/master' into netMem-9104
liyezhang556520 Nov 5, 2015
e8e2bdd
Merge remote-tracking branch 'apache/master' into netMem-9104
liyezhang556520 Nov 6, 2015
1dffa29
accroding to Imran's comment, refine the code
liyezhang556520 Nov 17, 2015
75e63c3
add first test case
liyezhang556520 Nov 17, 2015
0c1241c
fix scala style
liyezhang556520 Nov 17, 2015
c78628e
add more test cases, with eventloging test left
liyezhang556520 Nov 19, 2015
a93bd96
scala style fix
liyezhang556520 Nov 19, 2015
89214f3
fix test fail and add event logging unit test
liyezhang556520 Nov 23, 2015
1ed48c1
scala syle
liyezhang556520 Nov 23, 2015
cb307aa
merge to apache/master branch, fix merge conflict
liyezhang556520 Nov 24, 2015
b438077
roll back useless change
liyezhang556520 Nov 24, 2015
4123ac7
modify the code according to Imran's comments, mainly with unit test
liyezhang556520 Dec 8, 2015
2ce9fd9
fix scala style
liyezhang556520 Dec 8, 2015
17d094e
merge to master branch with tests update
liyezhang556520 Dec 8, 2015
4b3dbe4
change port to option and some bug fixes
liyezhang556520 Dec 9, 2015
0ea7cab
address comments of code refinement
liyezhang556520 Jan 12, 2016
5e031ce
merge to latest master branch from spark-9104-draft
liyezhang556520 Jan 12, 2016
87f8172
fix import ordering error
liyezhang556520 Jan 12, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
remove executor port and fix test failure
  • Loading branch information
@liyezhang556520
liyezhang556520 committed Aug 21, 2015
commit f21a8047e615c7dc17e5a823ebaabbbc84fffdb3
1 change: 0 additions & 1 deletion core/src/main/scala/org/apache/spark/executor/Executor.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,7 +87,6 @@ private[spark] class Executor(

private val executorMetrics: ExecutorMetrics = new ExecutorMetrics
executorMetrics.setHostname(Utils.localHostName)
Copy link
Contributor

@squito squito Sep 21, 2015

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

is the hostname really enough? What about with multiple executors on the same host? The Heartbeat already has the executorId, maybe that is all we need?

Copy link
Contributor Author

@liyezhang556520 liyezhang556520 Sep 24, 2015

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's better to use HOST:PORT, but executor port cannot get here, which should get from RpcEnv.adress.port. We can get executorId on the driver side when receiving the message, that might be enough to identify the different executors, but since we will show the removed executors on the page, so we cannot know where the executor locate by only executorId, because the Executor tab only shows the active executors. we can remove the hostname here if we support showing the removed executors on Executor tab.

Copy link
Contributor

@squito squito Nov 25, 2015

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

sorry I forgot about this in my last round. I don't understand the first sentence of your reply -- why can't you get the hostname & port? it looks like you can do exactly what you suggested to get the host & port from the rpc env with executorMetrics.setHostPort(env.rpcEnv.address.hostPort)

Copy link
Contributor Author

@liyezhang556520 liyezhang556520 Nov 30, 2015

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sorry that I forgot the reason why I didn't get the port from rpcEnv originally, but it seems we can get the port from it directly. Let me try to add it back.

executorMetrics.setPort(Some(env.blockTransferService.port))

// Create an RpcEndpoint for receiving RPCs from the driver
private val executorEndpoint = env.rpcEnv.setupEndpoint(
Expand Down
9 changes: 0 additions & 9 deletions core/src/main/scala/org/apache/spark/executor/ExecutorMetrics.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,15 +36,6 @@ class ExecutorMetrics extends Serializable {
def hostname: String = _hostname
private[spark] def setHostname(value: String) = _hostname = value

/**
* Host's port the executor runs on
*/
private var _port: Option[Int] = _
def port: Option[Int] = _port
private[spark] def setPort(value: Option[Int]) = _port = value

def hostPort: String = hostname + ":" + port.getOrElse(0)

private var _transportMetrics: Option[TransportMetrics] = None
Copy link
Contributor

@squito squito Nov 3, 2015

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

now that nio is gone, does this still to be an Option, or will it always be present?

Copy link
Contributor Author

@liyezhang556520 liyezhang556520 Nov 23, 2015

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

changed it to be always present in the updated code.

def transportMetrics: Option[TransportMetrics] = _transportMetrics
private[spark] def setTransportMetrics(value: Option[TransportMetrics]) = {
Expand Down
2 changes: 1 addition & 1 deletion core/src/main/scala/org/apache/spark/ui/memory/MemoryTab.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -120,7 +120,7 @@ class MemoryUIInfo {
case Some(transportMemSize) => transportInfo
case _ => Some(new transportMemSize)
}
executorAddress = execMetrics.hostPort
executorAddress = execMetrics.hostname
if (execMetrics.transportMetrics.isDefined) {
transportInfo.get.updateTransport(execMetrics.transportMetrics.get)
}
Expand Down
2 changes: 0 additions & 2 deletions core/src/main/scala/org/apache/spark/util/JsonProtocol.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -290,7 +290,6 @@ private[spark] object JsonProtocol {
val transportMetrics = executorMetrics.transportMetrics.map(
transportMetricsToJson).getOrElse(JNothing)
("Executor Hostname" -> executorMetrics.hostname) ~
("Executor Port" -> executorMetrics.port.map(new JInt(_)).getOrElse(JNothing)) ~
("TransportMetrics" -> transportMetrics)
}

Expand Down Expand Up @@ -715,7 +714,6 @@ private[spark] object JsonProtocol {
def executorMetricsFromJson(json: JValue): ExecutorMetrics = {
val metrics = new ExecutorMetrics
metrics.setHostname((json \ "Executor Hostname").extract[String])
metrics.setPort(Utils.jsonOption(json \ "Executor Port").map(_.extract[Int]))
metrics.setTransportMetrics(
Utils.jsonOption((json \ "TransportMetrics")).map(transportMetrisFromJson))
metrics
Expand Down
4 changes: 2 additions & 2 deletions core/src/test/scala/org/apache/spark/HeartbeatReceiverSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@ import org.mockito.Mockito.{mock, spy, verify, when}
import org.mockito.Matchers
import org.mockito.Matchers._

import org.apache.spark.executor.TaskMetrics
import org.apache.spark.executor.{ExecutorMetrics, TaskMetrics}
import org.apache.spark.rpc.{RpcCallContext, RpcEndpoint, RpcEnv, RpcEndpointRef}
import org.apache.spark.scheduler._
import org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages._
Expand Down Expand Up @@ -218,7 +218,7 @@ class HeartbeatReceiverSuite
assert(!response.reregisterBlockManager)
// Additionally verify that the scheduler callback is called with the correct parameters
verify(scheduler).executorHeartbeatReceived(
Matchers.eq(executorId), null, Matchers.eq(
Matchers.eq(executorId), Matchers.eq(null), Matchers.eq(
Array(1L -> metrics)), Matchers.eq(blockManagerId))
}
}
Expand Down
9 changes: 6 additions & 3 deletions core/src/test/scala/org/apache/spark/util/JsonProtocolSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -83,8 +83,8 @@ class JsonProtocolSuite extends SparkFunSuite {
val executorAdded = SparkListenerExecutorAdded(executorAddedTime, "exec1",
new ExecutorInfo("Hostee.awesome.com", 11, logUrlMap))
val executorRemoved = SparkListenerExecutorRemoved(executorRemovedTime, "exec2", "test reason")
val executorMetricsUpdate = SparkListenerExecutorMetricsUpdate("exec3", null, Seq(
(1L, 2, 3, makeTaskMetrics(300L, 400L, 500L, 600L, 700, 800,
val executorMetricsUpdate = SparkListenerExecutorMetricsUpdate("exec3", new ExecutorMetrics,
Seq((1L, 2, 3, makeTaskMetrics(300L, 400L, 500L, 600L, 700, 800,
hasHadoopInput = true, hasOutput = true))))

testEvent(stageSubmitted, stageSubmittedJsonString)
Expand Down Expand Up @@ -1619,7 +1619,10 @@ class JsonProtocolSuite extends SparkFunSuite {
|{
| "Event": "SparkListenerExecutorMetricsUpdate",
| "Executor ID": "exec3",
| "Metrics Updated": [
| "Executor Metrics Updated": {
| "Executor Hostname": null
| },
| "Task Metrics Updated": [
| {
| "Task ID": 1,
| "Stage ID": 2,
Expand Down
8 changes: 4 additions & 4 deletions sql/core/src/test/scala/org/apache/spark/sql/execution/ui/SQLListenerSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -108,15 +108,15 @@ class SQLListenerSuite extends SparkFunSuite with SharedSQLContext {

assert(listener.getExecutionMetrics(0).isEmpty)

listener.onExecutorMetricsUpdate(SparkListenerExecutorMetricsUpdate("", Seq(
listener.onExecutorMetricsUpdate(SparkListenerExecutorMetricsUpdate("", null, Seq(
// (task id, stage id, stage attempt, metrics)
(0L, 0, 0, createTaskMetrics(accumulatorUpdates)),
(1L, 0, 0, createTaskMetrics(accumulatorUpdates))
)))

assert(listener.getExecutionMetrics(0) === accumulatorUpdates.mapValues(_ * 2))

listener.onExecutorMetricsUpdate(SparkListenerExecutorMetricsUpdate("", Seq(
listener.onExecutorMetricsUpdate(SparkListenerExecutorMetricsUpdate("", null, Seq(
// (task id, stage id, stage attempt, metrics)
(0L, 0, 0, createTaskMetrics(accumulatorUpdates)),
(1L, 0, 0, createTaskMetrics(accumulatorUpdates.mapValues(_ * 2)))
Expand All @@ -127,7 +127,7 @@ class SQLListenerSuite extends SparkFunSuite with SharedSQLContext {
// Retrying a stage should reset the metrics
listener.onStageSubmitted(SparkListenerStageSubmitted(createStageInfo(0, 1)))

listener.onExecutorMetricsUpdate(SparkListenerExecutorMetricsUpdate("", Seq(
listener.onExecutorMetricsUpdate(SparkListenerExecutorMetricsUpdate("", null, Seq(
// (task id, stage id, stage attempt, metrics)
(0L, 0, 1, createTaskMetrics(accumulatorUpdates)),
(1L, 0, 1, createTaskMetrics(accumulatorUpdates))
Expand Down Expand Up @@ -167,7 +167,7 @@ class SQLListenerSuite extends SparkFunSuite with SharedSQLContext {
// Summit a new stage
listener.onStageSubmitted(SparkListenerStageSubmitted(createStageInfo(1, 0)))

listener.onExecutorMetricsUpdate(SparkListenerExecutorMetricsUpdate("", Seq(
listener.onExecutorMetricsUpdate(SparkListenerExecutorMetricsUpdate("", null, Seq(
// (task id, stage id, stage attempt, metrics)
(0L, 1, 0, createTaskMetrics(accumulatorUpdates)),
(1L, 1, 0, createTaskMetrics(accumulatorUpdates))
Expand Down