Skip to content

[SPARK-12108] Make event logs smaller #10115

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
wants to merge 2 commits into from
Closed

[SPARK-12108] Make event logs smaller #10115

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
5a0ebdf
Store only callsite short forms in event log
Dec 2, 2015
1ddd099
Revert unintended change
Dec 3, 2015
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
4 changes: 2 additions & 2 deletions core/src/main/scala/org/apache/spark/storage/RDDInfo.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@ class RDDInfo(
val numPartitions: Int,
var storageLevel: StorageLevel,
val parentIds: Seq[Int],
val callSite: CallSite = CallSite.empty,
val callSite: String = "",
val scope: Option[RDDOperationScope] = None)
extends Ordered[RDDInfo] {

Expand Down Expand Up @@ -58,6 +58,6 @@ private[spark] object RDDInfo {
val rddName = Option(rdd.name).getOrElse(Utils.getFormattedClassName(rdd))
val parentIds = rdd.dependencies.map(_.rdd.id)
new RDDInfo(rdd.id, rddName, rdd.partitions.length,
rdd.getStorageLevel, parentIds, rdd.creationSite, rdd.scope)
rdd.getStorageLevel, parentIds, rdd.creationSite.shortForm, rdd.scope)
}
}
4 changes: 2 additions & 2 deletions core/src/main/scala/org/apache/spark/ui/scope/RDDOperationGraph.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,7 @@ private[ui] case class RDDOperationGraph(
rootCluster: RDDOperationCluster)

/** A node in an RDDOperationGraph. This represents an RDD. */
private[ui] case class RDDOperationNode(id: Int, name: String, cached: Boolean, callsite: CallSite)
private[ui] case class RDDOperationNode(id: Int, name: String, cached: Boolean, callsite: String)

/**
* A directed edge connecting two nodes in an RDDOperationGraph.
Expand Down Expand Up @@ -178,7 +178,7 @@ private[ui] object RDDOperationGraph extends Logging {

/** Return the dot representation of a node in an RDDOperationGraph. */
private def makeDotNode(node: RDDOperationNode): String = {
val label = s"${node.name} [${node.id}]\n${node.callsite.shortForm}"
val label = s"${node.name} [${node.id}]\n${node.callsite}"
s"""${node.id} [label="$label"]"""
}

Expand Down
17 changes: 2 additions & 15 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 @@ -398,7 +398,7 @@ private[spark] object JsonProtocol {
("RDD ID" -> rddInfo.id) ~
("Name" -> rddInfo.name) ~
("Scope" -> rddInfo.scope.map(_.toJson)) ~
("Callsite" -> callsiteToJson(rddInfo.callSite)) ~
("Callsite" -> rddInfo.callSite) ~
("Parent IDs" -> parentIds) ~
("Storage Level" -> storageLevel) ~
("Number of Partitions" -> rddInfo.numPartitions) ~
Expand All @@ -408,11 +408,6 @@ private[spark] object JsonProtocol {
("Disk Size" -> rddInfo.diskSize)
}

def callsiteToJson(callsite: CallSite): JValue = {
("Short Form" -> callsite.shortForm) ~
("Long Form" -> callsite.longForm)
}

def storageLevelToJson(storageLevel: StorageLevel): JValue = {
("Use Disk" -> storageLevel.useDisk) ~
("Use Memory" -> storageLevel.useMemory) ~
Expand Down Expand Up @@ -857,9 +852,7 @@ private[spark] object JsonProtocol {
val scope = Utils.jsonOption(json \ "Scope")
.map(_.extract[String])
.map(RDDOperationScope.fromJson)
val callsite = Utils.jsonOption(json \ "Callsite")
.map(callsiteFromJson)
.getOrElse(CallSite.empty)
val callsite = Utils.jsonOption(json \ "Callsite").map(_.extract[String]).getOrElse("")
val parentIds = Utils.jsonOption(json \ "Parent IDs")
.map { l => l.extract[List[JValue]].map(_.extract[Int]) }
.getOrElse(Seq.empty)
Expand All @@ -880,12 +873,6 @@ private[spark] object JsonProtocol {
rddInfo
}

def callsiteFromJson(json: JValue): CallSite = {
val shortForm = (json \ "Short Form").extract[String]
val longForm = (json \ "Long Form").extract[String]
CallSite(shortForm, longForm)
}

def storageLevelFromJson(json: JValue): StorageLevel = {
val useDisk = (json \ "Use Disk").extract[Boolean]
val useMemory = (json \ "Use Memory").extract[Boolean]
Expand Down
35 changes: 14 additions & 21 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 @@ -111,7 +111,6 @@ class JsonProtocolSuite extends SparkFunSuite {
test("Dependent Classes") {
val logUrlMap = Map("stderr" -> "mystderr", "stdout" -> "mystdout").toMap
testRDDInfo(makeRddInfo(2, 3, 4, 5L, 6L))
testCallsite(CallSite("happy", "birthday"))
testStageInfo(makeStageInfo(10, 20, 30, 40L, 50L))
testTaskInfo(makeTaskInfo(999L, 888, 55, 777L, false))
testTaskMetrics(makeTaskMetrics(
Expand Down Expand Up @@ -343,13 +342,13 @@ class JsonProtocolSuite extends SparkFunSuite {
// "Scope" and "Parent IDs" were introduced in Spark 1.4.0
// "Callsite" was introduced in Spark 1.6.0
val rddInfo = new RDDInfo(1, "one", 100, StorageLevel.NONE, Seq(1, 6, 8),
CallSite("short", "long"), Some(new RDDOperationScope("fable")))
"callsite", Some(new RDDOperationScope("fable")))
val oldRddInfoJson = JsonProtocol.rddInfoToJson(rddInfo)
.removeField({ _._1 == "Parent IDs"})
.removeField({ _._1 == "Scope"})
.removeField({ _._1 == "Callsite"})
val expectedRddInfo = new RDDInfo(
1, "one", 100, StorageLevel.NONE, Seq.empty, CallSite.empty, scope = None)
1, "one", 100, StorageLevel.NONE, Seq.empty, "", scope = None)
assertEquals(expectedRddInfo, JsonProtocol.rddInfoFromJson(oldRddInfoJson))
}

Expand Down Expand Up @@ -397,11 +396,6 @@ class JsonProtocolSuite extends SparkFunSuite {
assertEquals(info, newInfo)
}

private def testCallsite(callsite: CallSite): Unit = {
val newCallsite = JsonProtocol.callsiteFromJson(JsonProtocol.callsiteToJson(callsite))
assert(callsite === newCallsite)
}

private def testStageInfo(info: StageInfo) {
val newInfo = JsonProtocol.stageInfoFromJson(JsonProtocol.stageInfoToJson(info))
assertEquals(info, newInfo)
Expand Down Expand Up @@ -726,8 +720,7 @@ class JsonProtocolSuite extends SparkFunSuite {
}

private def makeRddInfo(a: Int, b: Int, c: Int, d: Long, e: Long) = {
val r = new RDDInfo(a, "mayor", b, StorageLevel.MEMORY_AND_DISK,
Seq(1, 4, 7), CallSite(a.toString, b.toString))
val r = new RDDInfo(a, "mayor", b, StorageLevel.MEMORY_AND_DISK, Seq(1, 4, 7), a.toString)
r.numCachedPartitions = c
r.memSize = d
r.diskSize = e
Expand Down Expand Up @@ -870,7 +863,7 @@ class JsonProtocolSuite extends SparkFunSuite {
| {
| "RDD ID": 101,
| "Name": "mayor",
| "Callsite": {"Short Form": "101", "Long Form": "201"},
| "Callsite": "101",
| "Parent IDs": [1, 4, 7],
| "Storage Level": {
| "Use Disk": true,
Expand Down Expand Up @@ -1273,7 +1266,7 @@ class JsonProtocolSuite extends SparkFunSuite {
| {
| "RDD ID": 1,
| "Name": "mayor",
| "Callsite": {"Short Form": "1", "Long Form": "200"},
| "Callsite": "1",
| "Parent IDs": [1, 4, 7],
| "Storage Level": {
| "Use Disk": true,
Expand Down Expand Up @@ -1317,7 +1310,7 @@ class JsonProtocolSuite extends SparkFunSuite {
| {
| "RDD ID": 2,
| "Name": "mayor",
| "Callsite": {"Short Form": "2", "Long Form": "400"},
| "Callsite": "2",
| "Parent IDs": [1, 4, 7],
| "Storage Level": {
| "Use Disk": true,
Expand All @@ -1335,7 +1328,7 @@ class JsonProtocolSuite extends SparkFunSuite {
| {
| "RDD ID": 3,
| "Name": "mayor",
| "Callsite": {"Short Form": "3", "Long Form": "401"},
| "Callsite": "3",
| "Parent IDs": [1, 4, 7],
| "Storage Level": {
| "Use Disk": true,
Expand Down Expand Up @@ -1379,7 +1372,7 @@ class JsonProtocolSuite extends SparkFunSuite {
| {
| "RDD ID": 3,
| "Name": "mayor",
| "Callsite": {"Short Form": "3", "Long Form": "600"},
| "Callsite": "3",
| "Parent IDs": [1, 4, 7],
| "Storage Level": {
| "Use Disk": true,
Expand All @@ -1397,7 +1390,7 @@ class JsonProtocolSuite extends SparkFunSuite {
| {
| "RDD ID": 4,
| "Name": "mayor",
| "Callsite": {"Short Form": "4", "Long Form": "601"},
| "Callsite": "4",
| "Parent IDs": [1, 4, 7],
| "Storage Level": {
| "Use Disk": true,
Expand All @@ -1415,7 +1408,7 @@ class JsonProtocolSuite extends SparkFunSuite {
| {
| "RDD ID": 5,
| "Name": "mayor",
| "Callsite": {"Short Form": "5", "Long Form": "602"},
| "Callsite": "5",
| "Parent IDs": [1, 4, 7],
| "Storage Level": {
| "Use Disk": true,
Expand Down Expand Up @@ -1459,7 +1452,7 @@ class JsonProtocolSuite extends SparkFunSuite {
| {
| "RDD ID": 4,
| "Name": "mayor",
| "Callsite": {"Short Form": "4", "Long Form": "800"},
| "Callsite": "4",
| "Parent IDs": [1, 4, 7],
| "Storage Level": {
| "Use Disk": true,
Expand All @@ -1477,7 +1470,7 @@ class JsonProtocolSuite extends SparkFunSuite {
| {
| "RDD ID": 5,
| "Name": "mayor",
| "Callsite": {"Short Form": "5", "Long Form": "801"},
| "Callsite": "5",
| "Parent IDs": [1, 4, 7],
| "Storage Level": {
| "Use Disk": true,
Expand All @@ -1495,7 +1488,7 @@ class JsonProtocolSuite extends SparkFunSuite {
| {
| "RDD ID": 6,
| "Name": "mayor",
| "Callsite": {"Short Form": "6", "Long Form": "802"},
| "Callsite": "6",
| "Parent IDs": [1, 4, 7],
| "Storage Level": {
| "Use Disk": true,
Expand All @@ -1513,7 +1506,7 @@ class JsonProtocolSuite extends SparkFunSuite {
| {
| "RDD ID": 7,
| "Name": "mayor",
| "Callsite": {"Short Form": "7", "Long Form": "803"},
| "Callsite": "7",
| "Parent IDs": [1, 4, 7],
| "Storage Level": {
| "Use Disk": true,
Expand Down