Spenserrrr commented on code in PR #59019:
URL: https://github.com/apache/spark/pull/59019#discussion_r4127970828
##########
core/src/main/scala/org/apache/spark/api/python/PythonRunner.scala:
##########
@@ -210,6 +213,58 @@ private[spark] object BasePythonRunner extends Logging {
}
}
+ private lazy val workerMetricsMapper = new ObjectMapper()
+ .enable(JsonParser.Feature.STRICT_DUPLICATE_DETECTION)
+ .enable(DeserializationFeature.FAIL_ON_TRAILING_TOKENS)
+ .enable(DeserializationFeature.USE_BIG_DECIMAL_FOR_FLOATS)
+
+ /** Read one JSON object after METRICS_DATA, leaving the next section to the
caller. */
+ private[python] def readWorkerMetrics(stream: DataInputStream): JsonNode = {
Review Comment:
Updated `readWorkerMetrics` to handle JSON parsing, conversion, and
validation internally and return `WorkerMetrics` directly.
##########
python/pyspark/tests/test_worker_metrics.py:
##########
@@ -0,0 +1,224 @@
+#
Review Comment:
Moved the protocol tests into the existing `BasePythonRunnerSuite` and
`pyspark.tests.test_worker` suites and removed the two new test files. I also
removed the redundant Python subprocess test and kept the remaining cases
separate.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
