Problem

You want to use Group By in PYSpark SQL to find the average marks per student. So far you have created the students report and shown their respective subjects and marks from the students and subjects data. Let’s try to get a report on students to identify the average marks for each of them.

Solution

You might remember this output from the previous chapter. Let’s try to use Group By on this query to identify the average marks per student.

How It Works

Note that we haven’t used the Group By clause in this query. If we use an aggregate function without applying a Group By clause, we will get an error. While the UDF operates on one row per call, avg is actually what Spark calls User Defined Aggregate Functions (UDAF). These kinds of functions take a set of rows based on the Group By clause and apply the functionality on those sets of rows. Let’s see the Group By in action.

Note from this query that group by name applies Group By on the name column. The select clause only selects name and avg(marks). If you want to determine how to select the columns on which you want to apply the group column, remember that the columns that come in the Per section of the report (for example per student or per subject) will be the columns that participate in the Group By clause. You can also have multiple columns or expressions in the Group By clause. For example, if we want to find avg marks per student on a subject, we need to include both the name and the subject column in the Group By clause.

We are using both name and subject in the Group By. This will bring all the records that are unique for students and subjects. If the data contains marks obtained by students per subjects for all the semesters, this will provide us with the overall average marks of the students per subject.

Problem

You want to find the student scoring first and second in each of the subjects.

Solution

Apache Spark supports a special type of function, called windows functions, along with its built-in UDFs such as substr, round, etc. and aggregate functions such as sum and avg. While these UDFs and aggregate functions can perform most operations, there still are many operations that cannot be performed by these functions. So Spark introduces windows functions, which can operate on a set of rows while operating on a single row.

How It Works

Step 7-2-2. Using Partition By in a Rank Function

In the previous step we learned to apply the rank function . Now we will learn to use the PARTITION clause to classify the rows as multiple sets. You can visualize this as keeping a window over those set of rows, as shown in Figure 7-5.

As shown in Figure 7-5, there are four windows based on the subject. Each windows has the same subject. The C++, Java, Python, and Ruby subjects are created. Now let’s assume for each row, there are specific actions that need to be performed based on the rows. Then we can apply the window functions. Examples of such functions are:

• What is the average age of the students in their participating subjects?

• What is the percentile of marks obtained by the students in their participating subjects?

It is very hard to write such functions in pure SQL without window functions. Let’s go ahead and create the windows using the PARTITION clause .

>>> spark.sql("select name,subject,marks,rank() over (PARTITION BY subject order by name) as RANK from studentsdob st join subjects sb ON (st.studentId = sb.studentId)").show()

In this query we added a new PARTITION clause on the subject. This is to say that we are windowing based on the subject column. This will create a separate window for each distinct subject column value and will apply the window function rank only within the rows of the window set.

Here is the output:

From this output you can see that rank has been assigned to each of the subjects to students.

+-------+-------+-----+----+

|   name|subject|marks|RANK|

+-------+-------+-----+----+

|    Bob|    C++|   78|   1|

|  Julie|   Ruby|   72|   1|

|    Bob| Python|   84|   1|

|  Maria| Python|   85|   2|

|  Robin| Python|   75|   3|

|  Julie|   Java|   76|   1|

|  Maria|   Java|   83|   2|

|  Robin|   Java|   81|   3|

|William|   Java|   70|   4|

+-------+-------+-----+----+

But wait, note that we are not getting the correct rank. This is because the Order By column uses name rather than marks. Let go ahead and correct this to use the marks column for ordering the students in each of the subjects.

Here is the query:

>>> spark.sql("select name,subject,marks,rank() over (PARTITION BY subject order by marks desc) as RANK from studentsdob st join subjects sb ON (st.studentId = sb.studentId)").show()

Note that I have changed the ordering based on marks. Also I am making sure the data is in ascending order. This is to make sure the students with the highest mark get the first rank.

+-------+-------+-----+----+

|   name|subject|marks|RANK|

+-------+-------+-----+----+

|    Bob|    C++|   78|   1|

|  Julie|   Ruby|   72|   1|

|  Maria| Python|   85|   1|

|    Bob| Python|   84|   2|

|  Robin| Python|   75|   3|

|  Maria|   Java|   83|   1|

|  Robin|   Java|   81|   2|

|  Julie|   Java|   76|   3|

|William|   Java|   70|   4|

+-------+-------+-----+----+

Problem

You want to cache a table or dataset.

Solution

Many times when there is a dataset that we have to use repeatedly, we don’t want to read it every time we use it. For example, data containing country and country code will be smaller in size. This data could fit in the memory very easily without occupying a lot of space. But we wouldn’t want such data to be read each time it is used. So for such scenarios, we can use the cache provided by Spark using PySpark SQL.

When it comes to Machine Learning algorithms that involve applying logic over the same dataset multiple times, caching is the best option to get the best performance. With caching Spark will keep the data in memory so it needs to read it again. Reading the data from disk is a costly operation. For all its advantages, let’s try to use caching in the example.

How It Works

In this example, we are going to first cache the students data and create a cached view. After that, we apply a join that cached data with the subject and assignments.

Let’s try to use the cached data in queries.

While caching is very helpful, we need to use caching very carefully. We don’t want to cache huge volumes of data. It will lead to out of memory exceptions. Handling huge volumes of data in memory will leave Spark with less memory for other operations. We could get an out of memory error while performing a totally different operation, and it would actually be due to caching huge volumes data in memory.

Problem

You want to use Distribute By in PySpark SQL to effectively partition your data.

Solution

As explained at the start of this chapter , the key to getting highly effective data transformation with respect to high performance and optimum use of resources in a distributed computing environment solely depends on how effectively the data is distributed. If the data is equally distributed among all the data nodes, the Spark job will perform in an equally distributed manner. If not, we end up with some of the tasks taking the maximum time and resources while others finishing in a few seconds.

In both of these scenarios, the data is not partitioned correctly. Spark provides mechanisms for users to determine how to repartition data from Spark’s original partition. While it is called repartition in the Spark programmatic API, it is called distribute By PySpark SQL. This technique is one of the best ways to optimize Spark SQLs, yet it is not widely used because not many are aware of these clauses.

Let’s see how to use Distribute By in PySpark SQL

How It Works

This example applies the Distribute By clause on studentId since it’s the join key.

In the following queries, we create two new DataFrames called student_distributed and subjects_distributed. If you look at the query, we are applying the Distribute By clause on the studentId column .

After creating those DataFrames, we also register it as a temporary view. Up until now, there has been no action performed by Spark. When there is a call for an action further down, Spark will repartition the data based on studentId, so that the data will be evenly distributed among all the compute nodes.

There will not be any output from these queries since we are not calling the show() method and there are no actions being performed with this.

Once we have distributed views, we can start using these views in the query. Let’s join the students and subjects datasets based on the newly created distributed views .

Understanding the use of Distribute By will help us effectively parallelize the jobs with evenly distributed data. Now assume we are joining datasets based on a join key and the key is unordered in both datasets. In this case, Spark has to shuffle the data between the nodes to get the rows from both tables with the same key. See Figure 7-6 .

You can see from the Figure 7-6 that there is a lot of data moving around between the nodes. This is called shuffling across the nodes and reducing such shuffling will help us improve the performance of the jobs .

We just saw that Distribute By is one such option. Let’s try to visualize Distribute By. Assuming there are two partitions, we would get Figure 7-7.

In Figure 7-7, note that even though you have efficiently parallelized the join, there is still a lot of shuffling. Note that from the subjects that the data is going everywhere. Such data shuffling leads to performance degradation and out of memory errors when sufficient memory is not available.

The way to minimize data shuffling is by sorting the data in subjects and then joining it with students .

In the following query, we are applying both the Distribute By and Sort By clauses to achieve maximum performance .

Let’s try to visualize the join. See Figure 7-8.

In Figure 7-8, note that the number of data shuffles has been reduced and the flow of data is very smooth. There isn’t much shuffling happening compared to Figure 7-7.

Now that you have used the Distribute By and Sort By clauses together, Spark By provides an optional clause to combine the data—it’s called Cluster By.

Using Cluster By is the same as using Distribute By and Sort By together on the same columns. Let’s see Cluster By in action.

In this recipe, you have successfully used the Distribute By, Sort By, and Cluster By clauses to effectively optimize your PySpark SQL queries .