Hi, I am seeing different shuffle write sizes when using SchemaRDD (versus normal RDD). I'm doing the following:
case class DomainObj(a: String, b: String, c: String, d: String) val logs: RDD[String] = sc.textFile(...) val filtered: RDD[String] = logs.filter(...) val myDomainObjects: RDD[DomainObj] = filtered.flatMap(...) ------------------------------------------------------------ 1. Operations on RDD: ------------------------------------------------------------ val results = requests .filter(obj => obj.a == "SomeValue" || obj.a == "SomeOtherValue") .mapPartitions(objs => objs.map(obj => (obj, 1))) .reduceByKey(_ + _, 200) .collect() ------------------------------------------------------------ 2. Operations on SchemaRDD: ------------------------------------------------------------ myDomainObjects.registerTempTable("myDomainObjects") val results = sqlContext.sql(""" SELECT a, b, c, d, COUNT(*) total FROM myDomainObjects WHERE a IN ('SomeValue', 'SomeOtherValue') GROUP BY a, b, c, d """).collect() In the first case (RDD), the query returns in 2 minutes and 30 seconds with the input size 28.4GB, and shuffle write size 525.3MB and shuffle read size 472.5MB. In the second case (SchemaRDD), the query returns in 2 minutes and 9 seconds with input size 28.4GB, and shuffle write 258.9MB and shuffle read 233.0MB. Since in the second case, the shuffle size is half of the first case, I'd like to understand why. Thanks, Grega