Skip to content
Project: Cleaning an Orders Dataset with PySpark
As a Data Engineer at an electronics e-commerce company, Voltmart, you have been requested by a peer Machine Learning team to clean the data containing the information about orders made last year. They are planning to further use this cleaned data to build a demand forecasting model. To achieve this, they have shared their requirements regarding the desired output table format.
An analyst shared a parquet file called "orders_data.parquet" for you to clean and preprocess.
You can see the dataset schema below along with the cleaning requirements:
orders_data.parquet
orders_data.parquet| column | data type | description | cleaning requirements |
|---|---|---|---|
order_date | timestamp | Date and time when the order was made | Modify: Remove orders placed between 12am and 5am (inclusive); convert from timestamp to date |
time_of_day | string | Period of the day when the order was made | New column containing (lower bound inclusive, upper bound exclusive): "morning" for orders placed 5-12am, "afternoon" for orders placed 12-6pm, and "evening" for 6-12pm |
order_id | long | Order ID | N/A |
product | string | Name of a product ordered | Remove rows containing "TV" as the company has stopped selling this product; ensure all values are lowercase |
product_ean | double | Product ID | N/A |
category | string | Broader category of a product | Ensure all values are lowercase |
purchase_address | string | Address line where the order was made ("House Street, City, State Zipcode") | N/A |
purchase_state | string | US State of the purchase address | New column containing: the State that the purchase was ordered from |
quantity_ordered | long | Number of product units ordered | N/A |
price_each | double | Price of a product unit | N/A |
cost_price | double | Cost of production per product unit | N/A |
turnover | double | Total amount paid for a product (quantity x price) | N/A |
margin | double | Profit made by selling a product (turnover - cost) | N/A |
from pyspark.sql import (
SparkSession,
types,
functions as F,
)
spark = (
SparkSession
.builder
.appName('cleaning_orders_dataset_with_pyspark')
.getOrCreate()
)orders_data = spark.read.parquet('orders_data.parquet')
orders_data.toPandas().head()# Start here, using as many cells as you require
df =(orders_data
#create requested columns
.withColumn('hour',F.hour(F.col('order_date')))
.withColumn('order_date',F.col('order_date').cast(types.DateType()))
.withColumn('product',F.lower(F.col('product')))
.withColumn('time_of_day',
F.when(F.col('hour').between(5,11),'morning').when(F.col('hour')
.between(12,17),'afternoon').when(F.col('hour').between(18,24),'evening').otherwise(None))
.withColumn('category',F.lower(F.col('category')))
#use string functions to isolate state
.withColumn('purchase_state',(F.split(F.split('purchase_address',',')[2],' ')[1]))
#Filter out orders placed before 6am and those for TV's
.filter(F.col('hour')>=6)
.filter(~F.array_contains(F.split('product',' '),'tv'))
)
df_final = (df.select('order_date','time_of_day','order_id','product','category','purchase_address','purchase_state','quantity_ordered','price_each','cost_price','turnover','margin'))
df_final.show()df_final.write.mode('overwrite').save('orders_data_clean.parquet')