我需要对我的PySpark代码中的大型数据集进行大量聚合(大约9-10)。我可以通过2种方法来解决:

分组依据:

df.groupBy(col1, col2).agg({"col3":"sum", "col4":"avg", "col5":"min", "col6":"sum", "col7":"max", "col8":"avg", "col9":"sum"})


分组并加入

temp1 = df.groupBy(col1, col2).agg({"col3":"sum"})
temp2 = df.groupBy(col1, col2).agg({"col4":"avg"})
temp3 = df.groupBy(col1, col2).agg({"col5":"min"})
.
.
.
temp9 = df.groupBy(col1, col2).agg({"col9":"sum"})


然后将所有这9个数据帧结合起来以得到最终输出。

哪一个效率更高?

最佳答案

TL; DR请选择第一个。

这甚至不是比赛。仅可读性就足以拒绝第二种解决方案,该解决方案冗长且令人费解。

更不用说执行计划只是一个怪兽(这里只有2个表!):

== Physical Plan ==
*Project [col1#512L, col2#513L, sum(col3)#597L, avg(col4)#614, min(col5)#631L]
+- *SortMergeJoin [col1#512L, col2#513L], [col1#719L, col2#720L], Inner
   :- *Project [col1#512L, col2#513L, sum(col3)#597L, avg(col4)#614]
   :  +- *SortMergeJoin [col1#512L, col2#513L], [col1#704L, col2#705L], Inner
   :     :- *Sort [col1#512L ASC NULLS FIRST, col2#513L ASC NULLS FIRST], false, 0
   :     :  +- *HashAggregate(keys=[col1#512L, col2#513L], functions=[sum(col3#514L)])
   :     :     +- Exchange hashpartitioning(col1#512L, col2#513L, 200)
   :     :        +- *HashAggregate(keys=[col1#512L, col2#513L], functions=[partial_sum(col3#514L)])
   :     :           +- *Project [_1#491L AS col1#512L, _2#492L AS col2#513L, _3#493L AS col3#514L]
   :     :              +- *Filter (isnotnull(_1#491L) && isnotnull(_2#492L))
   :     :                 +- Scan ExistingRDD[_1#491L,_2#492L,_3#493L,_4#494L,_5#495L,_6#496L,_7#497L,_8#498L,_9#499L,_10#500L]
   :     +- *Sort [col1#704L ASC NULLS FIRST, col2#705L ASC NULLS FIRST], false, 0
   :        +- *HashAggregate(keys=[col1#704L, col2#705L], functions=[avg(col4#707L)])
   :           +- Exchange hashpartitioning(col1#704L, col2#705L, 200)
   :              +- *HashAggregate(keys=[col1#704L, col2#705L], functions=[partial_avg(col4#707L)])
   :                 +- *Project [_1#491L AS col1#704L, _2#492L AS col2#705L, _4#494L AS col4#707L]
   :                    +- *Filter (isnotnull(_2#492L) && isnotnull(_1#491L))
   :                       +- Scan ExistingRDD[_1#491L,_2#492L,_3#493L,_4#494L,_5#495L,_6#496L,_7#497L,_8#498L,_9#499L,_10#500L]
   +- *Sort [col1#719L ASC NULLS FIRST, col2#720L ASC NULLS FIRST], false, 0
      +- *HashAggregate(keys=[col1#719L, col2#720L], functions=[min(col5#723L)])
         +- Exchange hashpartitioning(col1#719L, col2#720L, 200)
            +- *HashAggregate(keys=[col1#719L, col2#720L], functions=[partial_min(col5#723L)])
               +- *Project [_1#491L AS col1#719L, _2#492L AS col2#720L, _5#495L AS col5#723L]
                  +- *Filter (isnotnull(_1#491L) && isnotnull(_2#492L))
                     +- Scan ExistingRDD[_1#491L,_2#492L,_3#493L,_4#494L,_5#495L,_6#496L,_7#497L,_8#498L,_9#499L,_10#500L]


与普通聚合(对于所有列)相比:

== Physical Plan ==
*HashAggregate(keys=[col1#512L, col2#513L], functions=[max(col7#518L), avg(col8#519L), sum(col3#514L), sum(col6#517L), sum(col9#520L), min(col5#516L), avg(col4#515L)])
+- Exchange hashpartitioning(col1#512L, col2#513L, 200)
   +- *HashAggregate(keys=[col1#512L, col2#513L], functions=[partial_max(col7#518L), partial_avg(col8#519L), partial_sum(col3#514L), partial_sum(col6#517L), partial_sum(col9#520L), partial_min(col5#516L), partial_avg(col4#515L)])
      +- *Project [_1#491L AS col1#512L, _2#492L AS col2#513L, _3#493L AS col3#514L, _4#494L AS col4#515L, _5#495L AS col5#516L, _6#496L AS col6#517L, _7#497L AS col7#518L, _8#498L AS col8#519L, _9#499L AS col9#520L]
         +- Scan ExistingRDD[_1#491L,_2#492L,_3#493L,_4#494L,_5#495L,_6#496L,_7#497L,_8#498L,_9#499L,_10#500L]

关于performance - 所有聚合是在单个组中还是单独进行?,我们在Stack Overflow上找到一个类似的问题:https://stackoverflow.com/questions/48444046/

10-14 19:06
查看更多