Skip to content

[SPARK-12031][Core][BUG]: Integer overflow when do sampling #10023

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
uncleGen wants to merge 4 commits into from
Closed

[SPARK-12031][Core][BUG]: Integer overflow when do sampling #10023

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
d7eec45
bug fix
uncleGen Nov 28, 2015
986fa5c
code style
uncleGen Nov 28, 2015
bb2ed41
modify some pointless and verbose code
uncleGen Nov 28, 2015
3cafda5
bug fix
uncleGen Nov 30, 2015
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
4 changes: 2 additions & 2 deletions core/src/main/scala/org/apache/spark/Partitioner.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -253,7 +253,7 @@ private[spark] object RangePartitioner {
*/
def sketch[K : ClassTag](
rdd: RDD[K],
sampleSizePerPartition: Int): (Long, Array[(Int, Int, Array[K])]) = {
sampleSizePerPartition: Int): (Long, Array[(Int, Long, Array[K])]) = {
val shift = rdd.id
// val classTagK = classTag[K] // to avoid serializing the entire partitioner object
val sketched = rdd.mapPartitionsWithIndex { (idx, iter) =>
Expand All @@ -262,7 +262,7 @@ private[spark] object RangePartitioner {
iter, sampleSizePerPartition, seed)
Iterator((idx, n, sample))
}.collect()
val numItems = sketched.map(_._2.toLong).sum
val numItems = sketched.map(_._2).sum
(numItems, sketched)
}

Expand Down
15 changes: 10 additions & 5 deletions core/src/main/scala/org/apache/spark/util/random/SamplingUtils.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,7 @@ private[spark] object SamplingUtils {
input: Iterator[T],
k: Int,
seed: Long = Random.nextLong())
: (Array[T], Int) = {
: (Array[T], Long) = {
val reservoir = new Array[T](k)
// Put the first k elements in the reservoir.
var i = 0
Expand All @@ -52,16 +52,21 @@ private[spark] object SamplingUtils {
(trimReservoir, i)
} else {
// If input size > k, continue the sampling process.
var l = i.toLong
val rand = new XORShiftRandom(seed)
while (input.hasNext) {
val item = input.next()
val replacementIndex = rand.nextInt(i)
val replacementIndex = if (l < Int.MaxValue) {
rand.nextInt(l.toInt)
} else {
rand.nextLong()

@srowen srowen Nov 28, 2015

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

This isn't valid, because it chooses a value from (negative) Long.MIN_VALUE to Long.MAX_VALUE. You need to choose a number in [0, l).

Rather than even special case it, just use

val replacementIndex = (random.nextDouble() * l).toLong

@uncleGen uncleGen Nov 30, 2015

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

OK

}
if (replacementIndex < k) {
reservoir(replacementIndex) = item
reservoir(replacementIndex.toInt) = item
}
i += 1
l += 1
}
(reservoir, i)
(reservoir, l)
}
}

Expand Down