Sounds good. I will manual merge this patch on 1.6.1, and test again for my case tomorrow on my environment and will update later. Thanks Yong
> Date: Wed, 23 Mar 2016 16:20:23 -0700 > Subject: Re: Spark 1.5.2, why the broadcast join shuffle so much data in the > last step > From: dav...@databricks.com > To: java8...@hotmail.com > CC: user@spark.apache.org > > On Wed, Mar 23, 2016 at 10:35 AM, Yong Zhang <java8...@hotmail.com> wrote: > > Here is the output: > > > > == Parsed Logical Plan == > > Project [400+ columns] > > +- Project [400+ columns] > > +- Project [400+ columns] > > +- Project [400+ columns] > > +- Join Inner, Some((((visid_high#460L = visid_high#948L) && > > (visid_low#461L = visid_low#949L)) && (date_time#25L > date_time#513L))) > > :- Relation[400+ columns] ParquetRelation > > +- BroadcastHint > > +- Project [soid_e1#30 AS > > account_id#976,visid_high#460L,visid_low#461L,date_time#25L,ip#127] > > +- Filter (instr(event_list#105,202) > 0) > > +- Relation[400+ columns] ParquetRelation > > > > == Analyzed Logical Plan == > > 400+ columns > > Project [400+ columns] > > +- Project [400+ columns] > > +- Project [400+ columns] > > +- Project [400+ columns] > > +- Join Inner, Some((((visid_high#460L = visid_high#948L) && > > (visid_low#461L = visid_low#949L)) && (date_time#25L > date_time#513L))) > > :- Relation[400+ columns] ParquetRelation > > +- BroadcastHint > > +- Project [soid_e1#30 AS > > account_id#976,visid_high#460L,visid_low#461L,date_time#25L,ip#127] > > +- Filter (instr(event_list#105,202) > 0) > > +- Relation[400+ columns] ParquetRelation > > > > == Optimized Logical Plan == > > Project [400+ columns] > > +- Join Inner, Some((((visid_high#460L = visid_high#948L) && (visid_low#461L > > = visid_low#949L)) && (date_time#25L > date_time#513L))) > > :- Relation[400+ columns] ParquetRelation > > +- Project [date_time#25L,visid_low#461L,visid_high#460L,account_id#976] > > +- BroadcastHint > > +- Project [soid_e1#30 AS > > account_id#976,visid_high#460L,visid_low#461L,date_time#25L,ip#127] > > +- Filter (instr(event_list#105,202) > 0) > > +- Relation[400+ columns] ParquetRelation > > There is a Project on top of BroadcastHint, which is inserted by > column pruning rule, that make > the SparkStratege can not regonize BroadcastHint anymore, it's fixed > recently in master [1] > > https://github.com/apache/spark/pull/11260 > > Your join should run as expected in master. > > > == Physical Plan == > > Project [400+ columns] > > +- Filter (date_time#25L > date_time#513L) > > +- SortMergeJoin [visid_high#948L,visid_low#949L], > > [visid_high#460L,visid_low#461L] > > :- Sort [visid_high#948L ASC,visid_low#949L ASC], false, 0 > > : +- TungstenExchange > > hashpartitioning(visid_high#948L,visid_low#949L,200), None > > : +- Scan ParquetRelation[400+ columns] InputPaths: > > hdfs://xxx/2015/12/17, hdfs://xxx/2015/12/18, hdfs://xxx/2015/12/19, > > hdfs://xxx/2015/12/20, hdfs://xxx/2015/12/21, hdfs://xxx/2015/12/22, > > hdfs://xxx/2015/12/23, hdfs://xxx/2015/12/24, hdfs://xxx/2015/12/25, > > hdfs://xxx/2015/12/26, hdfs://xxx/2015/12/27, hdfs://xxx/2015/12/28, > > hdfs://xxx/2015/12/29, hdfs://xxx/2015/12/30, hdfs://xxx/2015/12/31, > > hdfs://xxx/2016/01/01, hdfs://xxx/2016/01/02, hdfs://xxx/2016/01/03, > > hdfs://xxx/2016/01/04, hdfs://xxx/2016/01/05, hdfs://xxx/2016/01/06, > > hdfs://xxx/2016/01/07, hdfs://xxx/2016/01/08, hdfs://xxx/2016/01/09, > > hdfs://xxx/2016/01/10, hdfs://xxx/2016/01/11, hdfs://xxx/2016/01/12, > > hdfs://xxx/2016/01/13, hdfs://xxx/2016/01/14, hdfs://xxx/2016/01/15, > > hdfs://xxx/2016/01/16, hdfs://xxx/2016/01/17, hdfs://xxx/2016/01/18, > > hdfs://xxx/2016/01/19, hdfs://xxx/2016/01/20, hdfs://xxx/2016/01/21, > > hdfs://xxx/2016/01/22, hdfs://xxx/2016/01/23, hdfs://xxx/2016/01/24, > > hdfs://xxx/2016/01/25, hdfs://xxx/2016/01/26, hdfs://xxx/2016/01/27, > > hdfs://xxx/2016/01/28, hdfs://xxx/2016/01/29, hdfs://xxx/2016/01/30, > > hdfs://xxx/2016/01/31, hdfs://xxx/2016/02/01, hdfs://xxx/2016/02/02, > > hdfs://xxx/2016/02/03, hdfs://xxx/2016/02/04, hdfs://xxx/2016/02/05, > > hdfs://xxx/2016/02/06, hdfs://xxx/2016/02/07, hdfs://xxx/2016/02/08, > > hdfs://xxx/2016/02/09, hdfs://xxx/2016/02/10, hdfs://xxx/2016/02/11, > > hdfs://xxx/2016/02/12, hdfs://xxx/2016/02/13, hdfs://xxx/2016/02/14, > > hdfs://xxx/2016/02/15, hdfs://xxx/2016/02/16, hdfs://xxx/2016/02/17, > > hdfs://xxx/2016/02/18, hdfs://xxx/2016/02/19, hdfs://xxx/2016/02/20, > > hdfs://xxx/2016/02/21, hdfs://xxx/2016/02/22, hdfs://xxx/2016/02/23, > > hdfs://xxx/2016/02/24, hdfs://xxx/2016/02/25, hdfs://xxx/2016/02/26, > > hdfs://xxx/2016/02/27, hdfs://xxx/2016/02/28, hdfs://xxx/2016/02/29, > > hdfs://xxx/2016/03/01, hdfs://xxx/2016/03/02, hdfs://xxx/2016/03/03, > > hdfs://xxx/2016/03/04, hdfs://xxx/2016/03/05, hdfs://xxx/2016/03/06, > > hdfs://xxx/2016/03/07, hdfs://xxx/2016/03/08, hdfs://xxx/2016/03/09, > > hdfs://xxx/2016/03/10, hdfs://xxx/2016/03/11, hdfs://xxx/2016/03/12, > > hdfs://xxx/2016/03/13, hdfs://xxx/2016/03/14, hdfs://xxx/2016/03/15, > > hdfs://xxx/2016/03/16, hdfs://xxx/2016/03/17 > > +- Sort [visid_high#460L ASC,visid_low#461L ASC], false, 0 > > +- TungstenExchange > > hashpartitioning(visid_high#460L,visid_low#461L,200), None > > +- Project > > [date_time#25L,visid_low#461L,visid_high#460L,account_id#976] > > +- Project [soid_e1#30 AS > > account_id#976,visid_high#460L,visid_low#461L,date_time#25L,ip#127] > > +- Filter (instr(event_list#105,202) > 0) > > +- Scan > > ParquetRelation[visid_low#461L,ip#127,soid_e1#30,event_list#105,visid_high#460L,date_time#25L] > > InputPaths: hdfs://xxx/2016/03/17 > > > > This dataset has more than 480 columns in parquet file, so I replaced them > > with "400+ columns", without blow out the email, but I don't think this > > could do anything with "broadcast" problem. > > > > Thanks > > > > Yong > > > > > >> Date: Wed, 23 Mar 2016 10:14:19 -0700 > >> Subject: Re: Spark 1.5.2, why the broadcast join shuffle so much data in > >> the last step > >> From: dav...@databricks.com > >> To: java8...@hotmail.com > >> CC: user@spark.apache.org > > > >> > >> The broadcast hint does not work as expected in this case, could you > >> also how the logical plan by 'explain(true)'? > >> > >> On Wed, Mar 23, 2016 at 8:39 AM, Yong Zhang <java8...@hotmail.com> wrote: > >> > > >> > So I am testing this code to understand "broadcast" feature of DF on > >> > Spark 1.6.1. > >> > This time I am not disable "tungsten". Everything is default value, > >> > except setting memory and cores of my job on 1.6.1. > >> > > >> > I am testing the join2 case > >> > > >> > val join2 = historyRaw.join(broadcast(trialRaw), trialRaw("visid_high") > >> > === historyRaw("visid_high") && trialRaw("visid_low") === > >> > historyRaw("visid_low") && trialRaw("date_time") > > >> > historyRaw("date_time")) > >> > > >> > and here is the DAG visualization in the runtime of my testing job: > >> > > >> > > >> > > >> > > >> > > >> > So now, I don't understand how the "broadcast" works on DateFrame in > >> > Spark. I originally thought it will be the same as "mapjoin" in the hive, > >> > but can someone explain the DAG above me? > >> > > >> > I have one day data about 1.5G compressed parquet file, filter by > >> > "instr(loadRaw("event_list"), "202") > 0", which will only output about > >> > 1494 > >> > rows (very small), and it is the "trailRaw" DF in my example. > >> > Stage 3 has a filter, which I thought is for the trailRaw data, but the > >> > stage statics doesn't match with the data. I don't know why the input is > >> > only 78M, and shuffle write is about 97.6KB > >> > > >> > > >> > > >> > > >> > The historyRaw will be about 90 days history data, which should be about > >> > 100G, so it looks like stage 4 is scanning it > >> > > >> > > >> > > >> > > >> > Now, my original thought is that small data will be broadcasted to all > >> > the nodes, and most of history data will be filtered out by the join > >> > keys, > >> > at least that will be the "mapjoin" in Hive will do, but from the DAG > >> > above, > >> > I didn't see it working this way. > >> > It is more like that Spark use the SortMerge join to shuffle both data > >> > across network, and filter on the "reducers" side by the join keys, to > >> > get > >> > the final output. But that is not the "broadcast" join supposed to do, > >> > correct? > >> > In the last stage, it will be very slow, until it reach and process all > >> > the history data, shown below as "shuffle read" reaching 720G, to finish. > >> > > >> > > >> > > >> > > >> > One thing I notice that if tungsten is enable, the shuffle write volume > >> > on stage 4 is larger (720G) than when tungsten is disable (506G) in my > >> > originally run, for the exactly same input. It is an interesting point, > >> > does > >> > anyone have some idea about this? > >> > > >> > > >> > Overall, for my test case, "broadcast" join is the exactly most > >> > optimized way I should use; but somehow, I cannot make it do the same > >> > way as > >> > "mapjoin" of Hive, even in Spark 1.6.1. > >> > > >> > As I said, this is a just test case. We have some business cases making > >> > sense to use "broadcast" join, but until I understand exactly how to > >> > make it > >> > work as I expect in Spark, I don't know what to do. > >> > > >> > Yong > >> > > >> > ________________________________ > >> > From: java8...@hotmail.com > >> > To: user@spark.apache.org > >> > Subject: RE: Spark 1.5.2, why the broadcast join shuffle so much data in > >> > the last step > >> > Date: Tue, 22 Mar 2016 13:08:31 -0400 > >> > > >> > > >> > Please help me understand how the "broadcast" will work on DF in Spark > >> > 1.5.2. > >> > > >> > Below are the 2 joins I tested and the physical plan I dumped: > >> > > >> > val join1 = historyRaw.join(trialRaw, trialRaw("visid_high") === > >> > historyRaw("visid_high") && trialRaw("visid_low") === > >> > historyRaw("visid_low") && trialRaw("date_time") > > >> > historyRaw("date_time")) > >> > val join2 = historyRaw.join(broadcast(trialRaw), trialRaw("visid_high") > >> > === historyRaw("visid_high") && trialRaw("visid_low") === > >> > historyRaw("visid_low") && trialRaw("date_time") > > >> > historyRaw("date_time")) > >> > > >> > join1.explain(true) > >> > == Physical Plan == > >> > Filter (date_time#25L > date_time#513L) > >> > SortMergeJoin [visid_high#948L,visid_low#949L], > >> > [visid_high#460L,visid_low#461L] > >> > ExternalSort [visid_high#948L ASC,visid_low#949L ASC], false > >> > Exchange hashpartitioning(visid_high#948L,visid_low#949L) > >> > Scan ParquetRelation[hdfs://xxxxxxxx] > >> > ExternalSort [visid_high#460L ASC,visid_low#461L ASC], false > >> > Exchange hashpartitioning(visid_high#460L,visid_low#461L) > >> > Project [soid_e1#30,visid_high#460L,visid_low#461L,date_time#25L] > >> > Filter (instr(event_list#105,202) > 0) > >> > Scan > >> > ParquetRelation[hdfs://xxx/2016/03/17][visid_high#460L,visid_low#461L,date_time#25L,event_list#105,soid_e1#30] > >> > > >> > join2.explain(true) > >> > == Physical Plan == > >> > Filter (date_time#25L > date_time#513L) > >> > BroadcastHashJoin [visid_high#948L,visid_low#949L], > >> > [visid_high#460L,visid_low#461L], BuildRight > >> > Scan ParquetRelation[hdfs://xxxxxxxx] > >> > Project [soid_e1#30,visid_high#460L,visid_low#461L,date_time#25L] > >> > Filter (instr(event_list#105,202) > 0) > >> > Scan > >> > ParquetRelation[hdfs://xxx/2016/03/17][visid_high#460L,visid_low#461L,date_time#25L,event_list#105,soid_e1#30] > >> > > >> > Obvious, the explain plans are different, but the performance and the > >> > job execution steps are almost exactly same, as shown in the original > >> > picture in the email below. > >> > Keep in mind that I have to run with "--conf > >> > spark.sql.tungsten.enabled=false", otherwise, the execution plan will do > >> > the > >> > tungsten sort. > >> > > >> > Now what confusing me is following: > >> > When using the broadcast join, the job still generates 3 stages, same as > >> > SortMergeJoin, but I am not sure this makes sense. > >> > Ideally, in "Broadcast", the first stage scan the "trialRaw" data, using > >> > the filter (instr(event_list#105,202) > 0), which BTW will filter out > >> > 99% of > >> > data, then "broadcasting" remaining data to all the nodes. Then scan > >> > "historyRaw", while filtering by join with broadcasted data. In the end, > >> > we > >> > can say there is one more stage to save the data in the default "200" > >> > partitions. So there should be ONLY 2 stages enough for this case. Why > >> > there > >> > are still 3 stages in this case, just same as "SortMergeJoin", it looks > >> > like > >> > "broadcast" not taking effect at all? But the physical plan clearly shows > >> > that "Broadcast" hint? > >> > > >> > Thanks > >> > > >> > Yong > >> > > >> > > >> > ________________________________ > >> > From: java8...@hotmail.com > >> > To: user@spark.apache.org > >> > Subject: Spark 1.5.2, why the broadcast join shuffle so much data in the > >> > last step > >> > Date: Fri, 18 Mar 2016 16:54:16 -0400 > >> > > >> > Hi, Sparkers: > >> > > >> > I have some questions related to generate the parquet output in Spark > >> > 1.5.2. > >> > > >> > I have 2 data sets to join, and I know one is much smaller than the > >> > other one, so I have the following test code: > >> > > >> > val loadRaw = sqlContext.read.parquet("one days of data in parquet > >> > format") > >> > val historyRaw = sqlContext.read.parquet("90 days of history data in > >> > parquet format") > >> > > >> > // the trailRaw will be very small, normally only thousands of row from > >> > 20M of one day's data > >> > val trialRaw = loadRaw.filter(instr(loadRaw("event_list"), "202") > > >> > 0).selectExpr("e1 as account_id", "visid_high", "visid_low", "ip") > >> > > >> > trialRaw.count > >> > res0: Long = 1494 > >> > > >> > // so the trailRaw data is small > >> > > >> > val join = historyRaw.join(broadcast(trialRaw), trialRaw("visid_high") > >> > === historyRaw("visid_high") && trialRaw("visid_low") === > >> > historyRaw("visid_low") && trialRaw("date_time") > > >> > historyRaw("date_time")) > >> > > >> > val col_1 = trialRaw("visid_high") > >> > val col_2 = trialRaw("visid_low") > >> > val col_3 = trialRaw("date_time") > >> > val col_4 = trialRaw("ip") > >> > > >> > // drop the duplicate columns after join > >> > val output = join.drop(col1).drop(col2).drop(col3).drop(col4) > >> > output.write.parquet("hdfs location") > >> > > >> > First problem, I think I am facing Spark-10309 > >> > > >> > Caused by: java.io.IOException: Unable to acquire 67108864 bytes of > >> > memory > >> > at > >> > org.apache.spark.util.collection.unsafe.sort.UnsafeExternalSorter.acquireNewPage(UnsafeExternalSorter.java:351) > >> > at > >> > org.apache.spark.util.collection.unsafe.sort.UnsafeExternalSorter.<init>(UnsafeExternalSorter.java:138) > >> > at > >> > org.apache.spark.util.collection.unsafe.sort.UnsafeExternalSorter.create(UnsafeExternalSorter.java:106) > >> > > >> > > >> > so I have to disable tungsten (spark.sql.tungsten.enabled=false), > >> > > >> > Now the problem is the Spark finishes this job very slow, even worse > >> > than same logic done in Hive. > >> > The explain shows the broadcast join is used: > >> > join.explain(true) > >> > > >> > ..... > >> > == Physical Plan == > >> > Filter (date_time#25L > date_time#519L) > >> > BroadcastHashJoin [visid_high#954L,visid_low#955L], > >> > [visid_high#460L,visid_low#461L], BuildRight > >> > ConvertToUnsafe > >> > Scan ParquetRelation[hdfs://xxxxxx][400+ columns shown up here] > >> > ConvertToUnsafe > >> > Project [soid_e1#30 AS > >> > account_id#488,visid_high#460L,visid_low#461L,date_time#25L,ip#127] > >> > Filter (instr(event_list#105,202) > 0) > >> > Scan > >> > ParquetRelation[hdfs:xxx/data/event_parquet/2016/03/17][visid_high#460L,ip#127,visid_low#461L,date_time#25L,event_list#105,soid_e1#30] > >> > Code Generation: true > >> > > >> > I don't understand the statistics shown in the GUI below: > >> > > >> > > >> > > >> > It looks like the last task will shuffle read all 506.6G data, but this > >> > DOESN'T make any sense. The final output of 200 files shown below: > >> > > >> > hadoop fs -ls hdfs://finalPath | sort -u -k5n > >> > Found 203 items > >> > -rw-r--r-- 3 biginetl biginetl 44237 2016-03-18 16:47 > >> > finalPath/_common_metadata > >> > -rw-r--r-- 3 biginetl biginetl 105534 2016-03-18 15:45 > >> > finalPath/part-r-00069-c534cd56-64b5-4ec8-8ba9-1514791f05ab.snappy.parquet > >> > -rw-r--r-- 3 biginetl biginetl 107098 2016-03-18 16:24 > >> > finalPath/part-r-00177-c534cd56-64b5-4ec8-8ba9-1514791f05ab.snappy.parquet > >> > ............. > >> > -rw-r--r-- 3 biginetl biginetl 1031400 2016-03-18 16:35 > >> > finalPath/part-r-00187-c534cd56-64b5-4ec8-8ba9-1514791f05ab.snappy.parquet > >> > -rw-r--r-- 3 biginetl biginetl 1173678 2016-03-18 16:21 > >> > finalPath/part-r-00120-c534cd56-64b5-4ec8-8ba9-1514791f05ab.snappy.parquet > >> > -rw-r--r-- 3 biginetl biginetl 12257423 2016-03-18 16:47 > >> > finalPath/_metadata > >> > > >> > As we can see, the largest file is only 1.1M, so the total output is > >> > just about 150M for all 200 files. > >> > I really don't understand why stage 5 is so slow, and why the shuffle > >> > read is so BIG. > >> > Understanding the "broadcast" join in Spark 1.5 is very important for > >> > our use case, Please tell me what could the reasons behind this. > >> > > >> > Thanks > >> > > >> > Yong > >> > > >> > >> --------------------------------------------------------------------- > >> To unsubscribe, e-mail: user-unsubscr...@spark.apache.org > >> For additional commands, e-mail: user-h...@spark.apache.org > >> > > --------------------------------------------------------------------- > To unsubscribe, e-mail: user-unsubscr...@spark.apache.org > For additional commands, e-mail: user-h...@spark.apache.org >
