Spark: what’s the best strategy for joining a 2-tuple-key RDD with single-key RDD?

by

One option is to perform a broadcast join by collecting rdd1 to the driver and broadcasting it to all mappers; done correctly, this will let us avoid an expensive shuffle of the large rdd2 RDD: 

val rdd1 = sc.parallelize(Seq((1, "A"), (2, "B"), (3, "C")))
val rdd2 = sc.parallelize(Seq(((1, "Z"), 111), ((1, "ZZ"), 111), ((2, "Y"), 222), ((3, "X"), 333)))

val rdd1Broadcast = sc.broadcast(rdd1.collectAsMap())
val joined = rdd2.mapPartitions({ iter =>
  val m = rdd1Broadcast.value
  for {
    ((t, w), u) <- iter
    if m.contains(t)
  } yield ((t, w), (u, m.get(t).get))
}, preservesPartitioning = true)

The preservesPartitioning = true tells Spark that this map function doesn’t modify the keys of rdd2; this will allow Spark to avoid re-partitioning rdd2 for any subsequent operations that join based on the (t, w) key.

This broadcast could be inefficient since it involves a communications bottleneck at the driver. In principle, it’s possible to broadcast one RDD to another without involving the driver; I have a prototype of this that I’d like to generalize and add to Spark.

Another option is to re-map the keys of rdd2 and use the Spark join method; this will involve a full shuffle of rdd2 (and possibly rdd1):

rdd1.join(rdd2.map {
  case ((t, w), u) => (t, (w, u))
}).map {
  case (t, (v, (w, u))) => ((t, w), (u, v))
}.collect()

On my sample input, both of these methods produce the same result:

res1: Array[((Int, java.lang.String), (Int, java.lang.String))] = Array(((1,Z),(111,A)), ((1,ZZ),(111,A)), ((2,Y),(222,B)), ((3,X),(333,C)))

A third option would be to restructure rdd2 so that t is its key, then perform the above join.

More Related Contents:

  1. How to bucket the range of values from a column and count how many values fall into each interval in scala?
  2. How to store custom objects in Dataset?
  3. How to avoid duplicate columns after join?
  4. Flattening Rows in Spark
  5. Split 1 column into 3 columns in spark scala
  6. Provide schema while reading csv file as a dataframe
  7. How to flatmap a nested Dataframe in Spark
  8. How to force DataFrame evaluation in Spark
  9. How to write spark streaming DF to Kafka topic
  10. Better way to convert a string field into timestamp in Spark
  11. How to find spark RDD/Dataframe size?
  12. call of distinct and map together throws NPE in spark library
  13. How to get path to the uploaded file
  14. how to filter out a null value from spark dataframe
  15. Is there a reason not to use SparkContext.getOrCreate when writing a spark job?
  16. What are the various join types in Spark?
  17. Provide schema while reading csv file as a dataframe in Scala Spark
  18. Replace missing values with mean – Spark Dataframe
  19. How to calculate the size of dataframe in bytes in Spark?
  20. Why does partition parameter of SparkContext.textFile not take effect?
  21. Spark Scala list folders in directory
  22. Exception in thread “main” java.lang.NoSuchMethodError: scala.Product.$init$(Lscala/Product;)
  23. Spark-Monotonically increasing id not working as expected in dataframe?
  24. How to remove parentheses around records when saveAsTextFile on RDD[(String, Int)]?
  25. Reading in multiple files compressed in tar.gz archive into Spark [duplicate]
  26. Spark : how to run spark file from spark shell
  27. How can I connect to a postgreSQL database into Apache Spark using scala?
  28. Spark Dataframe Random UUID changes after every transformation/action
  29. Spark column string replace when present in other column (row)
  30. How to vectorize DataFrame columns for ML algorithms?

Leave a Comment