Aggregation Functions in Spark
Aggregation Functions are important part of big data analytics. When processing data, we need to a lot of different functions so it is a good thing Spark has provided us many in built functions. In this blog, we are going to learn aggregation functions in Spark.
Count
This is one of basic function where we count number of records or specify column to count. Let us see its example.
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
|
flights_df = spark.read\
.option("inferSchema", "true")\
.option("header","true")\
.csv("D:\\code\\spark\\spark-basics\\data\\flight-data\\csv")
flights_df.registerTempTable("flights")
flights_df.count()
1506
spark.sql("select count(*) from flights").show()
+--------+
|count(1)|
+--------+
| 1506|
+--------+
|
We can also use count with select expression for Data frame.
|
1
2
3
4
5
6
7
8
|
from pyspark.sql.functions import count
flights_df.select(count("DEST_COUNTRY_NAME")).show()
+------------------------+
|count(DEST_COUNTRY_NAME)|
+------------------------+
| 1506|
+------------------------+
|
If you want to know more about how to run SQL queries on spark data frames, you can read Running SQL queries on Spark DataFrames.
Count Distinct
We can also count distinct number of values from some column. For example, when we count number of countries surely we should not get 1506. Here, we will need to use count distinct function.
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
|
from pyspark.sql.functions import countDistinct
flights_df.select( \
countDistinct("DEST_COUNTRY_NAME")\
.alias("dest_countries"))\
.show()
+--------------+
|dest_countries|
+--------------+
| 170|
+--------------+
spark.sql("""select
count(distinct DEST_COUNTRY_NAME) as dest_countries
from flights""").show()
+--------------+
|dest_countries|
+--------------+
| 170|
+--------------+
|
Approximate Count Distinct
When we are dealing with huge data sets, many times we do not need an exact value for distinct count. We can work with approximate value only. This will run much faster compared to count distinct function.
|
1
2
3
4
5
6
7
8
9
10
11
|
from pyspark.sql.functions import approx_count_distinct
flights_df.select (\
approx_count_distinct("DEST_COUNTRY_NAME")\
.alias("approx countries"))\
.show()
+----------------+
|approx countries|
+----------------+
| 164|
+----------------+
|
We can see that output with count distinct 164 is approximately near to the actual value of 170. With approx count distinct function we can also pass second parameter which decides maximum acceptable error while calculating distinct count.
|
1
2
3
4
5
6
7
8
9
10
|
flights_df.select (\
approx_count_distinct("DEST_COUNTRY_NAME",0.25)\
.alias("approx countries"))\
.show()
+----------------+
|approx countries|
+----------------+
| 160|
+----------------+
|
We can also run SQL query to get approximate count.
|
1
2
3
4
5
6
7
8
9
10
11
12
|
spark.sql("""
SELECT approx_count_distinct(DEST_COUNTRY_NAME,0.1)
as approx_countries
FROM flights
""").show()
+----------------+
|approx_countries|
+----------------+
| 162|
+----------------+
|
First and Last
With First and Last function we can get first and last value of some column from data frame.
|
1
2
3
4
5
|
from pyspark.sql.functions import first, last
flights_df.select(\
first("DEST_COUNTRY_NAME"),
last("DEST_COUNTRY_NAME"))\
.show()
|
Min and Max
With these aptly named functions, we can find minimum and maximum value for some column in the data frame.
|
1
2
3
4
5
6
7
8
9
10
11
|
from pyspark.sql.functions import min, max
flights_df.select(\
min("count"),
max("count"))\
.show()
+----------+----------+
|min(count)|max(count)|
+----------+----------+
| 1| 348113|
+----------+----------+
|
|
1
2
|
# SQL query to get same result
spark.sql("select min(count),max(count) from flights").show()
|
Sum
Another function available is SUM, which we can use to sum all values from the column.
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
from pyspark.sql.functions import sum
flights_df.select(sum("count")).show()
# with SQL query
spark.sql("select sum(count) from flights").show()
# output
+----------+
|sum(count)|
+----------+
| 422269|
+----------+
|
Sum Distinct
Like Count Distinct, we can also sum only distinct values from some column. This below example, it will not make a lot of sense but it should give you an idea how to use sum distinct function.
|
1
2
3
4
5
6
7
8
|
from pyspark.sql.functions import sumDistinct
flights_df.select(sumDistinct("count")).show()
+-------------------+
|sum(DISTINCT count)|
+-------------------+
| 419432|
+-------------------+
|
Average
Though we can calculate average by sum of values divided by count for some column, there is in built average function available as well.
|
1
2
3
4
5
6
7
8
9
10
11
12
13
|
from pyspark.sql.functions import avg
flights_df.select(avg("count")).show()
# SQL code
spark.sql("select avg(count) from flights").show()
# output
+-----------------+
| avg(count)|
+-----------------+
|1655.956862745098|
+-----------------+
|
Collect Set and Collect List
We can also aggregate values from some column using collect set and collect list functions. Both functions create an array from all values of that column. Only difference is collect set does not have any duplicates whereas collect list will have duplicate values as well.
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
|
from pyspark.sql.functions import collect_list, collect_set
flights_df.select(\
collect_list("DEST_COUNTRY_NAME"),
collect_set("DEST_COUNTRY_NAME"))\
.show()
# SQL
spark.sql("""select
collect_Set(DEST_COUNTRY_NAME),
collect_list(DEST_COUNTRY_NAME)
from flights""").show()
+-------------------------------+------------------------------+
|collect_list(DEST_COUNTRY_NAME)|collect_set(DEST_COUNTRY_NAME)|
+-------------------------------+------------------------------+
| [United States, U...| [Italy, Slovakia,...|
+-------------------------------+------------------------------+
|
Grouping Data
Till now we have done aggregations on the data frame level. We can also split data in groups depending on some value and get aggregate values. Below are some examples for this.
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
|
flights_df.groupBy("DEST_COUNTRY_NAME").count().show(5)
# SQL
spark.sql("""
select DEST_COUNTRY_NAME, count(*) from
flights group by DEST_COUNTRY_NAME
limit 5""").show()
+-----------------+--------+
|DEST_COUNTRY_NAME|count(1)|
+-----------------+--------+
| Anguilla| 1|
| Russia| 1|
| Paraguay| 1|
| Senegal| 1|
| Sweden| 1|
+-----------------+--------+
|
|
1
2
3
4
5
6
7
8
9
10
11
12
|
flights_df.groupBy("DEST_COUNTRY_NAME").avg("count").show(5)
+-----------------+----------+
|DEST_COUNTRY_NAME|avg(count)|
+-----------------+----------+
| Anguilla| 21.0|
| Russia| 152.0|
| Paraguay| 90.0|
| Senegal| 29.0|
| Sweden| 65.0|
+-----------------+----------+
only showing top 5 rows
|
Conclusion
In this blog, we have gone through basic aggregation functions in Spark. There are many more functions available and we will go over them in the next few blogs. You can find code written in this blog at GitHub. See you in the next article.
