Spark sample is too slow

Question

I'm trying to execute a simple random sample with Scala from an existing table, containing around 100e6 records.

import org.apache.spark.sql.SaveMode

val nSamples = 3e5.toInt
val frac = 1e-5
val table = spark.table("db_name.table_name").sample(false, frac).limit(nSamples)
(table
  .write
  .mode(SaveMode.Overwrite)
  .saveAsTable("db_name.new_name")
)

But it is taking too long (~5h by my estimates).

Useful information:

I have ~6 workers. By analyzing the number of partitions of the table I get: 11433.
I'm not sure if the partitions/workers ratio is reasonable.
I'm running Spark 2.1.0 using Scala.

I have tried:

Removing the .limit() part.
Changing frac to 1.0, 0.1, etc.

Question: how can I make it faster?

Best,

zero323 · Accepted Answer · 2018-04-25T15:50:56.230

5

Limit is definitely worth removing, but the real problem is that sampling requires a full data scan. No matter how low is the fraction, the time complexity is still O(N)*.

If you don't require good statistical properties, you can try to limit amount of data you've loaded in the first place by sampling data files first, and then subsampling from the reduced dataset. This might work reasonably well, if data is uniformly distributed.

Otherwise there is not much you can do about it, other than scaling your cluster.

* How do simple random sampling and dataframe SAMPLE function work in Apache Spark (Scala)?

edited Apr 25 '18 at 15:50

answered Apr 24 '18 at 14:50

zero323

322,348
103
959
935

@user6910411, do you mind adding a link to your statement? Possibly: https://en.wikipedia.org/wiki/Reservoir_sampling – fmv1992 Apr 24 '18 at 16:32
@monteiro I am not sure if I understand your request. Do you believe that the statement is wrong and Reservoir sampling does resolve the problem? If the answer is positive it does not. Reservoir sampling addresses problem of memory complexity, which is really not an issue here. The problem you face is time complexity. It effectively doesn't matter how small is the fraction, we still scan all data points. It could be possibly improved if we had efficient random access to a single record, which we don't (mostly because we operate on stream like data, use large filesystem blocks, etc.). – zero323 Apr 24 '18 at 17:26
I believe the statement is right. I suggested indeed a wrong reference for your statement that sampling is *O(N)*. I'm kindly asking for you to put a reference to that statement so that people can verify it elsewhere. – fmv1992 Apr 25 '18 at 11:17

score 2 · Answer 2 · answered Apr 24 '18 at 19:53

you could first sample the partitions, and then sample from the partitions. Like this, you dont need a full table scan, but only works if your partitioning itself is random. AFAIK you need to use RDD API for this. This could look like this (plug in the numbers to match your desired number of samples):

val ds : Dataset[String] = ???

  val dsSampled = ds.rdd
  // take 1000 samples from every 10th partition
  .mapPartitionsWithIndex{case (i,rows) => if(i%10==0) scala.util.Random.shuffle(rows).take(1000) else Iterator.empty}
  .toDS()

Spark sample is too slow

2 Answers2

Linked

Related