How to transpose an RDD in Spark

by

Say you have an N×M matrix.

If both N and M are so small that you can hold N×M items in memory, it doesn’t make much sense to use an RDD. But transposing it is easy:

val rdd = sc.parallelize(Seq(Seq(1, 2, 3), Seq(4, 5, 6), Seq(7, 8, 9)))
val transposed = sc.parallelize(rdd.collect.toSeq.transpose)

If N or M is so large that you cannot hold N or M entries in memory, then you cannot have an RDD line of this size. Either the original or the transposed matrix is impossible to represent in this case.

N and M may be of a medium size: you can hold N or M entries in memory, but you cannot hold N×M entries. In this case you have to blow up the matrix and put it together again:

val rdd = sc.parallelize(Seq(Seq(1, 2, 3), Seq(4, 5, 6), Seq(7, 8, 9)))
// Split the matrix into one number per line.
val byColumnAndRow = rdd.zipWithIndex.flatMap {
  case (row, rowIndex) => row.zipWithIndex.map {
    case (number, columnIndex) => columnIndex -> (rowIndex, number)
  }
}
// Build up the transposed matrix. Group and sort by column index first.
val byColumn = byColumnAndRow.groupByKey.sortByKey().values
// Then sort by row index.
val transposed = byColumn.map {
  indexedRow => indexedRow.toSeq.sortBy(_._1).map(_._2)
}

More Related Contents:

  1. How does HashPartitioner work?
  2. How to convert rdd object to dataframe in spark
  3. (Why) do we need to call cache or persist on a RDD
  4. Case class equality in Apache Spark
  5. Spark performance for Scala vs Python
  6. Stackoverflow due to long RDD Lineage
  7. How to find spark RDD/Dataframe size?
  8. Modify collection inside a Spark RDD foreach
  9. Parsing multiline records in Scala
  10. Explanation of fold method of spark RDD
  11. Why does partition parameter of SparkContext.textFile not take effect?
  12. Why does Spark RDD partition has 2GB limit for HDFS?
  13. Reading in multiple files compressed in tar.gz archive into Spark [duplicate]
  14. How to sort an RDD in Scala Spark?
  15. How to store custom objects in Dataset?
  16. Split 1 column into 3 columns in spark scala
  17. Provide schema while reading csv file as a dataframe
  18. How to flatmap a nested Dataframe in Spark
  19. Joining Spark dataframes on the key
  20. Matrix Multiplication in Apache Spark [closed]
  21. How to write unit tests in Spark 2.0+?
  22. Customize SparkContext using sparkConf.set(..) when using spark-shell
  23. Spark: Transpose DataFrame Without Aggregating
  24. Exploding nested Struct in Spark dataframe
  25. How to compute cumulative sum using Spark
  26. Add Number of days column to Date Column in same dataframe for Spark Scala App
  27. Spark : how to run spark file from spark shell
  28. Spark Dataframe Random UUID changes after every transformation/action
  29. How to set hadoop configuration values from pyspark
  30. Spark ML VectorAssembler returns strange output

Leave a Comment