Skip to content

Commit a39fc87

Browse files
JoshRosenJoshRosen
committed
[SPARK-39636][CORE][UI] Fix multiple bugs in JsonProtocol, impacting off heap StorageLevels and Task/Executor ResourceRequests
### What changes were proposed in this pull request? This PR fixes three longstanding bugs in Spark's `JsonProtocol`: - `TaskResourceRequest` loses precision for `amount` < 0.5. The `amount` is a floating point number which is either between 0 and 0.5 or is a positive integer, but the JSON read path assumes it is an integer. - `ExecutorResourceRequest` integer overflows for values larger than Int.MaxValue because the write path writes longs but the read path assumes integers. - Off heap StorageLevels are not handled properly: the `useOffHeap` field isn't included in the JSON, so this StorageLevel cannot be round-tripped through JSON. This could cause the History Server to display inaccurate "off heap memory used" stats on the executors page. I discovered these bugs while working on #36885. ### Why are the changes needed? JsonProtocol should be able to roundtrip events through JSON without loss of information. ### Does this PR introduce _any_ user-facing change? Yes: it fixes bugs that impact information shown in the History Server Web UI. The new StorageLevel JSON field will be visible to tools which process raw event log JSON. ### How was this patch tested? Updated existing unit tests to cover the changed logic. Closes #37027 from JoshRosen/jsonprotocol-bugfixes. Authored-by: Josh Rosen <[email protected]> Signed-off-by: Josh Rosen <[email protected]>
1 parent 91f95a7 commit a39fc87

File tree

2 files changed

+64
-6
lines changed

2 files changed

+64
-6
lines changed

‎core/src/main/scala/org/apache/spark/util/JsonProtocol.scala‎

Lines changed: 14 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -512,6 +512,7 @@ private[spark] object JsonProtocol {
512512
def storageLevelToJson(storageLevel: StorageLevel): JValue = {
513513
("Use Disk" -> storageLevel.useDisk) ~
514514
("Use Memory" -> storageLevel.useMemory) ~
515+
("Use Off Heap" -> storageLevel.useOffHeap) ~
515516
("Deserialized" -> storageLevel.deserialized) ~
516517
("Replication" -> storageLevel.replication)
517518
}
@@ -750,15 +751,15 @@ private[spark] object JsonProtocol {
750751

751752
def executorResourceRequestFromJson(json: JValue): ExecutorResourceRequest = {
752753
val rName = (json \ "Resource Name").extract[String]
753-
val amount = (json \ "Amount").extract[Int]
754+
val amount = (json \ "Amount").extract[Long]
754755
val discoveryScript = (json \ "Discovery Script").extract[String]
755756
val vendor = (json \ "Vendor").extract[String]
756757
new ExecutorResourceRequest(rName, amount, discoveryScript, vendor)
757758
}
758759

759760
def taskResourceRequestFromJson(json: JValue): TaskResourceRequest = {
760761
val rName = (json \ "Resource Name").extract[String]
761-
val amount = (json \ "Amount").extract[Int]
762+
val amount = (json \ "Amount").extract[Double]
762763
new TaskResourceRequest(rName, amount)
763764
}
764765

@@ -1202,9 +1203,19 @@ private[spark] object JsonProtocol {
12021203
def storageLevelFromJson(json: JValue): StorageLevel = {
12031204
val useDisk = (json \ "Use Disk").extract[Boolean]
12041205
val useMemory = (json \ "Use Memory").extract[Boolean]
1206+
// The "Use Off Heap" field was added in Spark 3.4.0
1207+
val useOffHeap = jsonOption(json \ "Use Off Heap") match {
1208+
case Some(value) => value.extract[Boolean]
1209+
case None => false
1210+
}
12051211
val deserialized = (json \ "Deserialized").extract[Boolean]
12061212
val replication = (json \ "Replication").extract[Int]
1207-
StorageLevel(useDisk, useMemory, deserialized, replication)
1213+
StorageLevel(
1214+
useDisk = useDisk,
1215+
useMemory = useMemory,
1216+
useOffHeap = useOffHeap,
1217+
deserialized = deserialized,
1218+
replication = replication)
12081219
}
12091220

12101221
def blockStatusFromJson(json: JValue): BlockStatus = {

‎core/src/test/scala/org/apache/spark/util/JsonProtocolSuite.scala‎

Lines changed: 50 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -136,9 +136,14 @@ class JsonProtocolSuite extends SparkFunSuite {
136136
321L, 654L, 765L, 256912L, 123456L, 123456L, 61728L,
137137
30364L, 15182L, 10L, 90L, 2L, 20L, 80001L)))
138138
val rprofBuilder = new ResourceProfileBuilder()
139-
val taskReq = new TaskResourceRequests().cpus(1).resource("gpu", 1)
140-
val execReq =
141-
new ExecutorResourceRequests().cores(2).resource("gpu", 2, "myscript")
139+
val taskReq = new TaskResourceRequests()
140+
.cpus(1)
141+
.resource("gpu", 1)
142+
.resource("fgpa", 0.5)
143+
val execReq: ExecutorResourceRequests = new ExecutorResourceRequests()
144+
.cores(2)
145+
.resource("gpu", 2, "myscript")
146+
.resource("myCustomResource", amount = Int.MaxValue + 1L, discoveryScript = "myscript2")
142147
rprofBuilder.require(taskReq).require(execReq)
143148
val resourceProfile = rprofBuilder.build
144149
resourceProfile.setResourceProfileId(21)
@@ -203,6 +208,7 @@ class JsonProtocolSuite extends SparkFunSuite {
203208
testStorageLevel(StorageLevel.MEMORY_AND_DISK_2)
204209
testStorageLevel(StorageLevel.MEMORY_AND_DISK_SER)
205210
testStorageLevel(StorageLevel.MEMORY_AND_DISK_SER_2)
211+
testStorageLevel(StorageLevel.OFF_HEAP)
206212

207213
// JobResult
208214
val exception = new Exception("Out of Memory! Please restock film.")
@@ -319,6 +325,21 @@ class JsonProtocolSuite extends SparkFunSuite {
319325
val newMetrics = JsonProtocol.taskMetricsFromJson(oldJson)
320326
}
321327

328+
test("StorageLevel backward compatibility") {
329+
// "Use Off Heap" was added in Spark 3.4.0
330+
val level = StorageLevel(
331+
useDisk = false,
332+
useMemory = true,
333+
useOffHeap = true,
334+
deserialized = false,
335+
replication = 1
336+
)
337+
val newJson = JsonProtocol.storageLevelToJson(level)
338+
val oldJson = newJson.removeField { case (field, _) => field == "Use Off Heap" }
339+
val newLevel = JsonProtocol.storageLevelFromJson(oldJson)
340+
assert(newLevel.useOffHeap === false)
341+
}
342+
322343
test("BlockManager events backward compatibility") {
323344
// SparkListenerBlockManagerAdded/Removed in Spark 1.0.0 do not have a "time" property.
324345
val blockManagerAdded = SparkListenerBlockManagerAdded(1L,
@@ -1189,6 +1210,7 @@ private[spark] object JsonProtocolSuite extends Assertions {
11891210
| "Storage Level": {
11901211
| "Use Disk": true,
11911212
| "Use Memory": true,
1213+
| "Use Off Heap": false,
11921214
| "Deserialized": true,
11931215
| "Replication": 1
11941216
| },
@@ -1437,6 +1459,7 @@ private[spark] object JsonProtocolSuite extends Assertions {
14371459
| "Storage Level": {
14381460
| "Use Disk": true,
14391461
| "Use Memory": true,
1462+
| "Use Off Heap": false,
14401463
| "Deserialized": false,
14411464
| "Replication": 2
14421465
| },
@@ -1563,6 +1586,7 @@ private[spark] object JsonProtocolSuite extends Assertions {
15631586
| "Storage Level": {
15641587
| "Use Disk": true,
15651588
| "Use Memory": true,
1589+
| "Use Off Heap": false,
15661590
| "Deserialized": false,
15671591
| "Replication": 2
15681592
| },
@@ -1689,6 +1713,7 @@ private[spark] object JsonProtocolSuite extends Assertions {
16891713
| "Storage Level": {
16901714
| "Use Disk": true,
16911715
| "Use Memory": true,
1716+
| "Use Off Heap": false,
16921717
| "Deserialized": false,
16931718
| "Replication": 2
16941719
| },
@@ -1722,6 +1747,7 @@ private[spark] object JsonProtocolSuite extends Assertions {
17221747
| "Storage Level": {
17231748
| "Use Disk": true,
17241749
| "Use Memory": true,
1750+
| "Use Off Heap": false,
17251751
| "Deserialized": true,
17261752
| "Replication": 1
17271753
| },
@@ -1769,6 +1795,7 @@ private[spark] object JsonProtocolSuite extends Assertions {
17691795
| "Storage Level": {
17701796
| "Use Disk": true,
17711797
| "Use Memory": true,
1798+
| "Use Off Heap": false,
17721799
| "Deserialized": true,
17731800
| "Replication": 1
17741801
| },
@@ -1787,6 +1814,7 @@ private[spark] object JsonProtocolSuite extends Assertions {
17871814
| "Storage Level": {
17881815
| "Use Disk": true,
17891816
| "Use Memory": true,
1817+
| "Use Off Heap": false,
17901818
| "Deserialized": true,
17911819
| "Replication": 1
17921820
| },
@@ -1834,6 +1862,7 @@ private[spark] object JsonProtocolSuite extends Assertions {
18341862
| "Storage Level": {
18351863
| "Use Disk": true,
18361864
| "Use Memory": true,
1865+
| "Use Off Heap": false,
18371866
| "Deserialized": true,
18381867
| "Replication": 1
18391868
| },
@@ -1852,6 +1881,7 @@ private[spark] object JsonProtocolSuite extends Assertions {
18521881
| "Storage Level": {
18531882
| "Use Disk": true,
18541883
| "Use Memory": true,
1884+
| "Use Off Heap": false,
18551885
| "Deserialized": true,
18561886
| "Replication": 1
18571887
| },
@@ -1870,6 +1900,7 @@ private[spark] object JsonProtocolSuite extends Assertions {
18701900
| "Storage Level": {
18711901
| "Use Disk": true,
18721902
| "Use Memory": true,
1903+
| "Use Off Heap": false,
18731904
| "Deserialized": true,
18741905
| "Replication": 1
18751906
| },
@@ -1917,6 +1948,7 @@ private[spark] object JsonProtocolSuite extends Assertions {
19171948
| "Storage Level": {
19181949
| "Use Disk": true,
19191950
| "Use Memory": true,
1951+
| "Use Off Heap": false,
19201952
| "Deserialized": true,
19211953
| "Replication": 1
19221954
| },
@@ -1935,6 +1967,7 @@ private[spark] object JsonProtocolSuite extends Assertions {
19351967
| "Storage Level": {
19361968
| "Use Disk": true,
19371969
| "Use Memory": true,
1970+
| "Use Off Heap": false,
19381971
| "Deserialized": true,
19391972
| "Replication": 1
19401973
| },
@@ -1953,6 +1986,7 @@ private[spark] object JsonProtocolSuite extends Assertions {
19531986
| "Storage Level": {
19541987
| "Use Disk": true,
19551988
| "Use Memory": true,
1989+
| "Use Off Heap": false,
19561990
| "Deserialized": true,
19571991
| "Replication": 1
19581992
| },
@@ -1971,6 +2005,7 @@ private[spark] object JsonProtocolSuite extends Assertions {
19712005
| "Storage Level": {
19722006
| "Use Disk": true,
19732007
| "Use Memory": true,
2008+
| "Use Off Heap": false,
19742009
| "Deserialized": true,
19752010
| "Replication": 1
19762011
| },
@@ -2291,6 +2326,7 @@ private[spark] object JsonProtocolSuite extends Assertions {
22912326
| "Storage Level": {
22922327
| "Use Disk": true,
22932328
| "Use Memory": true,
2329+
| "Use Off Heap": false,
22942330
| "Deserialized": false,
22952331
| "Replication": 2
22962332
| },
@@ -2489,6 +2525,7 @@ private[spark] object JsonProtocolSuite extends Assertions {
24892525
| "Storage Level": {
24902526
| "Use Disk": false,
24912527
| "Use Memory": true,
2528+
| "Use Off Heap": false,
24922529
| "Deserialized": true,
24932530
| "Replication": 1
24942531
| },
@@ -2578,6 +2615,12 @@ private[spark] object JsonProtocolSuite extends Assertions {
25782615
| "Discovery Script":"",
25792616
| "Vendor":""
25802617
| },
2618+
| "myCustomResource":{
2619+
| "Resource Name":"myCustomResource",
2620+
| "Amount": 2147483648,
2621+
| "Discovery Script": "myscript2",
2622+
| "Vendor" : ""
2623+
| },
25812624
| "gpu":{
25822625
| "Resource Name":"gpu",
25832626
| "Amount":2,
@@ -2593,6 +2636,10 @@ private[spark] object JsonProtocolSuite extends Assertions {
25932636
| "gpu":{
25942637
| "Resource Name":"gpu",
25952638
| "Amount":1.0
2639+
| },
2640+
| "fgpa":{
2641+
| "Resource Name":"fgpa",
2642+
| "Amount":0.5
25962643
| }
25972644
| }
25982645
|}

0 commit comments

Comments
 (0)