PySpark DataFrame Practice Questions

PySpark DataFrames provide a powerful and user-friendly API for working with structured and semi-structured data. In this article, we present a set of practice questions to help you reinforce your understanding of PySpark DataFrames and their operations.

1. Loading Data

Load the "sales_data.csv" file into a PySpark DataFrame. The CSV file contains the following columns: "transaction_id", "customer_id", "product_name", "quantity", and "price". Ensure that the DataFrame correctly infers the schema and displays the first 5 rows.

from pyspark.sql import SparkSession

# Initialize SparkSession
spark = SparkSession.builder.appName("DataFramePractice").getOrCreate()

# Load the CSV file into a DataFrame
df = spark.read.csv("sales_data.csv", header=True, inferSchema=True)

# Display the first 5 rows of the DataFrame
df.show(5)

2. Filtering Data

Filter the DataFrame to show only the transactions where the "quantity" is greater than or equal to 10 and the "price" is less than 50.

filtered_df = df.filter((df["quantity"] >= 10) & (df["price"] < 50))
filtered_df.show()

3. Grouping and Aggregating Data

Find the total revenue generated by each product and display the results in descending order.

from pyspark.sql import functions as F

revenue_df = df.groupBy("product_name").agg(F.sum("quantity" * "price").alias("total_revenue"))
revenue_df = revenue_df.orderBy(F.desc("total_revenue"))
revenue_df.show()

4. Joining DataFrames

Load the "customer_data.csv" file into another DataFrame. The CSV file contains the following columns: "customer_id", "customer_name", and "email". Perform an inner join between the "df" DataFrame and the "customer_df" DataFrame based on the "customer_id" column and display the results.

customer_df = spark.read.csv("customer_data.csv", header=True, inferSchema=True)

joined_df = df.join(customer_df, on="customer_id", how="inner")
joined_df.show()

5. Data Transformation

Transform the DataFrame to add a new column "total_amount" that represents the total amount for each transaction (quantity * price).

from pyspark.sql import functions as F

# Add a new column "total_amount"
df = df.withColumn("total_amount", df["quantity"] * df["price"])
df.show()

6. Handling Missing Values

Count the number of missing values in each column of the DataFrame and display the results.

missing_values_df = df.select([F.count(F.when(F.isnan(c) | F.col(c).isNull(), c)).alias(c) for c in df.columns])
missing_values_df.show()

7. Data Visualization

Visualize the distribution of the "quantity" column using a histogram.

import matplotlib.pyplot as plt

# Convert DataFrame column to Pandas Series
quantity_series = df.select("quantity").toPandas()["quantity"]

# Plot histogram
plt.hist(quantity_series, bins=20, edgecolor='black')
plt.xlabel("Quantity")
plt.ylabel("Frequency")
plt.title("Distribution of Quantity")
plt.show()

8. Working with Dates

Convert the "transaction_date" column to a DateType and extract the year from it into a new column "transaction_year".

from pyspark.sql.functions import year, to_date

# Convert "transaction_date" column to DateType
df = df.withColumn("transaction_date", to_date("transaction_date", "yyyy-MM-dd"))

# Extract year and create "transaction_year" column
df = df.withColumn("transaction_year", year("transaction_date"))
df.show()

9. Data Aggregation and Window Functions

Calculate the average quantity of each product for the last three transactions using a window function.

from pyspark.sql import Window

# Define a Window specification
window_spec = Window.partitionBy("product_name").orderBy(F.desc("transaction_date")).rowsBetween(0, 2)

# Calculate average quantity for the last three transactions
df = df.withColumn("avg_quantity_last_three", F.avg("quantity").over(window_spec))
df.show()

10. Pivot Table

Create a pivot table that shows the total quantity of each product for each year.

# Pivot table
pivot_table = df.groupBy("product_name").pivot("transaction_year").agg(F.sum("quantity"))

pivot_table.show()

11. String Manipulation

Create a new column "upper_product_name" that contains the product names in uppercase.

# Uppercase product names
df = df.withColumn("upper_product_name", F.upper("product_name"))
df.show()

12. User-Defined Functions (UDFs)

Create a UDF that calculates the total amount for each transaction and apply it to the DataFrame to add a new column "total_amount_udf".

# User-Defined Function
def calculate_total_amount(quantity, price):
    return quantity * price

# Register UDF
spark.udf.register("calculate_total_amount_udf", calculate_total_amount)

# Apply UDF to create "total_amount_udf" column
df = df.withColumn("total_amount_udf", F.expr("calculate_total_amount_udf(quantity, price)"))
df.show()

13. Joins and Aggregations

Join the "df" DataFrame with the "customer_df" DataFrame using the "customer_id" column. Then, find the total revenue generated by each customer and display the results in descending order.

joined_df = df.join(customer_df, on="customer_id", how="inner")

revenue_by_customer = joined_df.groupBy("customer_name").agg(F.sum("quantity" * "price").alias("total_revenue"))
revenue_by_customer = revenue_by_customer.orderBy(F.desc("total_revenue"))

revenue_by_customer.show()

14. Filtering and Date Manipulation

Filter the DataFrame to show only the transactions that occurred after a specific date, and calculate the total revenue for each day.

from pyspark.sql.functions import col

# Define the specific date
specific_date = "2023-08-01"

# Filter transactions after the specific date
filtered_df = df.filter(col("transaction_date") > specific_date)

# Calculate total revenue for each day
daily_revenue = filtered_df.groupBy("transaction_date").agg(F.sum("quantity" * "price").alias("total_revenue"))
daily_revenue.show()

15. Working with Arrays

Create a new column "product_list" that contains an array of product names for each transaction.

# Working with Arrays
df = df.withColumn("product_list", F.array("product_name"))
df.show()

16. Window Functions and Ranking

Rank the customers based on their total revenue generated, and show the top 5 customers.

# Rank customers based on total revenue
ranked_customers = revenue_by_customer.withColumn("rank", F.rank().over(Window.orderBy(F.desc("total_revenue"))))

# Show top 5 customers
top_5_customers = ranked_customers.filter(F.col("rank") <= 5)
top_5_customers.show()

17. Data Deduplication

Remove duplicate rows from the DataFrame based on all columns and display the deduplicated DataFrame.

# Deduplicate rows based on all columns
deduplicated_df = df.dropDuplicates()

deduplicated_df.show()

18. Data Sampling

Take a random sample of 10% of the DataFrame and display the sampled data.

# Random sampling of 10%
sampled_df = df.sample(withReplacement=False, fraction=0.1)

sampled_df.show()

19. Data Reshaping

Melt the DataFrame from a wide format to a long format by unpivoting the columns "product_1", "product_2", "product_3" into a single "product" column.

# Data Melt - Unpivot the columns
from pyspark.sql.functions import array, concat_ws, explode

melted_df = df.withColumn("product", explode(array("product_1", "product_2", "product_3"))) \
    .select("transaction_id", "customer_id", "product", "quantity", "price")

melted_df.show()

20. Handling Null Values

Replace the null values in the "product_name" column with a default value "Unknown".

# Replace null values in "product_name" column
df = df.fillna("Unknown", subset=["product_name"])

df.show()