Basic ETL using Pyspark

In this post, we will perform ETL operations using PySpark.

We use two types of sources, MySQL as a database and CSV file as a filesystem, We divided the code into 3 major parts- 1. Extract 2. Transform 3. Load.

We have a total of 3 data sources- Two Tables CITY, COUNTRY and one csv file COUNTRY_LANGUAGE.csv

We will create 4 python files.

job.py — Execute ETL operations stepwise (Extract Transform and Load)
constant.py — To store all the constant details like columns, spark instance, and columns related to joins.
extract.py — Extract the source data by creating dataframe.
transform.py — All transformation logic is present here. We will use extracted dataframes to do all the transformations.
load.py — Store cleaned data into a database or filesystem.

First, we will extract all the data and create a spark dataframe then we will apply the transformation to these dataframes.

Let's first, create the function spark_inst() under constant.py to initiate SparkSession. Whenever we need a Spark instance we can call spark_inst() function.

from pyspark.sql import SparkSession

#  Using this function as spark instance
def spark_inst():
    return…