For the code below:
data1 = [(i,random.randint(1,5),random.randint(1,5)) for t in range(2) for i in range(3)]
df1=spark.createDataFrame(data1,schema = 'a int,b int,c int')
df1 = df1.repartition(4,'a')
df1.groupby('a').agg(F.max(col('b'))).explain()
the explain output is:
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- HashAggregate(keys=[a#3629], functions=[max(b#3630)])
+- HashAggregate(keys=[a#3629], functions=[partial_max(b#3630)])
+- Exchange hashpartitioning(a#3629, 4), REPARTITION_BY_NUM, [id=#5232]
+- Project [a#3629, b#3630]
+- Scan ExistingRDD[a#3629,b#3630,c#3631]
Why do we have HashAggregate(keys=[a#3629], functions=[partial_max(b#3630)]) in the physical plan? Shouldn't partial_max run on the map side? Essentially, groupby would compute partial sums and then all partial sums are collected and summed for each key, that is if the dataframe is not repartitioned. But, since I have repartitioned the data, what is the need for partial sums?