How many rows are you joining? How many rows in the output?
Regards
Sab
On 24-Oct-2015 2:32 am, "pratik khadloya" <tispra...@gmail.com> wrote:
> Actually the groupBy is not taking a lot of time.
> The join that i do later takes the most (95 %) amount of time.
> Also, the grouping i am doing is based on the DataFrame api, which does
> not contain any function for reduceBy... i guess the DF automatically uses
> reduce by when we do a group by.
>
> ~Pratik
>
> On Fri, Oct 23, 2015 at 1:38 PM Kartik Mathur <kar...@bluedata.com> wrote:
>
>> Don't use groupBy , use reduceByKey instead , groupBy should always be
>> avoided as it leads to lot of shuffle reads/writes.
>>
>> On Fri, Oct 23, 2015 at 11:39 AM, pratik khadloya <tispra...@gmail.com>
>> wrote:
>>
>>> Sorry i sent the wrong join code snippet, the actual snippet is
>>>
>>> ggImpsDf.join(
>>> aggRevenueDf,
>>> aggImpsDf("id_1") <=> aggRevenueDf("id_1")
>>> && aggImpsDf("id_2") <=> aggRevenueDf("id_2")
>>> && aggImpsDf("day_hour") <=> aggRevenueDf("day_hour")
>>> && aggImpsDf("day_hour_2") <=> aggRevenueDf("day_hour_2"),
>>> "inner")
>>> .select(
>>> aggImpsDf("id_1"), aggImpsDf("id_2"), aggImpsDf("day_hour"),
>>> aggImpsDf("day_hour_2"), aggImpsDf("metric1"),
>>> aggRevenueDf("metric2"))
>>> .coalesce(200)
>>>
>>>
>>> On Fri, Oct 23, 2015 at 11:16 AM pratik khadloya <tispra...@gmail.com>
>>> wrote:
>>>
>>>> Hello,
>>>>
>>>> Data about my spark job is below. My source data is only 916MB (stage
>>>> 0) and 231MB (stage 1), but when i join the two data sets (stage 2) it
>>>> takes a very long time and as i see the shuffled data is 614GB. Is this
>>>> something expected? Both the data sets produce 200 partitions.
>>>>
>>>> Stage IdDescriptionSubmittedDurationTasks:
>>>> Succeeded/TotalInputOutputShuffle
>>>> ReadShuffle Write2saveAsTable at Driver.scala:269
>>>> <http://sparkhs.rfiserve.net:18080/history/application_1437606252645_1034031/stages/stage?id=2&attempt=0>
>>>> +details
>>>>
>>>> 2015/10/22 18:48:122.3 h
>>>> 200/200
>>>> 614.6 GB1saveAsTable at Driver.scala:269
>>>> <http://sparkhs.rfiserve.net:18080/history/application_1437606252645_1034031/stages/stage?id=1&attempt=0>
>>>> +details
>>>>
>>>> 2015/10/22 18:46:022.1 min
>>>> 8/8
>>>> 916.2 MB3.9 MB0saveAsTable at Driver.scala:269
>>>> <http://sparkhs.rfiserve.net:18080/history/application_1437606252645_1034031/stages/stage?id=0&attempt=0>
>>>> +details
>>>>
>>>> 2015/10/22 18:46:0235 s
>>>> 3/3
>>>> 231.2 MB4.8 MBAm running Spark 1.4.1 and my code snippet which joins
>>>> the two data sets is:
>>>>
>>>> hc.sql(query).
>>>> mapPartitions(iter => {
>>>> iter.map {
>>>> case Row(
>>>> ...
>>>> ...
>>>> ...
>>>> )
>>>> }
>>>> }
>>>> ).toDF()
>>>> .groupBy("id_1", "id_2", "day_hour", "day_hour_2")
>>>> .agg($"id_1", $"id_2", $"day_hour", $"day_hour_2",
>>>> sum("attr1").alias("attr1"), sum("attr2").alias("attr2"))
>>>>
>>>>
>>>> Please advise on how to reduce the shuffle and speed this up.
>>>>
>>>>
>>>> ~Pratik
>>>>
>>>>
>>
