Spark DataFrame Aggregations: count, mean, approx_count_distinct, summary

1.count: three calls with the same name

The word count shows up in three places in PySpark, and each one returns something different. The simplest is the DataFrame method df.count(). It is an action that returns a plain Python int holding the number of rows. You get a number back, not a DataFrame or a Column.

df = spark.createDataFrame(
    [(14, "Tom"), (23, "Alice"), (16, "Bob")], ["age", "name"])

df.count()
# 3

The second is the aggregate function sf.count(col) from pyspark.sql.functions. It returns a Column, so you use it inside select() or agg(), and the result comes back as a one-row DataFrame. What you pass it changes the answer. Passing sf.expr("*") counts every row, and the output header is literally count(1). Passing a specific column counts only the rows where that column is not null.

from pyspark.sql import functions as sf
df = spark.createDataFrame([(None,), ("a",), ("b",), ("c",)], schema=["alphabets"])
df.select(sf.count(sf.expr("*"))).show()

+--------+
|count(1)|
+--------+
|       4|
+--------+

On that same DataFrame, df.select(sf.count(df.alphabets)) returns 3. The documentation shows the same split across two columns: for rows (1, "apple"), (2, "banana"), (3, None), count(id) is 3 and count(fruit) is 2. The third form is GroupedData.count(), which you call after groupBy. It counts the records in each group and returns a DataFrame with a count column next to the grouping key.

Sources123

2.mean and avg, through select or agg

Counting tells you how many rows there are. Averaging tells you something about their values. pyspark.sql.functions has both avg(col) and mean(col), and they are the same function under two names. Because mean is an alias, Spark labels the output column avg(age) even when your code calls sf.mean("age"). If a question shows avg(...) in a result header, don't conclude that the code must have called avg.

import pyspark.sql.functions as sf
df = spark.createDataFrame([(1982, None), (1990, 2), (2000, 4)], ["birth", "age"])
df.select(sf.avg("age")).show()

+--------+
|avg(age)|
+--------+
|     3.0|
+--------+

The answer is 3.0. Nulls are left out of both the sum and the divisor, so a null is not treated as zero. The Spark SQL reference for mean says the same thing and adds one edge case: if a group is empty or holds only nulls, the result is NULL.

So far the examples have used select(). The other common entry point is df.agg(), which aggregates over the whole DataFrame and is shorthand for df.groupBy().agg(). It accepts two forms. One is a dictionary that maps a column name to the name of an aggregate function. The other is one or more aggregate Column expressions, like the ones you have been building with sf.

from pyspark.sql import functions as sf
df = spark.createDataFrame([(2, "Alice"), (5, "Bob")], schema=["age", "name"])
df.agg({"age": "max"}).show()
# +--------+
# |max(age)|
# +--------+
# |       5|
# +--------+
df.agg(sf.min(df.age)).show()
# +--------+
# |min(age)|
# +--------+
# |       2|
# +--------+

After a groupBy, the same alias appears as a shortcut method: GroupedData.mean(*cols) and GroupedData.avg(*cols) both compute the average of each numeric column for each group. For example, df.groupBy("name").avg("age") returns 2.5 for Alice (ages 2 and 3) and 7.5 for Bob (ages 5 and 10).

Sources4563

3.approx_count_distinct: trading exactness for speed

count and mean always return exact answers. Counting distinct values exactly can be expensive on very large data, and Databricks names approximation as the way to make it cheaper. The function for this is sf.approx_count_distinct(col, rsd=None). It returns a Column that estimates the number of distinct values in a column. Like the other aggregate functions, you use it inside agg() or select().

from pyspark.sql import functions as sf
df = spark.createDataFrame([("apple",), ("orange",), ("apple",), ("banana",)], ['fruit'])
df.agg(sf. ? ("fruit")).show()

The function counts distinct values in one column. To count distinct combinations across several columns, the documentation first packs them into a struct with sf.struct("name", "value") and passes the struct column in. For rows (Alice, 1), (Alice, 2), (Bob, 3), (Bob, 3), that returns 3. Small examples like these come back exact. On a larger input the estimate starts to drift from the true count.

from pyspark.sql import functions as sf
spark.range(100000).agg(
    sf.approx_count_distinct("id").alias('with_default_rsd'),
    sf.approx_count_distinct("id", 0.1).alias('with_rsd_0.1')
).show()

+----------------+------------+
|with_default_rsd|with_rsd_0.1|
+----------------+------------+
|           95546|      102065|
+----------------+------------+

The default rsd of 0.05 lands at 95546, and the looser 0.1 lands at 102065. A larger rsd permits more error, and the estimate can fall above or below the true value. Lowering rsd makes the estimate tighter, but the documentation puts a floor on how far that is worth taking. Below 0.01, you are better off using the exact count_distinct.

Sources78

4.describe and summary: many statistics in one call

The functions above compute one statistic at a time. When you want a quick profile of a DataFrame, two DataFrame methods compute several at once: describe() and summary(). Both work on numeric and string columns, and both return a new DataFrame whose first column, summary, names the statistic in each row. Both are documented as tools for exploratory data analysis, and Databricks makes no promise that the layout of the result stays backward compatible. That makes them poor choices to feed into downstream pipeline code.

df.describe(['age']).show()
# +-------+----+
# |summary| age|
# +-------+----+
# |  count|   3|
# |   mean|12.0|
# | stddev| 1.0|
# |    min|  11|
# |    max|  13|
# +-------+----+

describe(*cols) takes a column name or a list of column names and covers every column if you pass none. It always returns the same five statistics. summary(*statistics) works the other way round. You choose columns beforehand with select(), and its arguments name the statistics you want. Called with no arguments, it returns count, mean, stddev, min, the 25%, 50% and 75% percentiles, and max. You can also request arbitrary approximate percentiles written as percentages. The describe documentation points to summary when you need more statistics or want to choose which ones are computed.

df.select("age", "weight", "height").summary("count", "min", "25%", "75%", "max").show()
# +-------+---+------+------+
# |summary|age|weight|height|
# +-------+---+------+------+
# |  count|  3|     3|     3|
# |    min| 11|  37.8| 142.2|
# |    25%| 11|  37.8| 142.2|
# |    75%| 13|  44.1| 150.5|
# |    max| 13|  44.1| 150.5|
# +-------+---+------+------+

Sources910

Exam traps

Each one states something that sounds right. Open it to see what is actually true.

1. 1.sf.count(df.col) counts every row in the DataFrame, the same way df.count() does.Why is that wrong?

   A count on a named column skips nulls in that column. Only count("*"), shown as count(1), or the DataFrame method df.count() counts every row.

   Covered in count: three calls with the same name

2. 2.approx_count_distinct returns the exact distinct count, and you should keep lowering rsd for better results.Why is that wrong?

   It returns an estimate: 95546 for 100,000 distinct ids at the default rsd. Below an rsd of 0.01, the exact count_distinct is the more efficient choice.

   Covered in approx_count_distinct: trading exactness for speed

3. 3.describe() and summary() are interchangeable, and both include percentiles.Why is that wrong?

   describe() returns only count, mean, stddev, min and max. summary() adds percentiles and lets you choose which statistics to compute.

   Covered in describe and summary: many statistics in one call