Beginner’s Guide To Create PySpark DataFrame

All the files and codes used below can be found here.

Here, we will use Google Colaboratory for practice purposes. To start importing our CSV Files in PySpark, we need to follow some prerequisites. First, we will install the pyspark library in Google Colaboratory using pip.

!pip install pyspark

Now, let’s see how to create the PySpark Dataframes using the two methods discussed above.

Create PySpark DataFrame From an Existing RDD

To create a PySpark DataFrame from an existing RDD, we will first create an RDD using the .parallelize() method and then convert it into a PySpark DataFrame using the .createDatFrame() method of SparkSession.

To start using PySpark, we first need to create a Spark Session. A spark session can be created by importing a library.

Importing the Libraries

from pyspark import SparkContext
from pyspark.sql import SparkSession

Creating a SparkContext

sc = SparkContext.getOrCreate()

We used the .getOrCreate() method of SparkContext to create a SparkContext for our exercise. The .getOrCreate() method will create and instantiate SparkContext into our variable sc or will fetch the old one if already created before.

Creating SparkSession

spark = SparkSession.builder.appName('PySpark DataFrame From RDD').getOrCreate()

Here, will have given the name to our Application by passing a string to .appName() as an argument. Next, we used .getOrCreate() which will create and instantiate SparkSession into our object spark. Using the .getOrCreate() method would use an existing SparkSession if one is already present else will create a new one.

Creating a Resilient Data Structure (RDD)

rdd = sc.parallelize([('C',85,76,87,91), ('B',85,76,87,91), ("A", 85,78,96,92), ("A", 92,76,89,96)], 4)

We used the .parallelize() method of SparkContext sc which took the tuples of marks of students. In the later steps, we will convert this RDD into a PySpark Dataframe. We passed numSlices value to 4 which is the number of partitions our data would parallelize into.

Checking the Datatype of RDD

print(type(rdd))

We can verify if our RDD creation is successful by checking the datatype of the variable rdd. On executing this, we will get pyspark.rdd.RDD

Converting the RDD into PySpark DataFrame

sub = ['Division','English','Mathematics','Physics','Chemistry']
marks_df = spark.createDataFrame(rdd, schema=sub)

Here, The .createDataFrame() method from SparkSession spark takes data as an RDD, a Python list or a Pandas DataFrame. Here we are passing the RDD as data. We also created a list of strings sub which will be passed into schema attribute of .createDataFrame() method.

The dataType of PySpark DataFrame

print(type(marks_df))

To verify if our operation is successful, we will check the datatype of marks_df. On executing this we will get pyspark.sql.dataframe.DataFrame as output.

Schema of PySpark DataFrame

marks_df.printSchema()

We can also check the schema of our file by using the .printSchema() method which is very useful when we have tens or hundreds of columns.

Contents of PySpark DataFrame

marks_df.show()

To view the contents of the file, we will use the .show() method on the PySpark Dataframe object. This will display the top 20 rows of our PySpark DataFrame.

Putting it all together

from pyspark import SparkContext
from pyspark.sql import SparkSession
sc = SparkContext.getOrCreate()
spark = SparkSession.builder.appName('PySpark DataFrame From RDD').getOrCreate()
rdd = sc.parallelize([('C',85,76,87,91), ('B',85,76,87,91), ("A", 85,78,96,92), ("A", 92,76,89,96)], 4)
#print(type(rdd))
sub = ['Division','English','Mathematics','Physics','Chemistry']
marks_df = spark.createDataFrame(rdd, schema=sub)
#print(type(marks_df))
#marks_df.printSchema()
marks_df.show()

On executing this, we get:

Create PySpark DataFrame From an External File

Importing the libraries

from pyspark.sql import SparkSession

Creating Spark Session

spark = SparkSession.builder.appName('PySpark DataFrame From External Files').getOrCreate()

Here, will have given the name to our Application by passing a string to .appName() as an argument. Next, we used .getOrCreate() which will create and instantiate SparkSession into our object spark. Using the .getOrCreate() method would use an existing SparkSession if one is already present else will create a new one.

Reading External Files into PySpark DataFrame

1. Reading a CSV File

csv_file = spark.read.csv('Fish.csv', sep = ',', inferSchema = True, header = True)

In the spark.read.csv(), first, we passed our CSV file Fish.csv. Second, we passed the delimiter used in the CSV file. Here the delimiter is a comma ‘,‘. Next, we set the inferSchema attribute as True, this will go through the CSV file and automatically adapt its schema into PySpark Dataframe. Though, setting inferSchema to True may take time but is highly useful when we are working with a huge dataset.

2. Reading a TXT File

txt_file = spark.read.text("example.txt")

In the spark.read.text() method, we passed our txt file example.txt as an argument. Each line in this text file will act as a new row. This is useful when we want to read multiple lines at once.

3. Reading a JSON File

json_file = spark.read.json("sample.json", multiLine=True)

In the spark.read.json() method, we passed our JSON file sample.json as an argument. Also, we have set the multiLine Attribute to True to read the data from multiple lines. By default, JSON file inferSchema is set to True.

Checking DataTypes of PySpark DataFrames

print(type(csv_file))

print(type(txt_file))

print(type(json_file))

Checking Schema of PySpark DataFrames

csv_file.printSchema()

txt_file.printSchema()

json_file.printSchema()

Putting it all together

from pyspark.sql import SparkSession
spark = SparkSession.builder.appName('PySpark DataFrame From External Files').getOrCreate()
csv_file = spark.read.csv('Fish.csv', sep = ',', inferSchema = True, header = True)
txt_file = spark.read.text("example.txt")
json_file = spark.read.json("sample.json", multiLine=True)
print(type(csv_file))
#print(type(txt_file))
#print(type(json_file))
csv_file.printSchema()
#txt_file.printSchema()
#json_file.printSchema()