Spark: Best practice for retrieving big data from RDD to local machine

by

Update: RDD.toLocalIterator method that appeared after the original answer has been written is a more efficient way to do the job. It uses runJob to evaluate only a single partition on each step.

TL;DR And the original answer might give a rough idea how it works:

First of all, get the array of partition indexes:

val parts = rdd.partitions

Then create smaller rdds filtering out everything but a single partition. Collect the data from smaller rdds and iterate over values of a single partition:

for (p <- parts) {
    val idx = p.index
    val partRdd = rdd.mapPartitionsWithIndex(a => if (a._1 == idx) a._2 else Iterator(), true)
    //The second argument is true to avoid rdd reshuffling
    val data = partRdd.collect //data contains all values from a single partition 
                               //in the form of array
    //Now you can do with the data whatever you want: iterate, save to a file, etc.
}

I didn’t try this code, but it should work. Please write a comment if it won’t compile. Of cause, it will work only if the partitions are small enough. If they aren’t, you can always increase the number of partitions with rdd.coalesce(numParts, true).

More Related Contents:

  1. Spark SQL – load data with JDBC using SQL statement, not table name
  2. Spark Dataframe validating column names for parquet writes
  3. Spark: subtract two DataFrames
  4. TypeError: Column is not iterable – How to iterate over ArrayType()?
  5. How to overwrite the output directory in spark
  6. Is groupByKey ever preferred over reduceByKey
  7. Filtering a spark dataframe based on date
  8. Pyspark : forward fill with last observation for a DataFrame
  9. spark.sql.crossJoin.enabled for Spark 2.x
  10. Spark DataFrame Schema Nullable Fields
  11. Why do I have to explicitly tell Spark what to cache?
  12. multiple conditions for filter in spark data frames
  13. How to find count of Null and Nan values for each column in a PySpark dataframe efficiently?
  14. Spark SQL broadcast hash join
  15. What is the difference between cache and persist?
  16. What is the difference between Apache Spark SQLContext vs HiveContext?
  17. Saving dataframe to local file system results in empty results
  18. Dealing with a large gzipped file in Spark
  19. What should be the optimal value for spark.sql.shuffle.partitions or how do we increase partitions when using Spark SQL?
  20. pyspark: Efficiently have partitionBy write to same number of total partitions as original table
  21. Spark lists all leaf node even in partitioned data
  22. How to get day of week in SparkSQL?
  23. AWS EMR – ModuleNotFoundError: No module named ‘pyarrow’
  24. Array Intersection in Spark SQL
  25. How to assign unique contiguous numbers to elements in a Spark RDD
  26. What is spark.driver.maxResultSize?
  27. Spark + s3 – error – java.lang.ClassNotFoundException: Class org.apache.hadoop.fs.s3a.S3AFileSystem not found
  28. How to get Kafka offsets for structured query for manual and reliable offset management?
  29. Spark: disk I/O on stage boundaries explanation
  30. spark.ml StringIndexer throws ‘Unseen label’ on fit()

Leave a Comment