La comparaison des colonnes dans Pyspark

Je suis en train de travailler sur un PySpark DataFrame n colonnes. J'ai un ensemble de m colonnes (m < n) et ma tâche est de choisir la colonne avec les valeurs max.

Par exemple:

D'entrée: PySpark DataFrame contenant :

col_1 = [1,2,3], col_2 = [2,1,4], col_3 = [3,2,5]

De sortie :

col_4 = max(col1, col_2, col_3) = [3,2,5]

Il y a quelque chose de semblable dans les pandas comme expliqué dans cette question.

Est-il possible de faire cela dans PySpark ou dois-je changer de convertir mon PySpark df pour les Pandas df et ensuite effectuer les opérations?

si la question est au sujet d'obtenir la valeur maximale de chaque colonne, il semblerait que la sortie attendue devrait être [max(col_1), max(col_2), max(col_3)] = [3, 4, 5]

OriginalL'auteur Hemant | 2016-06-07

  1. 13

    Vous pouvez réduire des expressions SQL sur une liste de colonnes:

    from pyspark.sql.functions import max as max_, col, when
from functools import reduce

def row_max(*cols):
    return reduce(
        lambda x, y: when(x > y, x).otherwise(y),
        [col(c) if isinstance(c, str) else c for c in cols]
    )

df = (sc.parallelize([(1, 2, 3), (2, 1, 2), (3, 4, 5)])
    .toDF(["a", "b", "c"]))

df.select(row_max("a", "b", "c").alias("max")))

    Étincelle 1.5+ fournit également least, greatest

    from pyspark.sql.functions import greatest

df.select(greatest("a", "b", "c"))

    Si vous souhaitez conserver le nom de max vous pouvez utiliser `des structures:

    from pyspark.sql.functions import struct, lit

def row_max_with_name(*cols):
    cols_ = [struct(col(c).alias("value"), lit(c).alias("col")) for c in cols]
    return greatest(*cols_).alias("greatest({0})".format(",".join(cols)))

 maxs = df.select(row_max_with_name("a", "b", "c").alias("maxs"))

    Et enfin, vous pouvez utiliser ci-dessus pour trouver sélectionnez "top" de la colonne:

    from pyspark.sql.functions import max

((_, c), ) = (maxs
    .groupBy(col("maxs")["col"].alias("col"))
    .count()
    .agg(max(struct(col("count"), col("col"))))
    .first())

df.select(c)

    OriginalL'auteur zero323

  2. 2

    Vous pouvez également utiliser le pyspark intégré moins:

    from pyspark.sql.functions import least, col
df = df.withColumn('min', least(col('c1'), col('c2'), col('c3')))

    OriginalL'auteur mattexx

  3. 0

    Un autre moyen simple de le faire. Disons que la ci-dessous df est votre dataframe

    df = sc.parallelize([(10, 10, 1 ), (200, 2, 20), (3, 30, 300), (400, 40, 4)]).toDF(["c1", "c2", "c3"])
df.show()

+---+---+---+
| c1| c2| c3|
+---+---+---+
| 10| 10|  1|
|200|  2| 20|
|  3| 30|300|
|400| 40|  4|
+---+---+---+

    Vous pouvez traiter le au-dessus de df comme ci-dessous pour obtenir le desited résultats

    from pyspark.sql.functions import lit, min

df.select( lit('c1').alias('cn1'), min(df.c1).alias('c1'),
           lit('c2').alias('cn2'), min(df.c2).alias('c2'),
           lit('c3').alias('cn3'), min(df.c3).alias('c3')
          )\
         .rdd.flatMap(lambda r: [ (r.cn1, r.c1), (r.cn2, r.c2), (r.cn3, r.c3)])\
         .toDF(['Columnn', 'Min']).show()

+-------+---+
|Columnn|Min|
+-------+---+
|     c1|  3|
|     c2|  2|
|     c3|  1|
+-------+---+
    Vous faites min(col1), alors que je veux min(row1), min(row2).. et ainsi de suite...

    OriginalL'auteur Rags

  4. 0

    Scala solution:

    df = sc.parallelize(Seq((10, 10, 1 ), (200, 2, 20), (3, 30, 300), (400, 40, 4))).toDF("c1", "c2", "c3"))  

df.rdd.map(row=>List[String](row(0).toString,row(1).toString,row(2).toString)).map(x=>(x(0),x(1),x(2),x.min)).toDF("c1","c2","c3","min").show    

    +---+---+---+---+  
| c1| c2| c3|min|  
+---+---+---+---+  
| 10| 10|  1|  1|    
|200|  2| 20|  2|  
|  3| 30|300|  3|  
|400| 40|  4|  4|  
+---+---+---+---+

    OriginalL'auteur Hareesh Adukkadukkam