将聚合列添加到无需连接的数据透视表答案

【问题标题】：Add aggregated columns to pivot without join将聚合列添加到无需连接的数据透视表
【发布时间】：2019-02-28 09:14:51
【问题描述】：

考虑表格：

df=sc.parallelize([(1,1,1),(5,0,2),(27,1,1),(1,0,3),(5,1,1),(1,0,2)]).toDF(['id', 'error', 'timestamp'])
df.show()

+---+-----+---------+
| id|error|timestamp|
+---+-----+---------+
|  1|    1|        1|
|  5|    0|        2|
| 27|    1|        1|
|  1|    0|        3|
|  5|    1|        1|
|  1|    0|        2|
+---+-----+---------+

我想在timestamp 列上进行透视，保留原始表中的一些其他汇总信息。我感兴趣的结果可以通过

df1=df.groupBy('id').agg(sf.sum('error').alias('Ne'),sf.count('*').alias('cnt'))
df2=df.groupBy('id').pivot('timestamp').agg(sf.count('*')).fillna(0)
df1.join(df2, on='id').filter(sf.col('cnt')>1).show()

使用结果表：

+---+---+---+---+---+---+
| id| Ne|cnt|  1|  2|  3|
+---+---+---+---+---+---+
|  5|  1|  2|  1|  1|  0|
|  1|  1|  3|  1|  1|  1|
+---+---+---+---+---+---+

但是，上述解决方案至少存在两个问题：

我在脚本末尾按cnt 过滤。如果我一开始就可以做到这一点，我可以避免几乎所有的处理，因为使用这种过滤会删除大部分数据。除了collect 和isin 方法之外，还有什么方法可以做到这一点？
我在id 上做了两次groupBy。首先，聚合我在结果中需要的一些列，第二次获取数据透视列。最后，我需要join 来合并这些列。我觉得我肯定错过了一些解决方案，因为应该可以只使用一个 groubBy 而没有 join，但我不知道该怎么做。

【问题讨论】：

您可以将过滤器移动到第一个 groupBy 查询，并且只过滤 df1 上的 cnt>1。这不应该改变你的结果

标签： dataframe group-by pyspark pivot-table

【解决方案1】：

我认为您无法绕过联接，因为枢轴将需要时间戳值，而第一个分组不应考虑它们。因此，如果您必须创建 NE 和 cnt 值，则必须仅按 id 对数据框进行分组，这会导致时间戳丢失，如果您想保留列中的值，则必须像您一样进行数据透视分开做，然后再加入。

唯一可以做的改进是将过滤器移动到 df1 创建。因此，正如您所说，这已经可以提高性能，因为 df1 在过滤您的真实数据后应该小得多。

from pyspark.sql.functions import *

df=sc.parallelize([(1,1,1),(5,0,2),(27,1,1),(1,0,3),(5,1,1),(1,0,2)]).toDF(['id', 'error', 'timestamp'])
df1=df.groupBy('id').agg(sum('error').alias('Ne'),count('*').alias('cnt')).filter(col('cnt')>1)
df2=df.groupBy('id').pivot('timestamp').agg(count('*')).fillna(0)
df1.join(df2, on='id').show()

输出：

+---+---+---+---+---+---+
| id| Ne|cnt|  1|  2|  3|
+---+---+---+---+---+---+
|  5|  1|  2|  1|  1|  0|
|  1|  1|  3|  1|  1|  1|
+---+---+---+---+---+---+

【讨论】：

【解决方案2】：

其实使用Window as 确实可以避免join

w1 = Window.partitionBy('id')
w2 = Window.partitionBy('id', 'timestamp')
df.select('id', 'timestamp', 
          sf.sum('error').over(w1).alias('Ne'), 
          sf.count('*').over(w1).alias('cnt'),
          sf.count('*').over(w2).alias('cnt_2')
         ).filter(sf.col('cnt')>1) \
  .groupBy('id', 'Ne', 'cnt').pivot('timestamp').agg(sf.first('cnt_2')).fillna(0).show()

【讨论】：