This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 898823df50bc [SPARK-50520][PYTHON] Respect timeout in
df.rdd.countApprox()
898823df50bc is described below
commit 898823df50bc6d83b529f9f62b60c7015924d411
Author: Rishav Sinha <[email protected]>
AuthorDate: Mon Jun 8 20:56:40 2026 -0700
[SPARK-50520][PYTHON] Respect timeout in df.rdd.countApprox()
### What changes were proposed in this pull request?
PySpark approximate RDD actions currently call getFinalValue() on the
PartialResult returned by Spark approximate job APIs. This introduces blocking
behavior and causes APIs such as countApprox(timeout=...),
sumApprox(timeout=...), and meanApprox(timeout=...) to wait for full job
completion instead of respecting timeout semantics. This PR updates PySpark to
use PartialResult.initialValue(), which contains the timeout-aware
approximation produced by ApproximateActionListener.awaitResul [...]
Additionally, regression tests were added and strengthened to validate:
- timeout-aware approximate behavior for both countApprox and meanApprox
using deliberately slow workloads,
- exact results when computation completes successfully,
- cleanup of background approximate jobs to avoid interference with
subsequent tests.
### Why are the changes needed?
Spark approximate actions are designed to return partial results after the
specified timeout. Scala APIs already expose this behavior through
PartialResult, but PySpark currently forces blocking completion by calling
getFinalValue(). As a result, PySpark approximate actions ignore timeout
semantics and wait for the entire job to finish, which is inconsistent with the
intended behavior of Spark's approximate execution APIs.
### Does this PR introduce _any_ user-facing change?
Yes. PySpark approximate RDD actions (countApprox, sumApprox, and
meanApprox) now correctly respect timeout semantics and may return
timeout-aware approximate results instead of blocking until full job completion.
### How was this patch tested?
- Reproduced the issue locally using a slow-running RDD workload.
- Verified timeout behavior before and after the fix.
- Verified exact results are still returned when the computation completes
within the timeout.
- Added regression tests in python/pyspark/tests/test_rdd.py.
- Ran: python/run-tests.py --testnames pyspark.tests.test_rdd
- Manually validated timeout and completed-result behavior for countApprox,
sumApprox, and meanApprox.
### Was this patch authored or co-authored using generative AI tooling?
No
Closes #56060 from rishav23/fix-spark-50520-countapprox-timeout.
Authored-by: Rishav Sinha <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit 3e7cae796325fc6b251022f6b695dc15100aa6f1)
Signed-off-by: Wenchen Fan <[email protected]>
---
python/pyspark/core/rdd.py | 6 ++++--
python/pyspark/tests/test_rdd.py | 30 ++++++++++++++++++++++++++++++
2 files changed, 34 insertions(+), 2 deletions(-)
diff --git a/python/pyspark/core/rdd.py b/python/pyspark/core/rdd.py
index 803bb6d5b882..190857bf3cec 100644
--- a/python/pyspark/core/rdd.py
+++ b/python/pyspark/core/rdd.py
@@ -4843,7 +4843,8 @@ class RDD(Generic[T_co]):
jrdd = self.mapPartitions(lambda it:
[float(sum(it))])._to_java_object_rdd()
assert self.ctx._jvm is not None
jdrdd = self.ctx._jvm.JavaDoubleRDD.fromRDD(jrdd.rdd())
- r = jdrdd.sumApprox(timeout, confidence).getFinalValue()
+ partial = jdrdd.sumApprox(timeout, confidence)
+ r = partial.initialValue()
return BoundedFloat(r.mean(), r.confidence(), r.low(), r.high())
def meanApprox(
@@ -4881,7 +4882,8 @@ class RDD(Generic[T_co]):
jrdd = self.map(float)._to_java_object_rdd()
assert self.ctx._jvm is not None
jdrdd = self.ctx._jvm.JavaDoubleRDD.fromRDD(jrdd.rdd())
- r = jdrdd.meanApprox(timeout, confidence).getFinalValue()
+ partial = jdrdd.meanApprox(timeout, confidence)
+ r = partial.initialValue()
return BoundedFloat(r.mean(), r.confidence(), r.low(), r.high())
def countApproxDistinct(self: "RDD[T]", relativeSD: float = 0.05) -> int:
diff --git a/python/pyspark/tests/test_rdd.py b/python/pyspark/tests/test_rdd.py
index 4c43d6181cc3..354830666ef1 100644
--- a/python/pyspark/tests/test_rdd.py
+++ b/python/pyspark/tests/test_rdd.py
@@ -638,6 +638,36 @@ class RDDTests(ReusedPySparkTestCase):
self.assertEqual(result.getNumPartitions(), 5)
self.assertEqual(result.count(), 3)
+ def test_count_approx_respects_timeout(self):
+ def slow(x):
+ time.sleep(1)
+ return x
+
+ rdd = self.sc.parallelize(range(20), 20).map(slow)
+ start = time.time()
+ rdd.countApprox(timeout=100)
+ elapsed = time.time() - start
+ self.assertLess(elapsed, 2)
+ # Cancel the background approximate job before subsequent tests run.
+ self.sc.cancelAllJobs()
+
+ def test_count_approx_returns_exact_when_completed(self):
+ rdd = self.sc.parallelize(range(1000), 8)
+ self.assertEqual(rdd.countApprox(timeout=5000), 1000)
+
+ def test_mean_approx_respects_timeout(self):
+ def slow(x):
+ time.sleep(1)
+ return float(x)
+
+ rdd = self.sc.parallelize(range(20), 20).map(slow)
+ start = time.time()
+ rdd.meanApprox(timeout=100)
+ elapsed = time.time() - start
+ self.assertLess(elapsed, 2)
+ # Cancel the background approximate job before subsequent tests run.
+ self.sc.cancelAllJobs()
+
def test_external_group_by_key(self):
self.sc._conf.set("spark.python.worker.memory", "1m")
N = 2000001
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
