Spark: Beginner Tips & Tricks

I recently worked on a hackathon team at my company. The purpose of the hackathon was to show how to use data science on Automated Identification System (AIS) ship data to determine anchorage insights. I decided to use this hackathon as a means to learn Spark (and PySpark specifically.) My part of the project mainly involved data wrangling, to format the data and stage it appropriately for the data science operations. This document is intended to convey the key things I learned as a Spark beginner.

NOTE: All examples in this document are PySpark/Python code.

The Environment

• AWS EMR cluster
• Jupyter Notebook
• S3 storage for holding the AIS records and staged data
• Python 3
• PySpark

Lazy Evaluation

One other thing to note before we go forward. Spark code does not get executed when a line of code runs. Your code is queued up but doesn’t execute until the code is needed to perform a task. This queuing of code is called ‘Lazy Evaluation.’ For instance, if you execute a line of code to read a file, the file is not read until that data is needed somewhere else. Understanding that code in Spark is not executed immediately is important to understand because it can make development and debugging tricky if this concept Is not fully recognized.

Spark Configuration

We were dealing with a year of AIS data, which is a large dataset, which included approximately 60-90 million records per month.

While processing and analyzing the AIS data, I would occasionally run into memory errors and MaxResultSize errors on the cluster. To resolve these issues, I had to modify the cluster configuration and increase the Spark memory size.

This is done by adjusting the maximizeResourceAllocation configuration on an EMR cluster tells Spark to use all of the resources available (cores, memory, etc.) to process the job. An AWS architect may not want to use this setting in a production environment, but for my purposes, it got the job done.

To configure this setting go to the “Software Configuration” page (shown in Figure 1) of the cluster configuration.

Figure 1: Software Configuration Page

 

Add the following configuration in the “Edit software settings” box (Figure 2: Edit software settings).

 

Figure 2: Edit software settings

 

The second configuration that I added was the maxResultSize configuration. This configuration can be set directly in your notebook. I used 4 gigabytes in this setting, but your mileage may vary. The required setting of this configuration will be dependent on your dataset. To configure this setting run the command shown in Figure 3 in a notebook cell.

 

Figure 3: maxResultSize Configuration

 

Dynamically Install Libraries

Prior to AWS EMR 5.26.0, in order to use an imported library in your notebook, you would need to configure that library in your cluster configuration. As of EMR 5.26.0, AWS allows notebook scoped libraries. See the documentation here:

https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-managed-notebooks-scoped-libraries.html

With notebook scoped libraries, you can quickly import required libraries for use in your notebook without having to stop and start your cluster. Figure 4 shows how to import libraries directly in your notebook dynamically.

 

Figure 4: Notebook Scoped Libraries

 

UDF Functions

PySpark user-defined functions (UDF) allow a developer to use Python native code to process PySpark DataFrames and columns. There are situations where native python functions cannot process DataFrames columns. To resolve this and allow the use of native python functions on PySpark columns, we need to use UDF functions.

Implementation of UDF functions requires two steps.

• Write the python function
• Register the python function as a UDF function

Figure 5 shows how to create and register a UDF function for use in your notebook. A haversine function is a method for calculating the distance between 2 geographic points. The haversine function is a native Python function. The udf_haversine function shows how to wrap a Python function as a UDF function.

Figure 5: Spark UDF Functions

 

The code in Figure 6 shows how to call the UDF function from your code. The columns lat_lag1 and lon_lag1 refer to the location from the previous AIS record, and lat and lon refer to the location from the current AIS record.

 

Figure 6: Call a UDF Function

 

Reading a Parquet File from an S3 Bucket

Reading a file from an S3 bucket is a straightforward but essential task for processing data. Figure 7 shows how to read a parquet file from an S3 bucket. Notice that I first set up an array of StructField objects to define the schema. The array allows me to specify the datatypes for the resulting DataFrame.

• createDataFrame creates an empty DataFrame
• read.load reads the file and loads it into the DataFrame

Figure 7: Reading from a Parquet File

 

Writing a Parquet File to an S3 Bucket

To perform tasks in parallel, Spark uses partitions. For a file write, this means breaking up the write into multiple files. The multiple files allow the write to execute more quickly for large datasets since Spark can perform the write in parallel.

Figure 8 shows how to define a partition for a file. In this case, I used the date of the record to determine which file to place the record. It also shows how to write the file using partitions.

 

Figure 8: Write Parquet File

 

How to use Windowing on a DataFrame

Windowing on a DataFrame is similar to windowing in SQL; it allows you to view other records in the DataFrame beyond the current record. There are many things you can do with windowing; this will be a cursory overview.

When you are processing a DataFrame, you process each record in the DataFrame similar to the way a SQL statement would process each record in a database table. For instance, take the code in Figure 9

Figure 9: withColumn Example Code

This code adds a new column that is the sum of colA and colB of the current record that is being processed. But what if we wanted to sum colA from the previous record in the set with the current record in the set. ColA from the previous record is not available to us under normal circumstances. This need to see the previous record is where ‘windowing’ comes in. It allows us to peer into other records within the dataset while processing the current record.

Here is an article with an in-depth look at Windowing on DataFrames.

https://databricks.com/blog/2015/07/15/introducing-window-functions-in-spark-sql.html

Using DataFrame windowing requires two steps:

• Configure the window
• Use the window to view records outside the current record

The first example shown in Figure 10 shows how to window to the previous record. The first step is to configure the window (here it is assigned to a variable ‘w’). Notice that the window has a ‘partitionBy’ and an ‘orderBy.’ An ‘orderBy’ is required to use the ‘lag’ function.

Then we use the ‘lag’ function to pull the ‘lat,’ ‘lon,’ and ‘timereceived’ from the previous record onto the current record.

 

Figure 10: Windowing to the Previous Record

 

With DataFrame windowing in addition to the ‘lag’ function, you can also use ‘first’ to get the first record in the partition, ‘last’, to get the last record in the partition. Furthermore, you can use aggregation functions such as count, sum, average over a Windowing partition.

Matplotlib Visualizations in Spark

Visualizing the data for analysis is an important part of data science. Using matplotlib we can visualize our data using graphs like line chart, bar charts, or scatter graphs.

Though visualizing data using matplotlib in Pandas is a common task, doing this with PySpark in an EMR notebook took me a while to figure out, so I will be listing several tips in this section that are important to make this work.

Figure 11 shows the code I used to visualize my data for analysis. I will breakdown this code for you.

• Remember, I had previously imported the matplotlib library as a notebook scoped library.
• Notice the imports: I imported the entire matplotlib library and have a second import for pyplot from that library
• I assigned my DataFrame to a variable called df_plot
• I bucketed my DataFrame data suitable for a bar chart according to the buckets
• I used a groupBy function to facilitate the aggregation
• I converted my Spark DataFrame to a Pandas DataFrame
• I executed the plot function, configured as a bar graph to display the buckets in the visualization

 

Figure 11: Charting using MATPLOTLIB

 

Now, because of ‘Lazy Evaluation’ nothing happens yet. To display the plot EMR notebooks include a ‘magic’ function that executes the previous code. This ‘magic’ function is shown in Figure 12. Execute this code to display your chart in the notebook.

 

Figure 12: matplotlib magic function

 

The key to making this visualization work is that I needed to convert my DataFrame to a Pandas DataFrame. However, Pandas is not very good at dealing with Big Data, so make sure that you aggregate your data appropriately before you convert your DataFrame.

Also, I needed to execute the ‘magic’ function in a separate cell; otherwise, I received errors.

Conclusion

These are just a few of the most important things I learned as I worked my way through my initial effort with Spark. I hope these tips were useful for you. I was able to process 10s of millions of records in a relatively short period of time. Spark is a powerful tool that makes quick work of Big Data.