[SPARK-30306][CORE][PYTHON][WIP] Instrument Python UDF execution time and throughput metrics using Spark Metrics system

[LucaCanali]

What changes were proposed in this pull request?

This proposes to extend Spark instrumentation to add metrics aimed at drilling down on the performance of Python code called by Spark: via UDF, Pandas UDF or with MapPartittions. Relevant performance counters, notably exuction time, are exposed using the Spark Metrics System (based on the Dropwizard library).

Why are the changes needed?

This allows to easily consume the metrics produced by executors, for example using a performance dashboard (this references to previous work as discucssed in https://db-blog.web.cern.ch/blog/luca-canali/2019-02-performance-dashboard-apache-spark ).
See also the screenshot that compares the existing state (no Python UDF time instrumentation) to the proposed new functionality

Does this PR introduce any user-facing change?

This PR adds the PythonMetrics source to the Spark Metrics system. The list of the implemented metrics has been added to the Monitoring documentation.

How was this patch tested?

Added relevant tests

• manually tested end-to-end on a YARN cluster and using an existing Spark dashboard extended with the metrics proposed here.

[maropu]

cc: @HyukjinKwon @viirya @ueshin

[HyukjinKwon]

I took a cursory look. My impression is that the functionality is slightly overlapped with Python profiler feature and the current implementation is too verbose. Let me take a closer look later.

[LucaCanali]

Thanks @HyukjinKwon for taking time for this.
This functionality propoesed in ths PR is different from Python profiler, or at least its intended use is. Also it is intended to be lightweight so that it can be used for measuring Spark workloads in production as part of a performance dashboard based on the metrics coming from the Spark metrics system.

I'd like to add some additional context on how we intend to use this.

• I have described how we implement and use a Spark performance dashboard based on the metrics system in a recent Spark Summit presentation https://databricks.com/session_eu19/performance-troubleshooting-using-apache-spark-metrics and in a blog entry http://db-blog.web.cern.ch/blog/luca-canali/2019-02-performance-dashboard-apache-spark
• Over the last couple of years I have helped improving this by adding to the Spark metrics system instrumentation for aggregated task metrics values per executor [SPARK-25228], JVM CPU usage [SPARK-22190], memory usage [SPARK-27189]. This still does not cover all the space of possible activities and "time usage" by Spark tasks and executors, in particular the problem that I am trying to solve with this PR is that time spent waiting for results to come back from Python workers (typically when executing a UDF) is currently not instrumented, so it appears in the dashboard as run time "without attribution" in the current dashboard, while it can be visualized using the metrics implemented here (see image )

As you mentioned, in the current PR I have implemented several details, which I guess can be useful when troubleshooting, but for the general use can be simplified as you propose, in particular in the number and detials exposed in the user-visible metrics in the PythonMetrics source.
A proposal for a simplified set of metrics to expose in the PythonMetrics source is:

• BatchCountFromWorker
• BatchCountToWorker
• BytesSentToWorker
• RunAndReadTimeFromWorker
• WriteTimeToWorker
• PandasUDFSentRowCount
• PandasUDFReceivedRowCount

[LucaCanali]

This has now been streamlined down to the following metrics (under namespace=PythonMetrics):

• BytesReceivedFromWorkers
• BytesSentToWorkers
• FetchResultsTimeFromWorkers
• NumBatchesFromWorkers
• NumBatchesToWorkers
• PandasUDFReceivedNumRows
• PandasUDFSentNumRows
• WriteTimeToWorkers

[HyukjinKwon]

ok to test

[HyukjinKwon]

Have a couple of questions.

• Is it possible to merge it with SQLMetric? It would be nicer if UI shows it as well.
• Is it possible to integrate with existing Python profiler? The current read time isn't purely Python execution time. It includes socket IO time which can potentially be large.

[LucaCanali]

• I like the idea of adding SQLMetrics for Python UDF instrumentation and use them in the WEBUI. However, I think the work would rather fit for a separate JIRA/PR. The implementation details and the overhead of SQLMetrics are different from Dropwizard-based metrics, so probably we would like to have only a limited number of SQLMetrics instrumenting task activities in this area. Also the implementation of SQLMetrics for [[PythonUDF]] execution may require some important changes to the current plan evaluation code.

• It is indeed the case that the “read time from worker” which is exposed to the users via the dropwizard library as “FetchResultsTimeFromWorkers” contains both socket I/O + deserialization time and Python UDF execution time. Measuring on the Python side could allow to separate the 2 time components, however currently I don’t see how to make a lightweight implementation for that. Python profiler has the possibility to measure on the Python side as you mentioned, but I see its usage more for debugging, while the proposed instrumentation is lightweight and intended to be used for production use cases too. Maybe future work can address this case if there is need?

[LucaCanali]

@HyukjinKwon I have finally managed to work on your suggestion to instrument Python execution using SQL Metrics, so that users can see the metrics via the WebUI. See [SPARK-34265]. I imagine that I could later refactor the work on this PR based on that.

[maropu]

PR link: #31367

[SparkQA]

Test build #116893 has finished for PR 26953 at commit 2d70042.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #117475 has finished for PR 26953 at commit af34f49.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #118782 has finished for PR 26953 at commit fd76071.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[LucaCanali]

retest this please

[BryanCutler]

retest this please

[SparkQA]

Test build #118929 has finished for PR 26953 at commit fd76071.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #119372 has finished for PR 26953 at commit 989dba1.

• This patch fails due to an unknown error code, -9.
• This patch merges cleanly.
• This patch adds no public classes.

[LucaCanali]

It looks like a case of flaky test. Can we test this again please?

[LucaCanali]

I would not worry very much about the performance impact of this additional instrumentation, as it hooks on something that is not very fast already, that is the serialization/deserialization JVM-Python. Moreover, the instrumentation mostly just takes timing values and does so per batch of serialized rows, so the impach on the total throughput is expected to be further reduced by this. So far, I have only tested this manually and did not observe any particular impact. If we have a Python UDF benchmark I could further test with that.

[SparkQA]

Test build #123188 has finished for PR 26953 at commit a39ad63.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #126228 has finished for PR 26953 at commit 277245c.

• This patch fails due to an unknown error code, -9.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #128116 has finished for PR 26953 at commit 2fd3f1b.

• This patch fails Python style tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #128117 has finished for PR 26953 at commit c391be0.

• This patch fails Python style tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #130035 has finished for PR 26953 at commit 337b67c.

• This patch fails Python style tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/34642/

[SparkQA]

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/34642/

[SparkQA]

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/34650/

[SparkQA]

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/34650/

[SparkQA]

Test build #130045 has finished for PR 26953 at commit 4b88804.

• This patch fails Python style tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/34654/

[SparkQA]

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/34653/

[SparkQA]

Test build #130043 has finished for PR 26953 at commit c7570bd.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/34653/

[SparkQA]

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/34654/

[SparkQA]

Test build #130046 has finished for PR 26953 at commit 02a0ca0.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/35660/

[SparkQA]

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/35660/

[SparkQA]

Test build #131055 has finished for PR 26953 at commit 750b973.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[LucaCanali]

I am closing this as the implementation should be refactored, what looks like a better way to implement this is to first work on the correctponding SQL metrics as in SPARK-34265 and then revisit the instrumentation for the Spark Metrics System.