Objective

You are analyzing the Private Equity (PE) landscape. You have three datasets: pe_firms (the parent companies), pe_funds (the financial vehicles managed by the firms), and pe_investments (the individual companies those funds bought into).

Task

Write a PySpark script that combines all three DataFrames into a single comprehensive view.

Because the data is messy, some investments may not map to a known fund, and some funds may not map to a known firm. You must use a join strategy that ensures no records are lost from any of the three tables during the merge. After merging, drop any rows where absolutely every column is null.

Save the resulting DataFrame as result_df. Select the columns in the exact order specified in the schema below.

File Path

• Firms: /home/interview/pe_firms.csv
• Funds: /home/interview/pe_funds.csv
• Investments: /home/interview/pe_investments.csv
• Starter script: /home/interview/pe_merger.py

Schema

pe_firms.csv

pe_funds.csv

pe_investments.csv

Expected Output Schema

Example

Given this sample input:

pe_firms

pe_funds

pe_investments

The output would be:

from pyspark.sql import SparkSession

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

firms = spark.read.csv("/home/interview/pe_firms.csv", header=True, inferSchema=True)
funds = spark.read.csv("/home/interview/pe_funds.csv", header=True, inferSchema=True)
investments = spark.read.csv("/home/interview/pe_investments.csv", header=True, inferSchema=True)

# Use full outer joins to retain records missing in either direction
joined_df = firms.join(funds, on="firm_id", how="outer") \
    .join(investments, on="fund_id", how="outer")

# Drop rows where every single column is null
cleaned_df = joined_df.dropna(how="all")

# Reorder columns to match the requested output schema
result_df = cleaned_df.select(
    "investment_id", "fund_id", "firm_id", "firm_name", "founded_year",
    "location", "fund_name", "fund_size", "fund_start_year", "fund_end_year",
    "company_name", "investment_amount", "investment_date"
)

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

Explanation

Step 1: The Full Outer Join

joined_df = firms.join(funds, on="firm_id", how="outer") \
    .join(investments, on="fund_id", how="outer")

When you have messy data where foreign keys might be missing (e.g., a fund that doesn't belong to a firm, or an investment without a fund), an inner join or left join will drop records. A full outer join ensures that all rows from both tables are kept. If there is no match, PySpark fills the missing sides with null values.

Step 2: Dropping All-Null Rows

cleaned_df = joined_df.dropna(how="all")

The dropna() method (an alias for na.drop()) allows you to remove rows containing nulls. By default, it uses how="any", which drops a row if even one column is null. By specifying how="all", PySpark will only drop a row if absolutely every single column in that row is completely empty.

Step 3: Enforcing the Schema Order

result_df = cleaned_df.select("investment_id", "fund_id", ...)

When performing multiple joins, PySpark rearranges columns, usually putting the join keys (firm_id, fund_id) at the very front. Using a final .select() block allows you to guarantee the final column order matches exactly what the downstream systems (or the prompt instructions) expect.