You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@spark.apache.org by Yong Zhang <ja...@hotmail.com> on 2016/03/18 21:54:16 UTC
Spark 1.5.2, why the broadcast join shuffle so much data in the
last step
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 dataval trialRaw = loadRaw.filter(instr(loadRaw("event_list"), "202") > 0).selectExpr("e1 as account_id", "visid_high", "visid_low", "ip")
trialRaw.countres0: 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 joinval output = join.drop(col1).drop(col2).drop(col3).drop(col4)output.write.parquet("hdfs location")
First problem, I think I am facing Spark-10309Caused 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 -k5nFound 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
RE: Spark 1.5.2, why the broadcast join shuffle so much data in the
last step
Posted by Yong Zhang <ja...@hotmail.com>.
I didn't change that. So it will be the default one.
I can confirm from the environment later.
Yong
Date: Fri, 18 Mar 2016 16:18:26 -0700
Subject: Re: Spark 1.5.2, why the broadcast join shuffle so much data in the last step
From: yuzhihong@gmail.com
To: java8964@hotmail.com
CC: user@spark.apache.org
What's the value for spark.sql.autoBroadcastJoinThreshold ?
Default is 10MB.
FYI
On Fri, Mar 18, 2016 at 4:13 PM, Yong Zhang <ja...@hotmail.com> wrote:
This is our production cluster, and we just upgrade from 1.3.1 to 1.5.2, so upgrading again is not easy for us as right now.
We do observe some other jobs performance is much better on 1.5.2, comparing with 1.3.1.
I am very excited of "broadcast" join of DF in 1.5.x, as we have several cases can utilize it, but right now, it doesn't work as I expect.
BTW, if I disable "tungsten", is it back to the default "sort shuffle"?
This case shouldn't have much of "shuffling data", as most of data will be filtered out in the mapper stage.
In fact, running in Hive using "map-side join" can be finished quite fast. I just want to use this case to test "broadcast" join of DF in 1.5.2.
Yong
Date: Fri, 18 Mar 2016 13:57:43 -0700
Subject: Re: Spark 1.5.2, why the broadcast join shuffle so much data in the last step
From: yuzhihong@gmail.com
To: java8964@hotmail.com
CC: user@spark.apache.org
bq. so I have to disable tungsten
Can you try 1.6.1 release so that you don't have to disable tungsten ?
Thanks
On Fri, Mar 18, 2016 at 1:54 PM, Yong Zhang <ja...@hotmail.com> wrote:
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 dataval trialRaw = loadRaw.filter(instr(loadRaw("event_list"), "202") > 0).selectExpr("e1 as account_id", "visid_high", "visid_low", "ip")
trialRaw.countres0: 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 joinval output = join.drop(col1).drop(col2).drop(col3).drop(col4)output.write.parquet("hdfs location")
First problem, I think I am facing Spark-10309Caused 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 -k5nFound 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
Re: Spark 1.5.2, why the broadcast join shuffle so much data in the
last step
Posted by Ted Yu <yu...@gmail.com>.
What's the value for spark.sql.autoBroadcastJoinThreshold ?
Default is 10MB.
FYI
On Fri, Mar 18, 2016 at 4:13 PM, Yong Zhang <ja...@hotmail.com> wrote:
> This is our production cluster, and we just upgrade from 1.3.1 to 1.5.2,
> so upgrading again is not easy for us as right now.
>
> We do observe some other jobs performance is much better on 1.5.2,
> comparing with 1.3.1.
>
> I am very excited of "broadcast" join of DF in 1.5.x, as we have several
> cases can utilize it, but right now, it doesn't work as I expect.
>
> BTW, if I disable "tungsten", is it back to the default "sort shuffle"?
>
> This case shouldn't have much of "shuffling data", as most of data will be
> filtered out in the mapper stage.
>
> In fact, running in Hive using "map-side join" can be finished quite fast.
> I just want to use this case to test "broadcast" join of DF in 1.5.2.
>
> Yong
>
> ------------------------------
> Date: Fri, 18 Mar 2016 13:57:43 -0700
> Subject: Re: Spark 1.5.2, why the broadcast join shuffle so much data in
> the last step
> From: yuzhihong@gmail.com
> To: java8964@hotmail.com
> CC: user@spark.apache.org
>
>
> bq. so I have to disable tungsten
>
> Can you try 1.6.1 release so that you don't have to disable tungsten ?
>
> Thanks
>
> On Fri, Mar 18, 2016 at 1:54 PM, Yong Zhang <ja...@hotmail.com> wrote:
>
> 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
> <https://issues.apache.org/jira/browse/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
>
>
>
RE: Spark 1.5.2, why the broadcast join shuffle so much data in the
last step
Posted by Yong Zhang <ja...@hotmail.com>.
This is our production cluster, and we just upgrade from 1.3.1 to 1.5.2, so upgrading again is not easy for us as right now.
We do observe some other jobs performance is much better on 1.5.2, comparing with 1.3.1.
I am very excited of "broadcast" join of DF in 1.5.x, as we have several cases can utilize it, but right now, it doesn't work as I expect.
BTW, if I disable "tungsten", is it back to the default "sort shuffle"?
This case shouldn't have much of "shuffling data", as most of data will be filtered out in the mapper stage.
In fact, running in Hive using "map-side join" can be finished quite fast. I just want to use this case to test "broadcast" join of DF in 1.5.2.
Yong
Date: Fri, 18 Mar 2016 13:57:43 -0700
Subject: Re: Spark 1.5.2, why the broadcast join shuffle so much data in the last step
From: yuzhihong@gmail.com
To: java8964@hotmail.com
CC: user@spark.apache.org
bq. so I have to disable tungsten
Can you try 1.6.1 release so that you don't have to disable tungsten ?
Thanks
On Fri, Mar 18, 2016 at 1:54 PM, Yong Zhang <ja...@hotmail.com> wrote:
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 dataval trialRaw = loadRaw.filter(instr(loadRaw("event_list"), "202") > 0).selectExpr("e1 as account_id", "visid_high", "visid_low", "ip")
trialRaw.countres0: 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 joinval output = join.drop(col1).drop(col2).drop(col3).drop(col4)output.write.parquet("hdfs location")
First problem, I think I am facing Spark-10309Caused 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 -k5nFound 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
Re: Spark 1.5.2, why the broadcast join shuffle so much data in the
last step
Posted by Ted Yu <yu...@gmail.com>.
bq. so I have to disable tungsten
Can you try 1.6.1 release so that you don't have to disable tungsten ?
Thanks
On Fri, Mar 18, 2016 at 1:54 PM, Yong Zhang <ja...@hotmail.com> wrote:
> 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
> <https://issues.apache.org/jira/browse/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
>
>
RE: Spark 1.5.2, why the broadcast join shuffle so much data in the
last step
Posted by Yong Zhang <ja...@hotmail.com>.
So I have one more question related to this test.
The patch works great, but the broadcast will generate lots of small parquet files in the end, which I want to see if I can avoid.
So I change the code to:
val output = join.drop(col_1).drop(col_2).drop(col_3).drop(col_4).coalesce(1)I added coalesce(1), to force the final output to be a single parquet file.
But after this change, I found out the stage 4 will become a only one task stage, which is not as good as before:
So my questions are following:
1) In this case, the small data is broadcasted, but I still want large dataset filtering by joining with small dataset as parallelism as possible2) In the end, the final output (since they are small), to merge by single reducer, and save into HDFS as one parquet file.
Is there a way to archive this?
Thanks
Yong
From: java8964@hotmail.com
To: davies@databricks.com
CC: user@spark.apache.org
Subject: RE: Spark 1.5.2, why the broadcast join shuffle so much data in the last step
Date: Wed, 23 Mar 2016 20:30:42 -0400
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: davies@databricks.com
> To: java8964@hotmail.com
> CC: user@spark.apache.org
>
> On Wed, Mar 23, 2016 at 10:35 AM, Yong Zhang <ja...@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: davies@databricks.com
> >> To: java8964@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 <ja...@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: java8964@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: java8964@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-unsubscribe@spark.apache.org
> >> For additional commands, e-mail: user-help@spark.apache.org
> >>
>
> ---------------------------------------------------------------------
> To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
> For additional commands, e-mail: user-help@spark.apache.org
>
RE: Spark 1.5.2, why the broadcast join shuffle so much data in the
last step
Posted by Yong Zhang <ja...@hotmail.com>.
The patch works as I expect.
The DAG shows the broadcast joining in stage 4, eliminating following stages, and data generated right after it (There are no shuffle write in this stage any more).Much faster than before.
If you like this patch back port to 1.5 and 1.6, please vote in JIRA SPARK-13383
Thanks
Yong
From: java8964@hotmail.com
To: davies@databricks.com
CC: user@spark.apache.org
Subject: RE: Spark 1.5.2, why the broadcast join shuffle so much data in the last step
Date: Wed, 23 Mar 2016 20:30:42 -0400
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: davies@databricks.com
> To: java8964@hotmail.com
> CC: user@spark.apache.org
>
> On Wed, Mar 23, 2016 at 10:35 AM, Yong Zhang <ja...@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: davies@databricks.com
> >> To: java8964@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 <ja...@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: java8964@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: java8964@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-unsubscribe@spark.apache.org
> >> For additional commands, e-mail: user-help@spark.apache.org
> >>
>
> ---------------------------------------------------------------------
> To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
> For additional commands, e-mail: user-help@spark.apache.org
>
RE: Spark 1.5.2, why the broadcast join shuffle so much data in the
last step
Posted by Yong Zhang <ja...@hotmail.com>.
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: davies@databricks.com
> To: java8964@hotmail.com
> CC: user@spark.apache.org
>
> On Wed, Mar 23, 2016 at 10:35 AM, Yong Zhang <ja...@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: davies@databricks.com
> >> To: java8964@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 <ja...@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: java8964@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: java8964@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-unsubscribe@spark.apache.org
> >> For additional commands, e-mail: user-help@spark.apache.org
> >>
>
> ---------------------------------------------------------------------
> To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
> For additional commands, e-mail: user-help@spark.apache.org
>
Re: Spark 1.5.2, why the broadcast join shuffle so much data in the
last step
Posted by Ted Yu <yu...@gmail.com>.
SPARK-13383 is fixed in 2.0 only, as of this moment.
Any chance of backporting to branch-1.6 ?
Thanks
On Wed, Mar 23, 2016 at 4:20 PM, Davies Liu <da...@databricks.com> wrote:
> On Wed, Mar 23, 2016 at 10:35 AM, Yong Zhang <ja...@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: davies@databricks.com
> >> To: java8964@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 <ja...@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: java8964@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: java8964@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-unsubscribe@spark.apache.org
> >> For additional commands, e-mail: user-help@spark.apache.org
> >>
>
> ---------------------------------------------------------------------
> To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
> For additional commands, e-mail: user-help@spark.apache.org
>
>
Re: Spark 1.5.2, why the broadcast join shuffle so much data in the
last step
Posted by Davies Liu <da...@databricks.com>.
On Wed, Mar 23, 2016 at 10:35 AM, Yong Zhang <ja...@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: davies@databricks.com
>> To: java8964@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 <ja...@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: java8964@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: java8964@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-unsubscribe@spark.apache.org
>> For additional commands, e-mail: user-help@spark.apache.org
>>
---------------------------------------------------------------------
To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
For additional commands, e-mail: user-help@spark.apache.org
RE: Spark 1.5.2, why the broadcast join shuffle so much data in the
last step
Posted by Yong Zhang <ja...@hotmail.com>.
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+ columnsProject [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
== 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: davies@databricks.com
> To: java8964@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 <ja...@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: java8964@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: java8964@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-unsubscribe@spark.apache.org
> For additional commands, e-mail: user-help@spark.apache.org
>
Re: Spark 1.5.2, why the broadcast join shuffle so much data in the
last step
Posted by Davies Liu <da...@databricks.com>.
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 <ja...@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: java8964@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: java8964@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-unsubscribe@spark.apache.org
For additional commands, e-mail: user-help@spark.apache.org
RE: Spark 1.5.2, why the broadcast join shuffle so much data in the
last step
Posted by Yong Zhang <ja...@hotmail.com>.
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: java8964@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: java8964@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 dataval trialRaw = loadRaw.filter(instr(loadRaw("event_list"), "202") > 0).selectExpr("e1 as account_id", "visid_high", "visid_low", "ip")
trialRaw.countres0: 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 joinval output = join.drop(col1).drop(col2).drop(col3).drop(col4)output.write.parquet("hdfs location")
First problem, I think I am facing Spark-10309Caused 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 -k5nFound 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
RE: Spark 1.5.2, why the broadcast join shuffle so much data in the
last step
Posted by Yong Zhang <ja...@hotmail.com>.
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: java8964@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 dataval trialRaw = loadRaw.filter(instr(loadRaw("event_list"), "202") > 0).selectExpr("e1 as account_id", "visid_high", "visid_low", "ip")
trialRaw.countres0: 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 joinval output = join.drop(col1).drop(col2).drop(col3).drop(col4)output.write.parquet("hdfs location")
First problem, I think I am facing Spark-10309Caused 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 -k5nFound 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
RE: Spark 1.5.2, why the broadcast join shuffle so much data in the
last step
Posted by Yong Zhang <ja...@hotmail.com>.
Any idea how I can debug this issue on Spark 1.5.2?
Thanks
Yong
From: java8964@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 dataval trialRaw = loadRaw.filter(instr(loadRaw("event_list"), "202") > 0).selectExpr("e1 as account_id", "visid_high", "visid_low", "ip")
trialRaw.countres0: 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 joinval output = join.drop(col1).drop(col2).drop(col3).drop(col4)output.write.parquet("hdfs location")
First problem, I think I am facing Spark-10309Caused 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 -k5nFound 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