GroupBy 一个数据框记录并使用 PySpark 显示所有列答案

【问题标题】：GroupBy a dataframe records and display all columns with PySparkGroupBy 一个数据框记录并使用 PySpark 显示所有列
【发布时间】：2021-07-21 16:55:31
【问题描述】：

我有以下数据框

dataframe - columnA, columnB, columnC, columnD, columnE

我想按 columnC 分组，然后考虑 columnE 的最大值

dataframe .select('*').groupBy('columnC').max('columnE')

预期输出

dataframe - columnA, columnB, columnC, columnD, columnE

实际输出

dataframe - columnC, columnE

为什么数据框中的所有列都没有按预期显示？

【问题讨论】：

当您只聚合每个组中一列的数据时，为什么希望显示所有列？
@It_is_Chris 因为根据我的理解 groupBy 和聚合 max() 函数的行为与 MYSQL 相同。我想按 columnC 对数据进行分组，然后从该结果中获得最大 colunE 的行

标签： python dataframe apache-spark pyspark group-by

【解决方案1】：

对于 Spark 版本 >= 3.0.0，您可以使用 max_by 选择其他列。

import random
from pyspark.sql import functions as F

#create some testdata
df = spark.createDataFrame(
  [[random.randint(1,3)] + random.sample(range(0, 30), 4)  for _ in range(10)], 
  schema=["columnC", "columnB", "columnA", "columnD", "columnE"]) \
  .select("columnA", "columnB", "columnC", "columnD", "columnE")

df.groupBy("columnC") \
  .agg(F.max("columnE"), 
      F.expr("max_by(columnA, columnE) as columnA"),
      F.expr("max_by(columnB, columnE) as columnB"),
      F.expr("max_by(columnD, columnE) as columnD")) \
  .show()

对于测试数据

+-------+-------+-------+-------+-------+
|columnA|columnB|columnC|columnD|columnE|
+-------+-------+-------+-------+-------+
|     25|     20|      2|      0|      2|
|     14|      2|      2|     24|      6|
|     26|     13|      3|      2|      1|
|      5|     24|      3|     19|     17|
|     22|      5|      3|     14|     21|
|     24|      5|      1|      8|      4|
|      7|     22|      3|     16|     20|
|      6|     17|      1|      5|      7|
|     24|     22|      2|      8|      3|
|      4|     14|      1|     16|     11|
+-------+-------+-------+-------+-------+

结果是

+-------+------------+-------+-------+-------+
|columnC|max(columnE)|columnA|columnB|columnD|
+-------+------------+-------+-------+-------+
|      1|          11|      4|     14|     16|
|      3|          21|     22|      5|     14|
|      2|           6|     14|      2|     24|
+-------+------------+-------+-------+-------+

【讨论】：

即使columnA,columnB,columnD有字符串值，这样用对吗？

【解决方案2】：

你想达到的可以通过WINDOW函数来完成。不是分组方式

按列对数据进行分区C
按 desc (rank) 对每个分区中的数据进行排序
过滤掉你想要的结果。

from pyspark.sql.window import Window
from pyspark.sql.functions import rank
from pyspark.sql.functions import col

windowSpec  = Window.partitionBy("columnC").orderBy(col("columnE").desc())

expectedDf = df.withColumn("rank", rank().over(windowSpec)) \
    .filter(col("rank") == 1)

您可能想重新构建您的问题。

【讨论】：