[SPARK-35799][SS] Fix the allUpdatesTimeMs metric measuring in FlatMapGroupsWithStateExec #32952
Conversation
|
Kubernetes integration test starting |
|
Kubernetes integration test status success |
|
Kubernetes integration test starting |
|
Test build #139942 has finished for PR 32952 at commit
|
|
Kubernetes integration test status success |
|
Test build #139949 has finished for PR 32952 at commit
|
|
retest this, please |
|
Kubernetes integration test starting |
|
Kubernetes integration test status failure |
…pGroupsWithStateExec
|
Kubernetes integration test starting |
|
Kubernetes integration test status failure |
|
Test build #140105 has finished for PR 32952 at commit
|
|
Jenkins passed. Thanks! Merging to master. |
|
Thanks @vkorukanti for your contribution! I merged into master. |
|
Test build #140108 has finished for PR 32952 at commit
|
…pGroupsWithStateExec ### What changes were proposed in this pull request? Fix how we measure the metric `allUpdatesTimeMs` in `FlatMapGroupsWithStateExec` similar to other streaming stateful operators. ### Why are the changes needed? Metric `allUpdatesTimeMs` meant to capture the start to end walltime of the operator `FlatMapGroupsWithStateExec`, but currently it just [captures](https://github.com/apache/spark/blob/79362c4efcb6bd4b575438330a14a6191cca5e4b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FlatMapGroupsWithStateExec.scala#L121) the iterator creation time. Fix it to measure similar to how other stateful operators measure. Example one [here](https://github.com/apache/spark/blob/79362c4efcb6bd4b575438330a14a6191cca5e4b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/statefulOperators.scala#L406). This measurement is not perfect due to the nature of the lazy iterator and also includes the time the consumer operator spent in processing the current operator output, but it should give a good signal when comparing the metric in one microbatch to the metric in another microbatch. ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? Existing UTs for regression. Due to the nature of metric type (time), it is hard to write a UT, but have manually verified. Closes apache#32952 from vkorukanti/SPARK-35799. Authored-by: Venki Korukanti <venki.korukanti@gmail.com> Signed-off-by: Jungtaek Lim <kabhwan.opensource@gmail.com>
What changes were proposed in this pull request?
Fix how we measure the metric
allUpdatesTimeMsinFlatMapGroupsWithStateExecsimilar to other streaming stateful operators.Why are the changes needed?
Metric
allUpdatesTimeMsmeant to capture the start to end walltime of the operatorFlatMapGroupsWithStateExec, but currently it just captures the iterator creation time.Fix it to measure similar to how other stateful operators measure. Example one here. This measurement is not perfect due to the nature of the lazy iterator and also includes the time the consumer operator spent in processing the current operator output, but it should give a good signal when comparing the metric in one microbatch to the metric in another microbatch.
Does this PR introduce any user-facing change?
No
How was this patch tested?
Existing UTs for regression. Due to the nature of metric type (time), it is hard to write a UT, but have manually verified.