Any tips on how to implement and broadcast left outer join using Scala? From: Michael Armbrust [mailto:[email protected]] Sent: Friday, June 19, 2015 12:40 PM To: Piero Cinquegrana Cc: [email protected] Subject: Re: SparkSQL: leftOuterJoin is VERY slow!
Broadcast outer joins are on my short list for 1.5. On Fri, Jun 19, 2015 at 10:48 AM, Piero Cinquegrana <[email protected]<mailto:[email protected]>> wrote: Hello, I have two RDDs: tv and sessions. I need to convert these DataFrames into RDDs because I need to use the groupByKey function. The reduceByKey function would not work here as I am not doing any aggregations, but I am grouping using a (K, V) pair. See the snippets of code below. The bottleneck is definitely not groupByKey as it runs in 78 seconds. I find that the leftOuterJoin takes 1.5 hours using 80 executors with 2g of memory each. The Spark driver has 5g of memory. I have tried using different numPartitions: such as 20, 50 and 200 and I found that 200 works best with 80 executors. I also tried to reduce spark.sql.shuffle.partitions to 40 at no avail. Any suggestions on how to improve the performance of this outer join? I have been told to try to broadcast the smaller tv table and do a mapper-side join. Do you think this is feasible with such a large table (4.4M rows)? Much appreciated, Piero Cinquegrana Marketing Scientist | MarketShare 11150 Santa Monica Blvd, 5th Floor, Los Angeles, CA 90025 P: 310.914.5677 x242 M: 323.377.9197 www.marketshare.com<http://www.marketshare.com><http://www.marketsharepartners.com/> twitter.com/marketsharep<http://twitter.com/marketsharep><http://twitter.com/marketsharep> > val > tv_key=tv.rdd.map(line=>((line.getInt(4).toString,line.getString(1)),line)).groupByKey(200) tv_key: org.apache.spark.rdd.RDD[((String, String), Iterable[org.apache.spark.sql.Row])] = ShuffledRDD[24] at groupByKey at <console>:46 > tv_key.count res33: Long = 4406433 Took 370 seconds > tv_key.take(5).foreach(println) ((537,2014-03-30 11:17:00),CompactBuffer([226,2014-03-30 11:17:00,2014-03-30 11:17:28,2014-03-30 11:17:28,537,Bangor,EST5EDT,UNKNOWN,AD15,TV Land,Weekend,M,UNKNOWN|15|TV Land|Weekend|M,815,0.005836018617973505])) ((516,2014-01-01 17:31:00),CompactBuffer([111,2014-01-01 17:31:00,2014-01-01 17:31:14,2014-01-01 17:31:14,516,Erie,EST5EDT,UNKNOWN,AD5,Headline News,Early Fringe,M,UNKNOWN|5|Headline News|Early Fringe|M,186,0.0011828298887122416], [111,2014-01-01 17:31:00,2014-01-01 17:31:19,2014-01-01 17:31:19,516,Erie,EST5EDT,UNKNOWN,AD10,Headline News,Early Fringe,M,UNKNOWN|10|Headline News|Early Fringe|M,186,0.0011828298887122416])) ((855,2014-08-28 03:32:00),CompactBuffer([113,2014-08-28 03:32:00,2014-08-28 03:32:50,2014-08-28 00:32:50,855,SantaBarbra-SanMar-SanLuOb,PST8PDT,UNKNOWN,AD15,History Channel,Late Fringe,M,UNKNOWN|15|History Channel|Late Fringe|M,410,0.001749818616362938])) ((743,2014-07-17 06:57:00),CompactBuffer([35,2014-07-17 06:57:00,2014-07-17 06:57:18,2014-07-17 10:57:18,743,Anchorage,YST9YDT,UNKNOWN,AD30,CMT: Country Music Television,Daytime,M,UNKNOWN|30|CMT: Country Music Television|Daytime|M,227,0.0014356185175815835])) ((574,2014-06-03 18:47:00),CompactBuffer([94,2014-06-03 18:47:00,2014-06-03 18:47:45,2014-06-03 18:47:45,574,Johnstown-Altoona-St Colge,EST5EDT,UNKNOWN,AD30,Game Show Network,Early Fringe,M,UNKNOWN|30|Game Show Network|Early Fringe|M,1629,0.005588527908333047])) > val sessions_key = sessions.rdd.map(line => > ((line.getString(2),line.getString(1)),line)) sessions_key: org.apache.spark.rdd.RDD[((String, String), org.apache.spark.sql.Row)] = MapPartitionsRDD[30] at map at <console>:38 > sessions_key.count res16: Long = 110383440 Took 78 seconds > sessions_key.take(5).foreach(println) ((737,2014-01-01 02:01:00),[2014-01-01 01:01:00,2014-01-01 02:01:00,737,Mankato,0.0,CST6CDT,Wednesday,1,1,1,1,2014,Wednesday|1]) ((569,2014-01-01 02:01:00),[2014-01-01 02:01:00,2014-01-01 02:01:00,569,Harrisonburg,0.0,EST5EDT,Wednesday,2,1,1,1,2014,Wednesday|2]) ((753,2014-01-01 02:01:00),[2014-01-01 00:01:00,2014-01-01 02:01:00,753,Phoenix (Prescott),11.0,MST,Wednesday,0,1,1,1,2014,Wednesday|0]) ((825,2014-01-01 02:01:00),[2013-12-31 23:01:00,2014-01-01 02:01:00,825,San Diego,0.0,PST8PDT,Tuesday,23,1,1,12,2013,Tuesday|23]) ((673,2014-01-01 02:01:00),[2014-01-01 01:01:00,2014-01-01 02:01:00,673,Columbus-Tupelo-W Pnt-Hstn,0.0,CST6CDT,Wednesday,1,1,1,1,2014,Wednesday|1]) > val stack = sessions_key.leftOuterJoin(tv_key, 200) stack: org.apache.spark.rdd.RDD[((String, String), (org.apache.spark.sql.Row, Option[Iterable[org.apache.spark.sql.Row]]))] = MapPartitionsRDD[33] at leftOuterJoin at <console>:54 > stack.count res18: Long = 110383440 Took 5283 seconds --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected]<mailto:[email protected]> For additional commands, e-mail: [email protected]<mailto:[email protected]>
