update

cloud-fan · cloud-fan · commit 39dd0eed15d9 · 2016-03-31T10:10:18.000+08:00
diff --git a/core/src/main/java/org/apache/spark/SparkExecutorInfo.java b/core/src/main/java/org/apache/spark/SparkExecutorInfo.java
@@ -0,0 +1,33 @@
+/*
+ * 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.io.Serializable;
+
+/**
+ * Exposes information about Spark Executors.
+ *
+ * This interface is not designed to be implemented outside of Spark.  We may add additional methods
+ * which may break binary compatibility with outside implementations.
+ */
+public interface SparkExecutorInfo extends Serializable {
+  String host();
+  int port();
+  long cacheSize();
+  int numRunningTasks();
+}
diff --git a/core/src/main/scala/org/apache/spark/SparkStatusTracker.scala b/core/src/main/scala/org/apache/spark/SparkStatusTracker.scala
@@ -108,32 +108,20 @@ class SparkStatusTracker private[spark] (sc: SparkContext) {
   }
 
   /**
-   * Returns a list of all known executors, represented by string with format: "host:port"
+   * Returns information of all known executors, including host, port, cacheSize, numRunningTasks.
    */
-  def getExecutorList(): Array[String] = {
-    sc.getExecutorStorageStatus.map { status =>
-      status.blockManagerId.hostPort
-    }
-  }
-
-  /**
-   * Returns a map contains executor id(host+port) and its cache size(memory used by caching RDDs).
-   */
-  def getCacheSizeByExecutors(): Map[String, Long] = {
-    sc.getExecutorStorageStatus.map { status =>
-      status.blockManagerId.hostPort -> status.cacheSize
-    }.toMap
-  }
-
-  /**
-   * Returns a map contains executor id(host+port) and its number of running tasks.
-   */
-  def getRunningTasksByExecutors(): Map[String, Int] = {
+  def getExecutorInfos: Array[SparkExecutorInfo] = {
     val executorIdToRunningTasks: Map[String, Int] =
       sc.taskScheduler.asInstanceOf[TaskSchedulerImpl].runningTasksByExecutors()
+
     sc.getExecutorStorageStatus.map { status =>
       val bmId = status.blockManagerId
-      bmId.hostPort -> executorIdToRunningTasks.getOrElse(bmId.executorId, 0)
-    }.toMap
+      new SparkExecutorInfoImpl(
+        bmId.host,
+        bmId.port,
+        status.cacheSize,
+        executorIdToRunningTasks.getOrElse(bmId.executorId, 0)
+      )
+    }
   }
 }
diff --git a/core/src/main/scala/org/apache/spark/StatusAPIImpl.scala b/core/src/main/scala/org/apache/spark/StatusAPIImpl.scala
@@ -18,18 +18,25 @@
 package org.apache.spark
 
 private class SparkJobInfoImpl (
-  val jobId: Int,
-  val stageIds: Array[Int],
-  val status: JobExecutionStatus)
- extends SparkJobInfo
+    val jobId: Int,
+    val stageIds: Array[Int],
+    val status: JobExecutionStatus)
+  extends SparkJobInfo
 
 private class SparkStageInfoImpl(
-  val stageId: Int,
-  val currentAttemptId: Int,
-  val submissionTime: Long,
-  val name: String,
-  val numTasks: Int,
-  val numActiveTasks: Int,
-  val numCompletedTasks: Int,
-  val numFailedTasks: Int)
- extends SparkStageInfo
+    val stageId: Int,
+    val currentAttemptId: Int,
+    val submissionTime: Long,
+    val name: String,
+    val numTasks: Int,
+    val numActiveTasks: Int,
+    val numCompletedTasks: Int,
+    val numFailedTasks: Int)
+  extends SparkStageInfo
+
+private class SparkExecutorInfoImpl(
+    val host: String,
+    val port: Int,
+    val cacheSize: Long,
+    val numRunningTasks: Int)
+  extends SparkExecutorInfo
