Objective

A bank maintains its transactions and customer data in separate databases. However, for certain exhaustive data analysis tasks and risk models, the bank needs to perform a cross join operation between the transactions and customers datasets.

Task

Write a PySpark function that performs a cross join operation on these two DataFrames.

Because both DataFrames contain a cust_id column, a direct cross join will result in duplicate column names. To resolve this and satisfy the Expected Output Schema (which only asks for one cust_id), drop the cust_id column from the customers DataFrame before performing the cross join.

Save your resulting DataFrame as result_df. Ensure the output matches the exact schema order requested, and order the final output by trans_id (ascending), then by first_name (ascending).

File Path

• Transactions Dataset: /home/interview/transactions.csv
• Customers Dataset: /home/interview/customers.csv
• Starter script: /home/interview/cross_join.py

Schema

transactions.csv

customers.csv

Expected Output Schema

Example

Given this sample input:

transactions

customers

The expected output would be:

(Note: Every transaction is paired with every customer, creating a Cartesian product).

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

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

transactions = spark.read.csv("/home/interview/transactions.csv", header=True, inferSchema=True)
customers = spark.read.csv("/home/interview/customers.csv", header=True, inferSchema=True)

# Step 1: Drop the overlapping cust_id column from customers to prevent ambiguity
customers_clean = customers.drop("cust_id")

# Step 2: Perform the cross join (Cartesian product)
joined_df = transactions.crossJoin(customers_clean)

# Step 3: Organize final columns to match the output schema exactly and sort
result_df = joined_df.select(
    "trans_id", "trans_amt", "date", "cust_id", 
    "first_name", "last_name", "age"
).orderBy("trans_id", "first_name")

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

Explanation

Step 1: Dropping Duplicate Columns

customers_clean = customers.drop("cust_id")

A cross join blindly merges every row from Table A with every row from Table B without a matching key. Because both tables have a column named cust_id, the resulting DataFrame would end up with two identically named columns. To prevent PySpark from throwing an "ambiguous column" error later, we simply drop cust_id from the customers table before executing the join.

Step 2: Executing the Cross Join

joined_df = transactions.crossJoin(customers_clean)

A cross join (often called a Cartesian Product) creates every possible combination of rows. If the transactions table has 10 rows and the customers table has 5 rows, .crossJoin() will output exactly 50 rows (10 x 5). It does not require an on="..." condition because it isn't trying to match matching records-it pairs everything.

Step 3: Schema Arrangement and Sorting

result_df = joined_df.select(
    "trans_id", "trans_amt", "date", "cust_id", 
    "first_name", "last_name", "age"
).orderBy("trans_id", "first_name")

Finally, we chain a .select() block to ensure the final output columns are arranged in the precise order specified by the prompt's Expected Output Schema. We then apply .orderBy("trans_id", "first_name") to ensure the Cartesian product is returned in a predictable, deterministic sequence.