Belgium Train Delay Analysis with PySpark on Azure Databricks Cluster

12 November 2022 - 173 mins read time
Tags: Python PySpark Azure Databricks EDA

I managed to get my hands on an interesting data set that I want to analyze this time on long past due with PySpark. As my current Azure trial account will expire later this month, I will take advantage of my still available free subscription and use a free tier Azure Databricks Cluster. So, let’s dive into a Spark notebook session right away.

For this data set about trains in Belgium, I do not have any metadata available, besides the header row inside the file itself. The set is not that large both feature or records-wise; as we will find out later the limited scope of features means that these are insufficient in explaining enough variance of any ‘delay labeled’ column, be it regression or classification framed problem. In praxis, such a barebone data set will have to be complemented with further data from various sources, e.g. data from business systems, IoT sensor data in trains and stations, and external data sources like e.g. local/regional weather data, provided by our goal of the analysis is the finding of correlating features with train delays and possible insights about root causes with proven causal relationships.

import pyspark
from pyspark.sql import SparkSession
from pyspark.sql import Row
appName= "hive_pyspark"
master= "local"

# Name of the Spark application instance
app_name = spark.conf.get('spark.app.name')

# Driver TCP port
driver_tcp_port = spark.conf.get('spark.driver.port')

# Number of join partitions
num_partitions = spark.conf.get('spark.sql.shuffle.partitions')

# Show the results
print("Name: %s" % app_name)
print("Driver TCP port: %s" % driver_tcp_port)
print("Number of partitions: %s" % num_partitions)
print('PySpark Version :'+spark.version)

spark = SparkSession.builder \
	.master(master).appName(appName).enableHiveSupport().getOrCreate()

# File location and type
file_location = "/FileStore/tables"
file_type = "csv"

# CSV options
infer_schema = "false"
first_row_is_header = "true"
delimiter = ";"

# The applied options are for CSV files. For other file types, these will be ignored.
df = spark.read.format(file_type) \
  .option("inferSchema", infer_schema) \
  .option("header", first_row_is_header) \
  .option("sep", delimiter) \
  .load(file_location)

display(df)