Estoy casi seguro de que esto se ha preguntado antes, pero una búsqueda a través de stackoverflow no respondió a mi pregunta. No es un duplicado de [2] ya que quiero el valor máximo, no el elemento más frecuente. Soy nuevo en pyspark y trato de hacer algo realmente simple: quiero agrupar por la columna "A" y luego solo mantener la fila de cada grupo que tiene el valor máximo en la columna "B". Me gusta esto:
df_cleaned = df.groupBy("A").agg(F.max("B"))Desafortunadamente, esto descarta todas las demás columnas: df_cleaned solo contiene las columnas "A" y el valor máximo de B. ¿Cómo mantengo las filas? ("A B C"...)
Puede hacer esto sin un udf usando una Window .
Considere el siguiente ejemplo:
import pyspark.sql.functions as f data = [ ('a', 5), ('a', 8), ('a', 7), ('b', 1), ('b', 3) ] df = sqlCtx.createDataFrame(data, ["A", "B"]) df.show() #+---+---+ #| A| B| #+---+---+ #| a| 5| #| a| 8| #| a| 7| #| b| 1| #| b| 3| #+---+---+ Cree una Window para particionar por la columna A y utilícela para calcular el máximo de cada grupo. Luego filtre las filas de modo que el valor en la columna B sea igual al máximo.
from pyspark.sql import Window w = Window.partitionBy('A') df.withColumn('maxB', f.max('B').over(w))\ .where(f.col('B') == f.col('maxB'))\ .drop('maxB')\ .show() #+---+---+ #| A| B| #+---+---+ #| a| 8| #| b| 3| #+---+---+ O de manera equivalente usando pyspark-sql :
df.registerTempTable('table') q = "SELECT A, B FROM (SELECT *, MAX(B) OVER (PARTITION BY A) AS maxB FROM table) M WHERE B = maxB" sqlCtx.sql(q).show() #+---+---+ #| A| B| #+---+---+ #| b| 3| #| a| 8| #+---+---+Otro enfoque posible es aplicar unir el marco de datos consigo mismo especificando "leftsemi". Este tipo de combinación incluye todas las columnas del marco de datos en el lado izquierdo y ninguna columna en el lado derecho.
Por ejemplo:
import pyspark.sql.functions as f data = [ ('a', 5, 'c'), ('a', 8, 'd'), ('a', 7, 'e'), ('b', 1, 'f'), ('b', 3, 'g') ] df = sqlContext.createDataFrame(data, ["A", "B", "C"]) df.show() +---+---+---+ | A| B| C| +---+---+---+ | a| 5| c| | a| 8| d| | a| 7| e| | b| 1| f| | b| 3| g| +---+---+---+El valor máximo de la columna B por la columna A se puede seleccionar haciendo:
df.groupBy('A').agg(f.max('B') +---+---+ | A| B| +---+---+ | a| 8| | b| 3| +---+---+ Usando esta expresión como lado derecho en una semiunión izquierda, y cambiando el nombre de la columna obtenida max(B) a su nombre original B , podemos obtener el resultado necesario:
df.join(df.groupBy('A').agg(f.max('B').alias('B')),on='B',how='leftsemi').show() +---+---+---+ | B| A| C| +---+---+---+ | 3| b| g| | 8| a| d| +---+---+---+El plan físico detrás de esta solución y el de la respuesta aceptada son diferentes y todavía no tengo claro cuál funcionará mejor en marcos de datos grandes.
El mismo resultado se puede obtener usando la sintaxis de Spark SQL haciendo:
df.registerTempTable('table') q = '''SELECT * FROM table a LEFT SEMI JOIN ( SELECT A, max(B) as max_B FROM table GROUP BY A ) t ON aA=tA AND aB=t.max_B ''' sqlContext.sql(q).show() +---+---+---+ | A| B| C| +---+---+---+ | b| 3| g| | a| 8| d| +---+---+---+solo quiero agregar la versión scala spark de la respuesta de @ndricca en caso de que alguien la necesite:
val data = Seq(("a", 5,"c"), ("a",8,"d"),("a",7,"e"),("b",1,"f"),("b",3,"g")) val df = data.toDF("A","B","C") df.show() +---+---+---+ | A| B| C| +---+---+---+ | a| 5| c| | a| 8| d| | a| 7| e| | b| 1| f| | b| 3| g| +---+---+---+ val rightdf = df.groupBy("A").max("B") rightdf.show() +---+------+ | A|max(B)| +---+------+ | b| 3| | a| 8| +---+------+ val resdf = df.join(rightdf, df("B") === rightdf("max(B)"), "leftsemi") resdf.show() +---+---+---+ | A| B| C| +---+---+---+ | a| 8| d| | b| 3| g| +---+---+---+