Python

PySpark - Using DataFrames

Froilan Miranda

Apr 15, 2020 • 7 min read

Spark DataFrames

Previously we looked at RDDs, and were the primary data set in Spark 1. In Spark 2 we rarely use RDDs only for low level transformations and control over the dataset. If the data is unstructured or streaming data we then have to rely on RDDs, for everything else we will use DataFrames

SparkSession vs. SparkContext

Up until now we have been using the SparkContext as the entry point to Spark. Moving forward, the SparkSession will be the entry point we will utilize

SparkSession offers:

Ease of Use
- SparkSession - simplified entry point
- No confusion about which context to use
- Encapsulates SQLContext and Hive Context

To create a SparkSession use SparkSession.builder()

Exploring Data with DataFrames

SparkSession and DataFrames

from pyspark.sql import SparkSession

Import the neccessary libraries

spark = SparkSession.builder\
                    .appName("Analyzing London Crime Data")\
                    .getOrCreate

Build a new SparkSession and assign it a name if a session with the same name does not exist. Otherwise, return the existing session with this app name. This will be the entry point to the Spark engine

data = spark.read\	# (1)
            .format("csv")\ # (2)
            .option("header", "true")\ # (3)
            .load("../datasets/london_crime_by_lsoa.csv") # (4)

.read() returns a DataFrameReader that can be used to read non-streaming data in as a DataFrame.
.format(...) sets the file format to be read
.options(...) adds input options for the underlying data source
.load(...) loads input in as a DataFrame from a data source

data.printSchema()

Remember DataFrames are always structured data. Using .printSchema will print the schema of the tabular data

data.count()

We can see the number of rows in this DataFrame

data.limit(5).show()

Examine data by looking at the fist 5 rows with the .show() function

Drop and Select Columns

data.dropna()

Drop rows which have values that are not available (N/A). As we know is a major part of data cleaning

data = data.drop('lsoa_code')

data.show(5)

To drop a column we can use .drop(...) and pass a column name as a string value.

total_boroughs = data.select('borough')\ # (1)
                    .distinct() # (2)
total_boroughs.show()

Select a column from the DataFrame
Select only distinct values

total_boroughs.count()

The number of distinct values for this column from the DataFrame

Filter Records

hackney_data = data.filter(data['borough'] == 'Hackney')
hackney_data.show(5)

Using .filter(...) we can filter records using columns their data

# (1)
data_2015_2016 = data.filter(data['year'].isin(["2015", "2016"]))

# (2)
data_2015_2016.sample(fraction=0.1).show()

Notice the use of .isin(...) withing the filter parameters. This will select the records with column values in the range of the parameters passed into it
.sample(...) returns a sampled subset of this DataFrame. fraction determines the fraction size of the full DataFrame to return

data_2014_onwards = data.filter(data['year'] >=2014)

data_2014_onwards.sample(fraction=0.1).show()

Another example of .filter(...) using >= comparator to select records with a coloumn value greater or equal to the value passed

Aggregations and grouping

borough_crime_count = data.groupBy('borough')\
                            .count()

borough_crime_count.show(5)

DataFrames support grouping of data, with the .groupBy(...) function. .groupBy(...) can be used on any column

borough_crime_count = data.groupBy('borough')\
                            .agg({"value":"sum"})

borough_crime_count.show(5)

.agg(...) is a function that will compute aggregates and return the result as a DataFrame.

Built-in aggregation functions:

avg
max
min
sum
count

borough_convictin_sum = data.groupBy('borough')\
                            .agg({"value":"sum"})\
                            .withColumnRenamed('sum(value)','convictions')

borough_convictin_sum.show(5)

Using .withColumnRenamed(<original name>, <new name>) will result in the column name being replaced by a new name

total_borough_convictions = borough_conviction_sum.agg({'convictions':'sum'})

total_borough_convictions.show()

By removing the grouping function, the aggregate will act on the whole DataFrame

total_convictions = total_borough_convictions.collect()[0][0]

Using matrix notation we can grab the value out of the collectiona and assign it to variable

import pyspark.sql.functions as fun

Imports some extra functionality from the PySpark library

borough_percentage_contribution = borough_conviction_sum.withColumn(
    '% contribution',
    func.round(borough_conviction_sum.convictions / total_convictions * 100, 2))

borough_percentage_contribution.printSchema()

Here we create a new column and use the previous variable to calculate the new column value

borough_percentage_contribution.orderBy(borough_percentage_contribution[2].desc())\
                                .show(10)

we can use .orderBy(...) and a column index to transform the DataFrame by ascending and descending order

conviction_monthly = data.filter(data['year'] == 2014)\
                            .groupBy('month')\
                            .agg({'value':'sum'})\
                            .withColumnRenamed('sum(value)', 'convictions')

Here we use a combination of group by, aggregate and column renaming to extract the data

total_conviction_monthly = conviction_monthly.agg({'convictions':'sum'})\
                                            .collect()[0][0]

total_conviction_monthly = conviction_monthly.withColumn(
                'percent',
                func.round(conviction_monthly.convictions/total_conviction_monthly * 100, 2))
total_conviction_monthly.columns

Now we use more transformations to alter the data more and print the resulting DataFrame columns

total_conviction_monthly.orderBy(total_conviction_monthly.percent.desc()).show()

Finally, we order the resulting DataFrame and display

Aggregations and Visualizations

crimes_category = data.groupBy('major_category')\
                        .agg({'value':'sum'})\
                        .withColumnRenamed('sum(value)','convictions')

use group by and aggregates to create a DataFrame

crimes_category.orderBy(crimes_category.convictions.desc()).show()

Order and display the new DataFrame

year_df = data.select('year')

Create a new DataFrame from one column

year_df.agg({'year':'min'}).show()

Use the min aggregate to return the minimum value

year_df.agg({'year':'max'}).show()

Use the max aggregate to reutrn the maximum value

year_df.describe().show()

.describe() will return:

count
mean
standard deviation
min
max

data.crosstab('borough', 'major_category')\
    .select('borough_major_category', 'Burglary', 'Drugs', 'Fraud or Forgery', 'Robbery')\
    .show()

.crosstab(...) computes a pair-wise frequency table of the given columns. Also known as a contingency table.

get_ipython().magic('matplotlib inline')
import matplotlib.pyplot as plt
plt.style.use('ggplot')

Matplotlib graphs displayed inline on this notebook

def describe_year(year):
    yearly_details = data.filter(data.year == year)\
                        .groupBy('borough')\
                        .agg({'value':'sum'})\
                        .withColumnRenamed('sum(value)', 'convictions')
    
    borough_list = [x[0] for x in yearly_details.toLocalIterator()]
    convictions_list = [x[1] for x in yearly_details.toLocalIterator()]
    
    plt.figure(figsize=(33,10))
    plt.bar(borough_list, convictions_list)
    
    plt.title('Crime for the year: ' + year, fontsize=30)
    plt.xlabel('Boroughs',fontsize=30)
    plt.ylabel('Convictions', fontsize=30)
    
    plt.xticks(rotation=90, fontsize=30)
    plt.yticks(fontsize=30)
    plt.autoscale()
    plt.show()

This is a helper function to contain all the necessary steps to create the DataFrame based off year and create the chart to visualize it

Extracting Data and User Defined Functions

In this section we will explore using DataFrames to explore and clean data. We will use User Defined Functions to assist us in the process

from pyspark.sql import SparkSession

First import SparkSession

spark = SparkSession.builder\
                .appName('Analyzing soccer players')\
                .getOrCreate()

Create SparkSession instance

players = spark.read\
                .format('csv')\
                .option('header', 'true')\
                .load('../datasets/player.csv')

Read in the data source into a DataFrame

players.printSchema()

Look at the schema

players.show(5)

Checkout the first few records

player_attributes = spark.read\
                        .format('csv')\
                        .option('header', 'true')\
                        .load('../datasets/Player_Attributes.csv')

Read in a second CSV data source in a DataFrame

player_attributes.printSchema()

Again, look at the schema

players.count(), player_attributes.count()

Lets view the total record count for each DataFrame

player_attributes.select('player_api_id')\
                .distinct()\
                .count()

Notice that entities from one DataFrame have a many to one relationship with the records of the other data set

players = players.drop('id', 'player_fifa_api_id')
players.columns

Get rid of unwanted data columns

player_attributes = player_attributes.drop(
    'id',
    'player_fifa_api_id',
    'preferred_foot',
    'attacking_work_rate',
    'defensive_work_rate',
    'crossing',
    'jumping',
    'sprint_speed',
    'balance',
    'aggression',
    'short_passing',
    'potential'
)
player_attributes.columns

Get rid of unwanted data colums

player_attributes = player_attributes.dropna()
players = players.dropna()

Remove records with non available data

players.count(), player_attributes.count()

Look at the new data count

User defined functions

from pyspark.sql.functions import udf

Import the User defined functions library

year_extract_udf = udf(lambda date: date.split('-')[0]) # (1)

player_attributes = player_attributes.withColumn( # (2)
    'year',
    year_extract_udf(player_attributes.date)
)

Create an UDF with a lambda function operate on a date value and return the year only
Create a new column for the year and extract the values from the data column using the UDF

player_attributes = player_attributes.drop('date')

Now we can drop the data column, as the year data has been copied to another column

player_attributes.columns

view the new schema

Joining DataFrames

Spark DataFrames can be joined much like SQL tables can be joined. In this section we will join data to create a new DataFrame.

pa_2016 = player_attributes.filter(player_attributes.year == 2016)

Create a new DataFrame from a subset of another DataFrame

pa_2016.count()

View the count

pa_2016.select(pa_2016.player_api_id)\
    .distinct()\
    .count()

Select only destinct value to make sure the unique ids match with the DataFrame we want to join

pa_striker_2016 = pa_2016.groupBy('player_api_id')\
                        .agg({
                            'finishing':'avg',
                            'shot_power':'avg',
                            'acceleration':'avg'
                        })

Since one data set has many records associated with an entity we will group the records by entity id first, then average the values of the columns we are interested in to create a one to one relatioinship

pa_striker_2016.count()

Check that the two DataFrame counts match

pa_striker_2016.show(5)

Take a quick look at the new aggregated data

pa_striker_2016 = pa_striker_2016.withColumnRenamed('avg(finishing)', 'finishing')\
                                 .withColumnRenamed('avg(shot_power)', 'shot_power')\
                                 .withColumnRenamed('avg(acceleration)', 'acceleration')

Rename the columns for readablity

weight_finishing = 1
weight_shot_power = 2
weight_acceleration = 1

total_weight = weight_finishing + weight_shot_power + weight_acceleration

Lets create a weighted grading system to apply more value to some attributes

strikers = pa_striker_2016.withColumn('striker_grade',
                                     (pa_striker_2016.finishing * weight_finishing + \
                                      pa_striker_2016.shot_power * weight_shot_power + \
                                      pa_striker_2016.acceleration * weight_acceleration) / total_weight)

Create a new column and apply the grading syestem to caluculate the each row value

strikers = strikers.drop('finishing',
                         'acceleration',
                         'shot_power')

Remove uneeded fields

strikers = strikers.filter(strikers.striker_grade > 70)\
                    .sort(strikers.striker_grade.desc())

strikers.show(10)

Drop lower grades from the dataset

strikers.count(), players.count()

See how many entities we have left

striker_details = players.join(strikers, players.player_api_id == strikers.player_api_id)

Now we can join the two DataFrames

striker_details.columns

View the columns, and take note the double both join fields

striker_details.count()

Check that count is inline with before

striker_details = players.join(strikers, ['player_api_id'])

Alternate way to join

striker_details.show(5)

view the data

striker_details.columns

View the columns, and take note the single join column

Saving DataFrames to CSV and JSON

Saving to file is pretty straight forward

CSV

striker_details.select("player_name", "striker_grade")\  # (1)
                .coalesce(1)\  # (2)
                .write\  # (3)
                .option('header', 'true')\  # (4)
                .csv('striker_grade.csv')  # (5)

Select the columns to export
how many files to break the data into
Begin the write command
Any options to apply
File format and file name

JSON

striker_details.select("player_name", "striker_grade")\
                .write\
                .json('striker_grade.json')

Going Further with Joins

Here we will cover other ways to join DataFrames

valuesA = [('John', 100000), ('James', 150000), ('Emily', 65000), ('Nina', 200000)]
tableA = spark.createDataFrame(valuesA, ['name', 'salary'])

Create a DataFrame from a list of tuples

tableA.show()

View DataFrame

valuesB = [('James', 2), ('Emily',3), ('Darth Vader', 5), ('Princess Leia', 6)]

tableB = spark.createDataFrame(valuesB, ['name', 'employee_id'])

Create a second DataFrame

tableB.show()