Best way to get the max value in a Spark dataframe column

Question

I'm trying to figure out the best way to get the largest value in a Spark dataframe column.

Consider the following example:

df = spark.createDataFrame([(1., 4.), (2., 5.), (3., 6.)], ["A", "B"])
df.show()

Which creates:

+---+---+
|  A|  B|
+---+---+
|1.0|4.0|
|2.0|5.0|
|3.0|6.0|
+---+---+

My goal is to find the largest value in column A (by inspection, this is 3.0). Using PySpark, here are four approaches I can think of:

# Method 1: Use describe()
float(df.describe("A").filter("summary = 'max'").select("A").first().asDict()['A'])

# Method 2: Use SQL
df.registerTempTable("df_table")
spark.sql("SELECT MAX(A) as maxval FROM df_table").first().asDict()['maxval']

# Method 3: Use groupby()
df.groupby().max('A').first().asDict()['max(A)']

# Method 4: Convert to RDD
df.select("A").rdd.max()[0]

Each of the above gives the right answer, but in the absence of a Spark profiling tool I can't tell which is best.

Any ideas from either intuition or empiricism on which of the above methods is most efficient in terms of Spark runtime or resource usage, or whether there is a more direct method than the ones above?

Methods 2 and 3 are equivalent and use identical physical and optimized logical plans. Method 4 applies reduce with max on rdd. It can be slower than operating directly on a DataFrame. Method 1 is more or less equivalent to 2 and 3. — zero323, Oct 19 '15 at 22:33
@zero323 What about `df.select(max("A")).collect()[0].asDict()['max(A)']`? Looks equivalent to Method 2 while more compact, and also more intuitive that Method 3. — desertnaut, Nov 02 '15 at 10:02
- The slowest is the method 4, because you do DF to RDD conversion of the whole column and then extract max value; — Danylo Zherebetskyy, Feb 13 '18 at 20:00

Burt · Accepted Answer · 2016-07-12T14:17:54.933

124

>df1.show()
+-----+--------------------+--------+----------+-----------+
|floor|           timestamp|     uid|         x|          y|
+-----+--------------------+--------+----------+-----------+
|    1|2014-07-19T16:00:...|600dfbe2| 103.79211|71.50419418|
|    1|2014-07-19T16:00:...|5e7b40e1| 110.33613|100.6828393|
|    1|2014-07-19T16:00:...|285d22e4|110.066315|86.48873585|
|    1|2014-07-19T16:00:...|74d917a1| 103.78499|71.45633073|

>row1 = df1.agg({"x": "max"}).collect()[0]
>print row1
Row(max(x)=110.33613)
>print row1["max(x)"]
110.33613

The answer is almost the same as method3. but seems the "asDict()" in method3 can be removed

edited Jul 12 '16 at 14:17

answered Jul 12 '16 at 13:08

Burt

1,520
1
12
10

can someone explain why collect()[0] is needed? – jibiel Oct 16 '17 at 08:54
2

@jibiel `collect()` returns a list (in this case with a single item), so you need to access the first (only) item in the list – Jason Wolosonovich Jan 30 '18 at 08:30
2

@Burt `head()` can be used instead if `collect()[0]`. – Aliaxander Jan 15 '19 at 10:57
@Aliaxander It's been a bit long. Don't have the code and Spark installed anymore. – Burt Jan 25 '19 at 09:21
16

While `.collect()[0]` works, it's probably safer to use `.first()[0]`. By definition, [collect()](https://spark.apache.org/docs/latest/rdd-programming-guide.html#printing-elements-of-an-rdd) will "Return all the elements of the dataset as an array at the driver program.", **which is a single machine**. If you get the syntax wrong you could end up using an excessive amount of memory. – Chris Koester Sep 09 '19 at 15:43

score 85 · Answer 2 · edited Nov 17 '22 at 16:21

85

Max value for a particular column of a dataframe can be achieved by using -

your_max_value = df.agg({"your-column": "max"}).collect()[0][0]

edited Nov 17 '22 at 16:21

Hadij

3,661
5
26
48

answered Sep 11 '17 at 14:09

Rudra Prasad Samal

916
6
2

2

I prefer your solution to the accepted solution. Adding two "[0]" gives result only – omnisius May 28 '19 at 07:18

score 41 · Answer 3 · edited Sep 14 '21 at 14:06

Remark: Spark is intended to work on Big Data - distributed computing. The size of the example DataFrame is very small, so the order of real-life examples can be altered with respect to the small example.

Slowest: Method_1, because .describe("A") calculates min, max, mean, stddev, and count (5 calculations over the whole column).

Medium: Method_4, because, .rdd (DF to RDD transformation) slows down the process.

Faster: Method_3 ~ Method_2 ~ Method_5, because the logic is very similar, so Spark's catalyst optimizer follows very similar logic with minimal number of operations (get max of a particular column, collect a single-value dataframe; .asDict() adds a little extra-time comparing 2, 3 vs. 5)

import pandas as pd
import time

time_dict = {}

dfff = self.spark.createDataFrame([(1., 4.), (2., 5.), (3., 6.)], ["A", "B"])
#--  For bigger/realistic dataframe just uncomment the following 3 lines
#lst = list(np.random.normal(0.0, 100.0, 100000))
#pdf = pd.DataFrame({'A': lst, 'B': lst, 'C': lst, 'D': lst})
#dfff = self.sqlContext.createDataFrame(pdf)

tic1 = int(round(time.time() * 1000))
# Method 1: Use describe()
max_val = float(dfff.describe("A").filter("summary = 'max'").select("A").collect()[0].asDict()['A'])
tac1 = int(round(time.time() * 1000))
time_dict['m1']= tac1 - tic1
print (max_val)

tic2 = int(round(time.time() * 1000))
# Method 2: Use SQL
dfff.registerTempTable("df_table")
max_val = self.sqlContext.sql("SELECT MAX(A) as maxval FROM df_table").collect()[0].asDict()['maxval']
tac2 = int(round(time.time() * 1000))
time_dict['m2']= tac2 - tic2
print (max_val)

tic3 = int(round(time.time() * 1000))
# Method 3: Use groupby()
max_val = dfff.groupby().max('A').collect()[0].asDict()['max(A)']
tac3 = int(round(time.time() * 1000))
time_dict['m3']= tac3 - tic3
print (max_val)

tic4 = int(round(time.time() * 1000))
# Method 4: Convert to RDD
max_val = dfff.select("A").rdd.max()[0]
tac4 = int(round(time.time() * 1000))
time_dict['m4']= tac4 - tic4
print (max_val)

tic5 = int(round(time.time() * 1000))
# Method 5: Use agg()
max_val = dfff.agg({"A": "max"}).collect()[0][0]
tac5 = int(round(time.time() * 1000))
time_dict['m5']= tac5 - tic5
print (max_val)

print time_dict

Result on an edge-node of a cluster in milliseconds (ms):

small DF (ms): {'m1': 7096, 'm2': 205, 'm3': 165, 'm4': 211, 'm5': 180}

bigger DF (ms): {'m1': 10260, 'm2': 452, 'm3': 465, 'm4': 916, 'm5': 373}

score 26 · Answer 4 · answered Sep 03 '18 at 16:30

Another way of doing it:

df.select(f.max(f.col("A")).alias("MAX")).limit(1).collect()[0].MAX

On my data, I got this benchmarks:

df.select(f.max(f.col("A")).alias("MAX")).limit(1).collect()[0].MAX
CPU times: user 2.31 ms, sys: 3.31 ms, total: 5.62 ms
Wall time: 3.7 s

df.select("A").rdd.max()[0]
CPU times: user 23.2 ms, sys: 13.9 ms, total: 37.1 ms
Wall time: 10.3 s

df.agg({"A": "max"}).collect()[0][0]
CPU times: user 0 ns, sys: 4.77 ms, total: 4.77 ms
Wall time: 3.75 s

All of them give the same answer

"df.limit(1).collect()[0]" can be replaced by "df.first()" – Chris H. Sep 07 '21 at 16:57 — Chris H., Sep 07 '21 at 16:57

score 14 · Answer 5 · answered May 16 '19 at 13:51

The below example shows how to get the max value in a Spark dataframe column.

from pyspark.sql.functions import max

df = sql_context.createDataFrame([(1., 4.), (2., 5.), (3., 6.)], ["A", "B"])
df.show()
+---+---+
|  A|  B|
+---+---+
|1.0|4.0|
|2.0|5.0|
|3.0|6.0|
+---+---+

result = df.select([max("A")]).show()
result.show()
+------+
|max(A)|
+------+
|   3.0|
+------+

print result.collect()[0]['max(A)']
3.0

Similarly min, mean, etc. can be calculated as shown below:

from pyspark.sql.functions import mean, min, max

result = df.select([mean("A"), min("A"), max("A")])
result.show()
+------+------+------+
|avg(A)|min(A)|max(A)|
+------+------+------+
|   2.0|   1.0|   3.0|
+------+------+------+

Agree. I'm new to pyspark (old to Python) and this is more intuitive. — Ulf Aslak, Nov 27 '20 at 06:42
extending on this answer - if you've NaN's following will work: df.select('A').dropna().select([max('A')]) — r_hudson, Mar 09 '21 at 23:43

score 8 · Answer 6 · edited Feb 06 '20 at 15:20

8

First add the import line:

from pyspark.sql.functions import min, max

To find the min value of age in the dataframe:

df.agg(min("age")).show()

+--------+
|min(age)|
+--------+
|      29|
+--------+

To find the max value of age in the dataframe:

df.agg(max("age")).show()

+--------+
|max(age)|
+--------+
|      77|
+--------+

edited Feb 06 '20 at 15:20

tardis

1,280
3
23
48

answered Sep 30 '19 at 05:48

satprem rath

91
1
3

proutray · Answer 7 · 2020-06-11T12:49:07.123

8

I used another solution (by @satprem rath) already present in this chain.

To find the min value of age in the dataframe:

df.agg(min("age")).show()

+--------+
|min(age)|
+--------+
|      29|
+--------+

edit: to add more context.

While the above method printed the result, I faced issues when assigning the result to a variable to reuse later.

Hence, to get only the int value assigned to a variable:

from pyspark.sql.functions import max, min  

maxValueA = df.agg(max("A")).collect()[0][0]
maxValueB = df.agg(max("B")).collect()[0][0]

edited Jun 11 '20 at 12:49

answered May 05 '20 at 20:51

proutray

1,943
3
30
48

4

Please add a bit of context and explanation around your solution. – MegaIng May 05 '20 at 23:48

score 5 · Answer 8 · edited Sep 14 '21 at 13:59

5

To just get the value use any of these

df1.agg({"x": "max"}).collect()[0][0]
df1.agg({"x": "max"}).head()[0]
df1.agg({"x": "max"}).first()[0]

Alternatively we could do these for 'min'

from pyspark.sql.functions import min, max
df1.agg(min("id")).collect()[0][0]
df1.agg(min("id")).head()[0]
df1.agg(min("id")).first()[0]

edited Sep 14 '21 at 13:59

ZygD

22,092
39
79
102

answered Apr 18 '20 at 06:50

Blue Clouds

7,295
4
71
112

score 4 · Answer 9 · answered Nov 22 '16 at 10:29

4

In case some wonders how to do it using Scala (using Spark 2.0.+), here you go:

scala> df.createOrReplaceTempView("TEMP_DF")
scala> val myMax = spark.sql("SELECT MAX(x) as maxval FROM TEMP_DF").
    collect()(0).getInt(0)
scala> print(myMax)
117

answered Nov 22 '16 at 10:29

Boern

7,233
5
55
86

score 3 · Answer 10 · edited Jan 14 '19 at 08:35

3

I believe the best solution will be using head()

Considering your example:

+---+---+
|  A|  B|
+---+---+
|1.0|4.0|
|2.0|5.0|
|3.0|6.0|
+---+---+

Using agg and max method of python we can get the value as following :

from pyspark.sql.functions import max df.agg(max(df.A)).head()[0]

This will return: 3.0

Make sure you have the correct import:
from pyspark.sql.functions import max The max function we use here is the pySPark sql library function, not the default max function of python.

edited Jan 14 '19 at 08:35

Jean-François Corbett

37,420
30
139
188

answered Jul 06 '18 at 19:17

Vyom Shrivastava

41
4

1

Make sure you have the correct imports, You need to import the following: `from pyspark.sql.functions import max` The max we use here is the pySpark sql function not the python max It is better if you use use alias for it `from pyspark.sql.functions import max as mx` – Vyom Shrivastava Jul 10 '18 at 17:21

score 1 · Answer 11 · answered Feb 04 '19 at 20:01

1

in pyspark you can do this:

max(df.select('ColumnName').rdd.flatMap(lambda x: x).collect())

answered Feb 04 '19 at 20:01

Grant Shannon

4,709
1
46
36

user 923227 · Answer 12 · 2018-09-12T18:58:44.973

Here is a lazy way of doing this, by just doing compute Statistics:

df.write.mode("overwrite").saveAsTable("sampleStats")
Query = "ANALYZE TABLE sampleStats COMPUTE STATISTICS FOR COLUMNS " + ','.join(df.columns)
spark.sql(Query)

df.describe('ColName')

or

spark.sql("Select * from sampleStats").describe('ColName')

or you can open a hive shell and

describe formatted table sampleStats;

You will see the statistics in the properties - min, max, distinct, nulls, etc.

score 0 · Answer 13 · answered Jan 18 '19 at 03:31

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._

val testDataFrame = Seq(
  (1.0, 4.0), (2.0, 5.0), (3.0, 6.0)
).toDF("A", "B")

val (maxA, maxB) = testDataFrame.select(max("A"), max("B"))
  .as[(Double, Double)]
  .first()
println(maxA, maxB)

And the result is (3.0,6.0), which is the same to the testDataFrame.agg(max($"A"), max($"B")).collect()(0).However, testDataFrame.agg(max($"A"), max($"B")).collect()(0) returns a List, [3.0,6.0]

Best way to get the max value in a Spark dataframe column

13 Answers13

To find the min value of age in the dataframe:

To find the max value of age in the dataframe:

Linked

Related