Comparing columns in Pyspark

by

You can reduce using SQL expressions over a list of columns:

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")))

Spark 1.5+ also provides least, greatest

from pyspark.sql.functions import greatest

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

If you want to keep name of the max you can use `structs:

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"))

And finally you can use above to find select “top” column:

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)

More Related Contents:

  1. Calling Java/Scala function from a task
  2. Efficient string matching in Apache Spark
  3. Load CSV file with Spark
  4. How to link PyCharm with PySpark?
  5. How to turn off INFO logging in Spark?
  6. How do I add a new column to a Spark DataFrame (using PySpark)?
  7. How to perform union on two DataFrames with different amounts of columns in spark?
  8. Count number of non-NaN entries in each column of Spark dataframe with Pyspark
  9. importing pyspark in python shell
  10. collect_list by preserving order based on another variable
  11. Rename nested field in spark dataframe
  12. Spark DataFrame: Computing row-wise mean (or any aggregate operation)
  13. How to determine if object is a valid key-value pair in PySpark
  14. PySpark converting a column of type ‘map’ to multiple columns in a dataframe
  15. A list as a key for PySpark’s reduceByKey
  16. Best way to get the max value in a Spark dataframe column
  17. How to map features from the output of a VectorAssembler back to the column names in Spark ML?
  18. Apache Spark Python Cosine Similarity over DataFrames
  19. Pyspark: Parse a column of json strings
  20. GroupBy column and filter rows with maximum value in Pyspark
  21. Add Jar to standalone pyspark
  22. PySpark: match the values of a DataFrame column against another DataFrame column
  23. Spark mllib predicting weird number or NaN
  24. Multiple condition filter on dataframe
  25. How to return a “Tuple type” in a UDF in PySpark?
  26. How to connect HBase and Spark using Python?
  27. Pyspark 2.4.0, read avro from kafka with read stream – Python
  28. How to pivot on multiple columns in Spark SQL?
  29. Apache Spark — Assign the result of UDF to multiple dataframe columns
  30. how to add Row id in pySpark dataframes [duplicate]

Leave a Comment