[SPARK-16333][Core] Enable EventLoggingListener to log less #16714
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -62,20 +62,23 @@ private[spark] object JsonProtocol { | |
| * JSON serialization methods for SparkListenerEvents | | ||
| * -------------------------------------------------- */ | ||
|
|
||
| def sparkEventToJson( | ||
| event: SparkListenerEvent, | ||
| omitInternalAccums: Boolean = false, | ||
| omitUpdatedBlockStatuses: Boolean = false): JValue = { | ||
| event match { | ||
|
Contributor
There was a problem hiding this comment. stageSubmitted You didn't seem to use omitInternalAccums/omitUpdatedBlockStatuses in these cases, although you had changed the underlying methods to support these flags. Intended?
Contributor
Author
There was a problem hiding this comment. stageSubmitted/stageCompleted/jobStart should use |
||
| case stageSubmitted: SparkListenerStageSubmitted => | ||
| stageSubmittedToJson(stageSubmitted, omitInternalAccums) | ||
| case stageCompleted: SparkListenerStageCompleted => | ||
| stageCompletedToJson(stageCompleted, omitInternalAccums) | ||
| case taskStart: SparkListenerTaskStart => | ||
| taskStartToJson(taskStart, omitInternalAccums) | ||
| case taskGettingResult: SparkListenerTaskGettingResult => | ||
| taskGettingResultToJson(taskGettingResult, omitInternalAccums) | ||
| case taskEnd: SparkListenerTaskEnd => | ||
| taskEndToJson(taskEnd, omitInternalAccums, omitUpdatedBlockStatuses) | ||
| case jobStart: SparkListenerJobStart => | ||
| jobStartToJson(jobStart, omitInternalAccums) | ||
| case jobEnd: SparkListenerJobEnd => | ||
| jobEndToJson(jobEnd) | ||
| case environmentUpdate: SparkListenerEnvironmentUpdate => | ||
|
|
@@ -97,61 +100,80 @@ private[spark] object JsonProtocol { | |
| case logStart: SparkListenerLogStart => | ||
| logStartToJson(logStart) | ||
| case metricsUpdate: SparkListenerExecutorMetricsUpdate => | ||
| executorMetricsUpdateToJson(metricsUpdate, omitInternalAccums) | ||
| case blockUpdated: SparkListenerBlockUpdated => | ||
| throw new MatchError(blockUpdated) // TODO(ekl) implement this | ||
| case _ => parse(mapper.writeValueAsString(event)) | ||
| } | ||
| } | ||
|
|
||
| def stageSubmittedToJson( | ||
| stageSubmitted: SparkListenerStageSubmitted, | ||
| omitInternalAccums: Boolean = false): JValue = { | ||
| val stageInfo = stageInfoToJson(stageSubmitted.stageInfo, omitInternalAccums) | ||
| val properties = propertiesToJson(stageSubmitted.properties) | ||
| ("Event" -> SPARK_LISTENER_EVENT_FORMATTED_CLASS_NAMES.stageSubmitted) ~ | ||
| ("Stage Info" -> stageInfo) ~ | ||
| ("Properties" -> properties) | ||
| } | ||
|
|
||
| def stageCompletedToJson( | ||
| stageCompleted: SparkListenerStageCompleted, | ||
| omitInternalAccums: Boolean = false): JValue = { | ||
| val stageInfo = stageInfoToJson(stageCompleted.stageInfo, omitInternalAccums) | ||
| ("Event" -> SPARK_LISTENER_EVENT_FORMATTED_CLASS_NAMES.stageCompleted) ~ | ||
| ("Stage Info" -> stageInfo) | ||
| } | ||
|
|
||
| def taskStartToJson( | ||
| taskStart: SparkListenerTaskStart, | ||
| omitInternalAccums: Boolean = false): JValue = { | ||
| val taskInfo = taskStart.taskInfo | ||
| ("Event" -> SPARK_LISTENER_EVENT_FORMATTED_CLASS_NAMES.taskStart) ~ | ||
| ("Stage ID" -> taskStart.stageId) ~ | ||
| ("Stage Attempt ID" -> taskStart.stageAttemptId) ~ | ||
| ("Task Info" -> taskInfoToJson(taskInfo, omitInternalAccums)) | ||
| } | ||
|
|
||
| def taskGettingResultToJson( | ||
| taskGettingResult: SparkListenerTaskGettingResult, | ||
| omitInternalAccums: Boolean = false): JValue = { | ||
| val taskInfo = taskGettingResult.taskInfo | ||
| ("Event" -> SPARK_LISTENER_EVENT_FORMATTED_CLASS_NAMES.taskGettingResult) ~ | ||
| ("Task Info" -> taskInfoToJson(taskInfo, omitInternalAccums)) | ||
| } | ||
|
|
||
| def taskEndToJson( | ||
| taskEnd: SparkListenerTaskEnd, | ||
| omitInternalAccums: Boolean = false, | ||
| omitUpdatedBlockStatuses: Boolean = false): JValue = { | ||
| val taskEndReason = taskEndReasonToJson(taskEnd.reason) | ||
| val taskInfo = taskEnd.taskInfo | ||
| val taskMetrics = taskEnd.taskMetrics | ||
| val taskMetricsJson = if (taskMetrics != null) { | ||
| taskMetricsToJson(taskMetrics, omitUpdatedBlockStatuses) | ||
| } else { | ||
| JNothing | ||
| } | ||
| ("Event" -> SPARK_LISTENER_EVENT_FORMATTED_CLASS_NAMES.taskEnd) ~ | ||
| ("Stage ID" -> taskEnd.stageId) ~ | ||
| ("Stage Attempt ID" -> taskEnd.stageAttemptId) ~ | ||
| ("Task Type" -> taskEnd.taskType) ~ | ||
| ("Task End Reason" -> taskEndReason) ~ | ||
| ("Task Info" -> taskInfoToJson(taskInfo, omitInternalAccums)) ~ | ||
| ("Task Metrics" -> taskMetricsJson) | ||
| } | ||
|
|
||
| def jobStartToJson( | ||
| jobStart: SparkListenerJobStart, | ||
| omitInternalAccums: Boolean = false | ||
| ): JValue = { | ||
| val properties = propertiesToJson(jobStart.properties) | ||
| ("Event" -> SPARK_LISTENER_EVENT_FORMATTED_CLASS_NAMES.jobStart) ~ | ||
| ("Job ID" -> jobStart.jobId) ~ | ||
| ("Submission Time" -> jobStart.time) ~ | ||
| ("Stage Infos" -> | ||
| jobStart.stageInfos.map(stageInfoToJson(_, omitInternalAccums))) ~ // Added in Spark 1.2.0 | ||
| ("Stage IDs" -> jobStart.stageIds) ~ | ||
| ("Properties" -> properties) | ||
| } | ||
|
|
@@ -231,7 +253,10 @@ private[spark] object JsonProtocol { | |
| ("Spark Version" -> SPARK_VERSION) | ||
| } | ||
|
|
||
| def executorMetricsUpdateToJson( | ||
| metricsUpdate: SparkListenerExecutorMetricsUpdate, | ||
| omitInternalAccums: Boolean = false | ||
| ): JValue = { | ||
| val execId = metricsUpdate.execId | ||
| val accumUpdates = metricsUpdate.accumUpdates | ||
| ("Event" -> SPARK_LISTENER_EVENT_FORMATTED_CLASS_NAMES.metricsUpdate) ~ | ||
|
|
@@ -240,15 +265,15 @@ private[spark] object JsonProtocol { | |
| ("Task ID" -> taskId) ~ | ||
| ("Stage ID" -> stageId) ~ | ||
| ("Stage Attempt ID" -> stageAttemptId) ~ | ||
| ("Accumulator Updates" -> JArray(accumulablesToJson(updates, omitInternalAccums))) | ||
| }) | ||
| } | ||
|
|
||
| /** ------------------------------------------------------------------- * | ||
| * JSON serialization methods for classes SparkListenerEvents depend on | | ||
| * -------------------------------------------------------------------- */ | ||
|
|
||
| def stageInfoToJson(stageInfo: StageInfo, omitInternalAccums: Boolean = false): JValue = { | ||
| val rddInfo = JArray(stageInfo.rddInfos.map(rddInfoToJson).toList) | ||
| val parentIds = JArray(stageInfo.parentIds.map(JInt(_)).toList) | ||
| val submissionTime = stageInfo.submissionTime.map(JInt(_)).getOrElse(JNothing) | ||
|
|
@@ -265,10 +290,10 @@ private[spark] object JsonProtocol { | |
| ("Completion Time" -> completionTime) ~ | ||
| ("Failure Reason" -> failureReason) ~ | ||
| ("Accumulables" -> JArray( | ||
| accumulablesToJson(stageInfo.accumulables.values, omitInternalAccums))) | ||
| } | ||
|
|
||
| def taskInfoToJson(taskInfo: TaskInfo, omitInternalAccums: Boolean = false): JValue = { | ||
| ("Task ID" -> taskInfo.taskId) ~ | ||
| ("Index" -> taskInfo.index) ~ | ||
| ("Attempt" -> taskInfo.attemptNumber) ~ | ||
|
|
@@ -281,7 +306,13 @@ private[spark] object JsonProtocol { | |
| ("Finish Time" -> taskInfo.finishTime) ~ | ||
| ("Failed" -> taskInfo.failed) ~ | ||
| ("Killed" -> taskInfo.killed) ~ | ||
| ("Accumulables" -> JArray(accumulablesToJson(taskInfo.accumulables, omitInternalAccums))) | ||
| } | ||
|
|
||
| def accumulablesToJson( | ||
| accumulables: Iterable[AccumulableInfo], | ||
| omitInternalAccums: Boolean = false): List[JValue] = { | ||
| accumulables.filter(p => !omitInternalAccums || !p.internal).map(accumulableInfoToJson).toList | ||
| } | ||
|
|
||
| def accumulableInfoToJson(accumulableInfo: AccumulableInfo): JValue = { | ||
|
|
@@ -324,7 +355,9 @@ private[spark] object JsonProtocol { | |
| } | ||
| } | ||
|
|
||
| def taskMetricsToJson( | ||
| taskMetrics: TaskMetrics, | ||
| omitUpdatedBlockStatuses: Boolean = false): JValue = { | ||
| val shuffleReadMetrics: JValue = | ||
| ("Remote Blocks Fetched" -> taskMetrics.shuffleReadMetrics.remoteBlocksFetched) ~ | ||
| ("Local Blocks Fetched" -> taskMetrics.shuffleReadMetrics.localBlocksFetched) ~ | ||
|
|
@@ -343,10 +376,14 @@ private[spark] object JsonProtocol { | |
| ("Bytes Written" -> taskMetrics.outputMetrics.bytesWritten) ~ | ||
| ("Records Written" -> taskMetrics.outputMetrics.recordsWritten) | ||
| val updatedBlocks = | ||
| if (omitUpdatedBlockStatuses) { | ||
|
Contributor
Author
There was a problem hiding this comment. @vanzin @ajbozarth #17412 gets rid of updated block statuses from the accumulable but not from task metrics. If you think it's ok to not to have an option to get rid of updated block statuses, then I can just get rid of updated block statuses here. |
||
| JNothing | ||
| } else { | ||
| JArray(taskMetrics.updatedBlockStatuses.toList.map { case (id, status) => | ||
| ("Block ID" -> id.toString) ~ | ||
| ("Status" -> blockStatusToJson(status)) | ||
| }) | ||
| } | ||
| ("Executor Deserialize Time" -> taskMetrics.executorDeserializeTime) ~ | ||
| ("Executor Deserialize CPU Time" -> taskMetrics.executorDeserializeCpuTime) ~ | ||
| ("Executor Run Time" -> taskMetrics.executorRunTime) ~ | ||
|
|
||
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
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.
Is this information useful at all in event logs? If there's nothing that uses it, then not writing it is probably better than having yet another option that has to be documented and that people have to explicitly enable.
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.
I don't think this information is used to reconstruct job UI. I am not sure how this information got included in event logs, but I think some people might be using it to get internal metrics for a stage from the history server using its REST API. For example, CPU time metrics is not included in stage metrics you can get by querying history server endpoint
/applications/[app-id]/stages/[stage-id].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.
Actually I see CPU time in both stage-level data and task-level data in the REST API...
Do you mind checking the code for when this was introduced and whether it was a conscious decision (as in, it covered some user case we're not seeing)?
If possible it's always better to avoid adding more config options, especially in this kind of situation. For example, if this data is needed for something, the config would be disabling that functionality, and it would be better to instead figure out how to save it in a way that does not waste so much space. And there's always compression.
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.
@vanzin I added CPU time because back then I was pulling stage metrics from history server and needed CPU time. Here's the PR for the change: #10212 . Looking at the code, CPU time should be there, so I think there's something on my end. That's a separate problem though, and I don't think CPU time metric should increase size of event logs much. I can't think of a use case for internal accumulables then, so I think it makes sense to delete this. If anyone wants to use internal accumulables for stage metrics, they should be able to catch it after a stage finishes, not from History server.