forked from apache/spark
-
Notifications
You must be signed in to change notification settings - Fork 2
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add a gateway SparkListener to simplify event logging
Instead of having each SparkListener log an independent set of events, centralize event logging to avoid differentiating events across UI's and thus duplicating logged events. Also rename the "fromDisk" parameter to "live". TODO: Storage page currently still relies on the previous SparkContext and is not rendering correctly.
- Loading branch information
1 parent
904c729
commit 4273013
Showing
15 changed files
with
315 additions
and
229 deletions.
There are no files selected for viewing
This file contains 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 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 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
152 changes: 152 additions & 0 deletions
152
core/src/main/scala/org/apache/spark/ui/UISparkListener.scala
This file contains 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,152 @@ | ||
| /* | ||
| * 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.ui | ||
|
|
||
| import scala.collection.mutable.ArrayBuffer | ||
|
|
||
| import org.apache.spark.scheduler._ | ||
| import org.apache.spark.scheduler.SparkListenerLoadEnvironment | ||
| import org.apache.spark.scheduler.SparkListenerJobStart | ||
| import org.apache.spark.SparkContext | ||
| import org.apache.spark.storage.StorageStatus | ||
| import org.apache.spark.util.FileLogger | ||
|
|
||
| import net.liftweb.json.JsonAST._ | ||
|
|
||
| private[spark] case class UISparkListener(gateway: GatewayUISparkListener) extends SparkListener { | ||
| // Register with gateway listener | ||
| gateway.registerSparkListener(this) | ||
| } | ||
|
|
||
| /** | ||
| * A SparkListener that serves as a gateway for all events posted to the UI. | ||
| * | ||
| * GatewayUISparkListener achieves two functions: | ||
| * | ||
| * (1) If the UI is live, GatewayUISparkListener posts each event to all attached listeners | ||
| * then logs it as JSON. This centralizes event logging and avoids having all attached | ||
| * listeners log the events on their own. By default, GatewayUISparkListener logs one | ||
| * file per job, though this needs not be the case. | ||
| * | ||
| * (2) If the UI is rendered from disk, GatewayUISparkListener replays each event deserialized | ||
| * from the event logs to all attached listeners. | ||
| */ | ||
| private[spark] class GatewayUISparkListener(live: Boolean) extends SparkListener { | ||
|
|
||
| // Log events only if the UI is live | ||
| private val logger: Option[FileLogger] = if (live) Some(new FileLogger()) else None | ||
|
|
||
| // Children listeners for which this gateway is responsible | ||
| private val listeners = ArrayBuffer[UISparkListener]() | ||
|
|
||
| def registerSparkListener(listener: UISparkListener) = { | ||
| listeners += listener | ||
| } | ||
|
|
||
| /** Log the event as JSON */ | ||
| private def logEvent(event: SparkListenerEvent) { | ||
| logger.foreach(_.logLine(compactRender(event.toJson))) | ||
| } | ||
|
|
||
| override def onStageSubmitted(stageSubmitted: SparkListenerStageSubmitted) { | ||
| listeners.foreach(_.onStageSubmitted(stageSubmitted)) | ||
| logEvent(stageSubmitted) | ||
| logger.foreach(_.flush()) | ||
| } | ||
|
|
||
| override def onStageCompleted(stageCompleted: SparkListenerStageCompleted) { | ||
| listeners.foreach(_.onStageCompleted(stageCompleted)) | ||
| logEvent(stageCompleted) | ||
| logger.foreach(_.flush()) | ||
| } | ||
|
|
||
| override def onTaskStart(taskStart: SparkListenerTaskStart) { | ||
| listeners.foreach(_.onTaskStart(taskStart)) | ||
| logEvent(taskStart) | ||
| } | ||
| override def onTaskGettingResult(taskGettingResult: SparkListenerTaskGettingResult) { | ||
| listeners.foreach(_.onTaskGettingResult(taskGettingResult)) | ||
| logEvent(taskGettingResult) | ||
| } | ||
| override def onTaskEnd(taskEnd: SparkListenerTaskEnd) { | ||
| listeners.foreach(_.onTaskEnd(taskEnd)) | ||
| logEvent(taskEnd) | ||
| } | ||
|
|
||
| override def onJobStart(jobStart: SparkListenerJobStart) { | ||
| listeners.foreach(_.onJobStart(jobStart)) | ||
| logger.foreach(_.start()) | ||
| logEvent(jobStart) | ||
| } | ||
|
|
||
| override def onJobEnd(jobEnd: SparkListenerJobEnd) { | ||
| listeners.foreach(_.onJobEnd(jobEnd)) | ||
| logEvent(jobEnd) | ||
| logger.foreach(_.close()) | ||
| } | ||
|
|
||
| override def onLoadEnvironment(loadEnvironment: SparkListenerLoadEnvironment) { | ||
| listeners.foreach(_.onLoadEnvironment(loadEnvironment)) | ||
| logEvent(loadEnvironment) | ||
| logger.foreach(_.flush()) | ||
| } | ||
|
|
||
| override def onStorageStatusFetch(storageStatusFetch: SparkListenerStorageStatusFetch) { | ||
| listeners.foreach(_.onStorageStatusFetch(storageStatusFetch)) | ||
| logEvent(storageStatusFetch) | ||
| logger.foreach(_.flush()) | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * A SparkListener that fetches storage information from SparkEnv. | ||
| * | ||
| * The frequency at which this occurs is by default every time a stage event is triggered. | ||
| * This needs not be the case, however; a stage can be arbitrarily long, so any failure | ||
| * in the middle of a stage causes the storage status for that stage to be lost. | ||
| */ | ||
| private[spark] class StorageStatusFetchSparkListener( | ||
| sc: SparkContext, | ||
| gateway: GatewayUISparkListener, | ||
| live: Boolean) | ||
| extends UISparkListener(gateway) { | ||
| var storageStatusList: Seq[StorageStatus] = sc.getExecutorStorageStatus | ||
|
|
||
| /** | ||
| * Fetch storage information from SparkEnv, which involves a query to the driver. This is | ||
| * expensive and should be invoked sparingly. | ||
| */ | ||
| def fetchStorageStatus() { | ||
| if (live) { | ||
| // Fetch only this is a live UI | ||
| val storageStatus = sc.getExecutorStorageStatus | ||
| val event = new SparkListenerStorageStatusFetch(storageStatus) | ||
| gateway.onStorageStatusFetch(event) | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Update local state with fetch result, and log the appropriate event | ||
| */ | ||
| override def onStorageStatusFetch(storageStatusFetch: SparkListenerStorageStatusFetch) { | ||
| storageStatusList = storageStatusFetch.storageStatusList | ||
| } | ||
|
|
||
| override def onStageSubmitted(stageSubmitted: SparkListenerStageSubmitted) = fetchStorageStatus() | ||
| override def onStageCompleted(stageCompleted: SparkListenerStageCompleted) = fetchStorageStatus() | ||
| } |
Oops, something went wrong.