PySpark: compute row maximum of the subset of columns and add to an exisiting dataframe

by

Let’s start with a couple of imports

from pyspark.sql.functions import col, lit, coalesce, greatest

Next define minus infinity literal:

minf = lit(float("-inf"))

Map columns and pass the result to greatest:

rowmax = greatest(*[coalesce(col(x), minf) for x in ['v2','v3','v4']])

Finally withColumn:

df1.withColumn("rowmax", rowmax)

with result:

+---+---+---+----+------+
| v1| v2| v3|  v4|rowmax|
+---+---+---+----+------+
|foo|1.0|3.0|null|   3.0|
|bar|2.0|2.0| -10|   2.0|
|baz|3.3|1.2|null|   3.3|
+---+---+---+----+------+

You can use the same pattern with different row wise operations replacing minf with neutral element. For example:

rowsum = sum([coalesce(col(x), lit(0)) for x in ['v2','v3','v4']])

or:

from operator import mul
from functools import reduce

rowproduct = reduce(
  mul, 
  [coalesce(col(x), lit(1)) for x in ['v2','v3','v4']]
)

Your own code could be significantly simplified with udf:

from pyspark.sql.types import DoubleType
from pyspark.sql.functions import udf

def get_max_row_with_None_(*cols):
    return float(max(x for x in cols if x is not None))

get_max_row_with_None = udf(get_max_row_with_None_, DoubleType())
df1.withColumn("rowmax", get_max_row_with_None('v2','v3','v4'))

Replace minf with lit(float("inf")) and greatest with least to get the smallest value per row.

More Related Contents:

  1. Convert pyspark string to date format
  2. How to split Vector into columns – using PySpark
  3. How to use JDBC source to write and read data in (Py)Spark?
  4. Spark Dataframe distinguish columns with duplicated name
  5. Pyspark: Split multiple array columns into rows
  6. How to change dataframe column names in pyspark?
  7. Retrieve top n in each group of a DataFrame in pyspark
  8. Pyspark: explode json in column to multiple columns
  9. How can we JOIN two Spark SQL dataframes using a SQL-esque “LIKE” criterion?
  10. How to use a Scala class inside Pyspark
  11. Pyspark ‘NoneType’ object has no attribute ‘_jvm’ error
  12. Why is Apache-Spark – Python so slow locally as compared to pandas?
  13. Cast column containing multiple string date formats to DateTime in Spark
  14. How to explode multiple columns of a dataframe in pyspark
  15. PySpark: multiple conditions in when clause
  16. PySpark create new column with mapping from a dict
  17. Count number of non-NaN entries in each column of Spark dataframe in PySpark
  18. Create Spark DataFrame. Can not infer schema for type
  19. Load CSV file with PySpark
  20. Using UDF ignores condition in when
  21. Create single row dataframe from list of list PySpark
  22. Total size of serialized results of 16 tasks (1048.5 MB) is bigger than spark.driver.maxResultSize (1024.0 MB)
  23. What is the best way to remove accents with Apache Spark dataframes in PySpark?
  24. Concatenate two PySpark dataframes
  25. How do I convert an array (i.e. list) column to Vector
  26. Filtering DataFrame using the length of a column
  27. Add column sum as new column in PySpark dataframe
  28. How to add suffix and prefix to all columns in python/pyspark dataframe
  29. How to drop all columns with null values in a PySpark DataFrame?
  30. Pyspark changing type of column from date to string

Leave a Comment