Introduction to PySpark Part 1 – Creating DataFrames and Reading Data from Files

This is the first part in a series I’m putting together as an introduction to PySpark. I find myself looking up a lot as reference so I’ve put this together as a kind of cheatsheet for myself and hopefully it helps others too.

The other parts of this blog post series can be found here:

• Part 1 – Creating Data Frames and Reading Data from Files
• Part 2 – Selecting, Filtering and Sorting Data
• Part 3 – Adding, Updating and Removing Columns
• Part 4 – Summarising Data
• Part 5 – Aggregating Data

The entire table of contents across these posts is here:

• Part 1 – Creating DataFrames and Reading Data from Files
  • Creating a DataFrame
  • Reading CSV Files as DataFrames
  • Reading JSON Files as DataFrames
  • Reading Parquet Files as DataFrames
  • Reading Data from a SQL DB
• Part 2 – Selecting, Filtering and Sorting Data
  • Selecting Columns
  • Filtering Rows
    • Chaining Conditions
    • String Matching
    • Matching against a list
    • Not (~)
    • Filtering on Nulls
    • Dropping Duplicates
  • Sorting Data
• Part 3 – Adding, Updating and Removing Columns
  • Adding Columns
    • Summing Two Columns
    • expr for Cumulative Sum
    • Other Arithmetic Operators
    • String Concatenation
  • Updating Columns
    • Renaming Columns
    • Casting Columns
    • When…Otherwise
    • UDFs
    • Filling Nulls
  • Removing Columns
• Part 4 – Summarising Data
  • Summary Statistics (Describe)
  • Getting Column max, min, mean, stdev
  • Getting Column quantiles
  • Counting Rows and Distinct values
• Part 5 – Aggregating Data
  • GroupBy
  • Pivot
  • Windowing
  • Time Series Aggregation
  • Aggregation
  • Windowing
  • Joining DataFrames
  • Concatenating DataFrames using Union

Creating a DataFrame

Processing of data on Spark is done using a data structure known as a DataFrame and a python API is provided for interacting with these data structures – PySpark. Those familiar with pandas and R will be comfortable working with DataFrames though there are some differences. For example, unlike a pandas DataFrame, you don’t have an index of rows as well as columns that you can access to select particular rows as Series objects. However, like a pandas DataFrame, we can index on column names and each column must be of a single data type.

DataFrames are tabular structures and processing is done on these data structures in parallel across the worker nodes of the spark cluster.

Let’s look at creating a DataFrame from some data we’ve defined, let’s say we have some weather data:

We could create a DataFrame by passing a list of PySpark Row objects to the spark.createDataFrame function, in which each of the row’s attributes are passed in as keyword arguments (kwargs) to the Row object.

Let’s take a look:

Alternatively, we can use pandas as an intermediate, in which we first convert to a pandas DataFrame and use the same spark.createDataFrame function to create our spark DataFame.

Let’s take a look:

Reading CSV Files as DataFrames

We can read CSV files by using spark.read.csv and passing the filepath in as an argument. If the CSV files have a header supplied, you’ll also need to pass in the header=True flag.

By default, all 3 columns here are strings and the delimiter is chosen by default to be a comma, there are two ways around this we can pass in some options when we read our CSV:

Now our Population is a DoubleType, the Year is an IntegerType and the City is a StringType.

If we wanted to define these ourselves to ensure our data types are as we expect, we can provide a schema to do so:

Reading multiple CSVs from the same directory

We can provide a path to multiple CSVs in the same directory by providing a directory name or a wilcard using an asterisk.

Let’s say we have two CSV files in a data directory we want to read, one with data from 2013 and one with data from 2014. I’m using Databricks here so I’ll use the Databricks file system utility dbutils.fs to list my directory:

I can read both files in by providing the directory:

We can see that we have the data from 2013 and if we look at the last few rows we can also see we have the 2014 data:

Reading JSON Files as DataFrames

JSON files can be read from using spark.read.json, if you have your data in the format:

[
    {'id': 1, 'name': 'Ben'},
    {'id': 2, 'name': 'Alex'},
]

You’ll also need to provide the multiline=True option otherwise spark will try and read the lines with [ and ] in as a record as well.

By default the columns are read in alphabetical order (in a later post we’ll look at selecting columns, which can be used to re-order the columns):

If you have nested JSON structures or you want to provide your own schema you can do so and I have a dedicated blog post for this – Using PySpark to Read and Flatton JSON data with an enforced schema.

Reading Parquet Files as DataFrames

Parquet files are a columnar file storage format and on large datasets can provide significant space savings and improved scan and deserialisation times. It is often the default choice of file to work with intermediate datasets when working with big data where data is often needed and queried column-wise rather than row-wise like it might in a CSV.

With Parquet files you can choose to only read in the columns you need to and there are flexible compression options for your needs.

Let’s look at reading from a Parquet File using spark.read.parquet:

Unlike CSV files, we don’t need to specify that we want to infer the schema here as it is stored with the metadata of the parquet file.

Reading Data from a SQL DB

If you want to read data from a SQL DB you can do so using a JDBC connection.

If you want to connect to an Azure SQL DB, you’ll need to install the maven library "com.microsoft.azure:spark-mssql-connector_<version>"

I will often include this as part of a deployment pipeline by installing the Databricks CLI as part of the CD pipeline, connecting it to my Databricks cluster and installing it with the line:

databricks libraries install --cluster-id $(DATABRICKS-CLUSTER-ID) --maven-coordinates "com.microsoft.azure:spark-mssql-connector_2.12_3.0:1.0.0-alpha"

I also store my database keys in the Databricks secrets utility, but for now let’s take a look at an example without the secrets utility to keep it simple: