Case class equality in Apache Spark

by

This is a known issue with Spark REPL. You can find more details in SPARK-2620. It affects multiple operations in Spark REPL including most of transformations on the PairwiseRDDs. For example:

case class Foo(x: Int)

val foos = Seq(Foo(1), Foo(1), Foo(2), Foo(2))
foos.distinct.size
// Int = 2

val foosRdd = sc.parallelize(foos, 4)
foosRdd.distinct.count
// Long = 4  

foosRdd.map((_, 1)).reduceByKey(_ + _).collect
// Array[(Foo, Int)] = Array((Foo(1),1), (Foo(1),1), (Foo(2),1), (Foo(2),1))

foosRdd.first == foos.head
// Boolean = false

Foo.unapply(foosRdd.first) == Foo.unapply(foos.head)
// Boolean = true

What makes it even worse is that the results depend on the data distribution:

sc.parallelize(foos, 1).distinct.count
// Long = 2

sc.parallelize(foos, 1).map((_, 1)).reduceByKey(_ + _).collect
// Array[(Foo, Int)] = Array((Foo(2),2), (Foo(1),2))

The simplest thing you can do is to define and package required case classes outside REPL. Any code submitted directly using spark-submit should work as well.

In Scala 2.11+ you can create a package directly in the REPL with paste -raw. 

scala> :paste -raw
// Entering paste mode (ctrl-D to finish)

package bar

case class Bar(x: Int)


// Exiting paste mode, now interpreting.

scala> import bar.Bar
import bar.Bar

scala> sc.parallelize(Seq(Bar(1), Bar(1), Bar(2), Bar(2))).distinct.collect
res1: Array[bar.Bar] = Array(Bar(1), Bar(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. How to define schema for custom type in Spark SQL?
  5. How to override apply in a case class companion
  6. Spark performance for Scala vs Python
  7. Stackoverflow due to long RDD Lineage
  8. How to find spark RDD/Dataframe size?
  9. Modify collection inside a Spark RDD foreach
  10. Parsing multiline records in Scala
  11. Explanation of fold method of spark RDD
  12. Why does partition parameter of SparkContext.textFile not take effect?
  13. Why does Spark RDD partition has 2GB limit for HDFS?
  14. Reading in multiple files compressed in tar.gz archive into Spark [duplicate]
  15. How to sort an RDD in Scala Spark?
  16. How is this case class match pattern working?
  17. How to transpose an RDD in Spark
  18. Spark – load CSV file as DataFrame?
  19. How to define partitioning of DataFrame?
  20. Why accesing DataFrame from UDF results in NullPointerException?
  21. How to read records in JSON format from Kafka using Structured Streaming?
  22. Spark Dataframe :How to add a index Column : Aka Distributed Data Index
  23. Reading DataFrame from partitioned parquet file
  24. Perform a typed join in Scala with Spark Datasets
  25. Left Anti join in Spark?
  26. DataFrame-ified zipWithIndex
  27. How to explode an array into multiple columns in Spark
  28. How to print the contents of RDD?
  29. Scala-Spark Dynamically call groupby and agg with parameter values
  30. Joining two dataframes without a common column

Leave a Comment