Broadcast outer joins are on my short list for 1.5. On Fri, Jun 19, 2015 at 10:48 AM, Piero Cinquegrana < [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.marketsharepartners.com/> > 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] > For additional commands, e-mail: [email protected] >
