Objective

In a manufacturing company, data is constantly collected about products and manufacturing processes. You are given two DataFrames: the first contains information about the products, and the second logs the manufacturing processes they undergo.

Task

Write a PySpark function that removes any exact duplicate rows from both DataFrames, and then combines them using their ProductID. Save the resulting DataFrame as result_df. Ensure the final columns match the order specified in the output schema.

File Path

• Products Dataset: /home/interview/products.csv
• Processes Dataset: /home/interview/processes.csv
• Starter script: /home/interview/factory_cleanup.py

Schema

products.csv

processes.csv

Expected Output Schema

Example

Given this sample input:

products_df

manufacturing_processes_df

The output would be:

Notice that the duplicate "Widget A" row in the products table was removed before the join, preventing duplicate rows in the final output.

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("PrepareshSpark").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")

products_df = spark.read.csv("/home/interview/products.csv", header=True, inferSchema=True)
processes_df = spark.read.csv("/home/interview/processes.csv", header=True, inferSchema=True)

# Clean both DataFrames by removing duplicate rows
clean_products = products_df.dropDuplicates()
clean_processes = processes_df.dropDuplicates()

# Join the cleaned DataFrames
joined_df = clean_products.join(clean_processes, on="ProductID", how="inner")

# Reorder columns to match the expected schema
result_df = joined_df.select(
    "ProductID", "ProductName", "Category", 
    "ProcessID", "ProcessName", "Duration"
)

# --- Do not edit below this line ---
result_df.coalesce(1).write.csv("/home/interview/output", header=True, mode="overwrite")
spark.stop()

Explanation

Step 1: Removing Duplicates

clean_products = products_df.dropDuplicates()
clean_processes = processes_df.dropDuplicates()

Before joining, you must ensure your data is clean. Joining tables that contain duplicate rows can lead to a "combinatorial explosion" where duplicates multiply against each other, inflating your metrics. PySpark provides the .dropDuplicates() method (or .distinct()) which, when called without arguments, evaluates every column in the row and removes exact identical matches.

Step 2: Joining on ProductID

joined_df = clean_products.join(clean_processes, on="ProductID", how="inner")

To combine the product details with their respective manufacturing steps, we use a standard .join(). Because ProductID exists in both DataFrames, we pass it as the on parameter. An inner join ensures we only keep records that successfully match across both tables.

Step 3: Selecting and Reordering Columns

result_df = joined_df.select(
    "ProductID", "ProductName", "Category", 
    "ProcessID", "ProcessName", "Duration"
)

By default, PySpark will usually place the join key (ProductID) at the front of the resulting DataFrame, but the remaining columns might be in an unpredictable order. Using .select() at the very end guarantees that the final output matches the requested schema exactly.