gaogaotiantian commented on code in PR #59019:
URL: https://github.com/apache/spark/pull/59019#discussion_r4127576752
##########
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:
I think `readWorkerMetrics` should return a `WorkerMetrics` instead of a
`JsonNode`. Parsing it as `json` should be internal - we can change that in the
future. It's unnecessary to split these processes. We can do all the conversion
and validations in this method and return a readable metrics directly.
##########
python/pyspark/tests/test_worker_metrics.py:
##########
@@ -0,0 +1,224 @@
+#
Review Comment:
Let's avoid creating a new test file everything we change something. This is
a habit for a lot of agents I think. If we just changed the protocol of sending
metrics between the UDF and the executor, we should not need a complete new
file to test that. Let's see if there are existing tests for it. If not, make
some new test cases in the worker test file should be good.
--
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]
