Skip to content

[SPARK-40301][PYTHON] Add parameter validations in pyspark.rdd #37752

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
zhengruifeng wants to merge 1 commit into from
Closed

[SPARK-40301][PYTHON] Add parameter validations in pyspark.rdd #37752

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 8 additions & 1 deletion python/pyspark/rdd.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1039,7 +1039,8 @@ def sample(
>>> 6 <= rdd.sample(False, 0.1, 81).count() <= 14
True
"""
assert fraction >= 0.0, "Negative fraction value: %s" % fraction
if not fraction >= 0:
raise ValueError("Fraction must be nonnegative.")
Copy link
Member

@Yikun Yikun Sep 2, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Change assert to ValueError is right definately, I believe all assert in main code (espacailly, use when validation params) should be fix (test mode is okay)

[1] https://mail.python.org/pipermail/python-list/2013-November/810940.html

Copy link
Contributor Author

@zhengruifeng zhengruifeng Sep 2, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

AssertionError is replaced with a ValueError, but I think it maybe trivial to add it to migration-doc

return self.mapPartitionsWithIndex(RDDSampler(withReplacement, fraction, seed).func, True)

def randomSplit(
Expand Down Expand Up @@ -1077,7 +1078,11 @@ def randomSplit(
>>> 250 < rdd2.count() < 350
True
"""
if not all(w >= 0 for w in weights):
raise ValueError("Weights must be nonnegative")
s = float(sum(weights))
if not s > 0:
raise ValueError("Sum of weights must be positive")
cweights = [0.0]
for w in weights:
cweights.append(cweights[-1] + w / s)
Expand Down Expand Up @@ -4565,6 +4570,8 @@ def coalesce(self: "RDD[T]", numPartitions: int, shuffle: bool = False) -> "RDD[
>>> sc.parallelize([1, 2, 3, 4, 5], 3).coalesce(1).glom().collect()
[[1, 2, 3, 4, 5]]
"""
if not numPartitions > 0:
raise ValueError("Number of partitions must be positive.")
if shuffle:
# Decrease the batch size in order to distribute evenly the elements across output
# partitions. Otherwise, repartition will possibly produce highly skewed partitions.
Expand Down