Column expressions and functions in Apache Spark - Mini Project: Build & Apply

Column expressions and functions

📖 Scenario: You work as a data analyst for a retail company. You have sales data with product names and their prices. You want to create a new column that shows the price after applying a 10% discount.

🎯 Goal: Create a Spark DataFrame with product names and prices, define a discount rate, apply a column expression to calculate discounted prices, and display the final DataFrame.

📋 What You'll Learn

Create a Spark DataFrame named products_df with columns product and price using the exact data provided.

Create a variable named discount_rate and set it to 0.10.

Use Spark column expressions and functions to add a new column discounted_price to products_df that applies the discount.

Show the resulting DataFrame using show().

💡 Why This Matters

🌍 Real World

Retail companies often need to adjust prices dynamically, such as applying discounts or taxes, and analyze the updated prices.

💼 Career

Data analysts and data engineers use Spark column expressions to efficiently transform and analyze large datasets in real time.

1

Create the initial DataFrame

Create a Spark DataFrame called products_df with the following data: [('Apple', 100), ('Banana', 80), ('Cherry', 120)]. The columns should be named product and price.

Apache Spark

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName('DiscountApp').getOrCreate()
# Create products_df DataFrame with columns 'product' and 'price' and the exact data
# Your code here

Hint

Use spark.createDataFrame() with a list of tuples and specify the column names.

2

Define the discount rate

Create a variable called discount_rate and set it to 0.10 to represent a 10% discount.

Apache Spark

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName('DiscountApp').getOrCreate()

products_data = [('Apple', 100), ('Banana', 80), ('Cherry', 120)]
products_df = spark.createDataFrame(products_data, ['product', 'price'])
# Create discount_rate variable and set it to 0.10
# Your code here

Hint

Just assign 0.10 to a variable named discount_rate.

3

Add discounted price column

Use Spark column expressions to add a new column called discounted_price to products_df. Calculate it by subtracting the discount from the original price using the discount_rate. Use the withColumn method and the col function from pyspark.sql.functions.

Apache Spark

from pyspark.sql import SparkSession
from pyspark.sql.functions import col

spark = SparkSession.builder.appName('DiscountApp').getOrCreate()

products_data = [('Apple', 100), ('Banana', 80), ('Cherry', 120)]
products_df = spark.createDataFrame(products_data, ['product', 'price'])

discount_rate = 0.10
# Add a new column 'discounted_price' to products_df using withColumn and col
# Your code here

Hint

Use withColumn('discounted_price', col('price') * (1 - discount_rate)) to create the new column.

4

Show the final DataFrame

Use the show() method on products_df to display the DataFrame with the new discounted_price column.

Apache Spark

from pyspark.sql import SparkSession
from pyspark.sql.functions import col

spark = SparkSession.builder.appName('DiscountApp').getOrCreate()

products_data = [('Apple', 100), ('Banana', 80), ('Cherry', 120)]
products_df = spark.createDataFrame(products_data, ['product', 'price'])

discount_rate = 0.10

products_df = products_df.withColumn('discounted_price', col('price') * (1 - discount_rate))
# Show the DataFrame
# Your code here

Hint

Just call products_df.show() to display the table.

Practice

1. What does the withColumn function do in Apache Spark DataFrame?

easy

A. It adds a new column or replaces an existing column with a new value.

B. It deletes a column from the DataFrame.

C. It filters rows based on a condition.

D. It groups data by a column.

Solution

1. Step 1: Understand the purpose of withColumn

   The withColumn function is used to add a new column or replace an existing one in a DataFrame.

2. Step 2: Compare with other functions

   Deleting columns is done by drop, filtering by filter, and grouping by groupBy. So these are not correct.

3. Final Answer:

   It adds a new column or replaces an existing column with a new value. -> Option A

4. Quick Check:

   withColumn = add/replace column [OK]

Hint: Remember: withColumn adds or changes columns fast [OK]

Common Mistakes:

• Confusing withColumn with drop or filter
• Thinking it deletes columns
• Mixing it with groupBy

2. Which of the following is the correct syntax to select a column named age from a DataFrame df using Spark's column expression?

easy

A. df.select("age")

B. df.select(age)

C. df.select(df.age())

D. df.select(`age`)

Solution

1. Step 1: Understand select syntax

   The select method expects column names as strings or Column objects. Using double quotes around column names is the standard correct syntax.

2. Step 2: Check each option

   df.select("age") (A) uses a valid string literal.
   df.select(age) (B) references undefined variable age - error.
   df.select(df.age()) (C) tries to call df.age as a function - Column not callable.
   df.select(`age`) (D) uses invalid backtick syntax.

3. Step 3: Identify the correct syntax

   Only df.select("age") executes without error and selects the column correctly.

4. Final Answer:

   df.select("age") -> Option A

5. Quick Check:

   Column names as strings in select = correct [OK]

Hint: Use quotes around column names in select [OK]

Common Mistakes:

• Using variable names without quotes
• Confusing Column objects with strings
• Missing quotes causing errors

3. Given the following Spark code, what will be the output of df.show()?

from pyspark.sql import SparkSession
from pyspark.sql.functions import col

spark = SparkSession.builder.getOrCreate()
data = [(1, 10), (2, 20), (3, 30)]
df = spark.createDataFrame(data, ['id', 'value'])
df = df.withColumn('double_value', col('value') * 2)
df.show()

medium

A. +---+-----+------------+
|id |value|double_value|
+---+-----+------------+
|1 |10 |10 |
|2 |20 |20 |
|3 |30 |30 |
+---+-----+------------+

B. Error: col function not found

C. +---+-----+------------+
|id |value|double_value|
+---+-----+------------+
|1 |10 |20 |
|2 |20 |40 |
|3 |30 |60 |
+---+-----+------------+

D. +---+-----+
|id |value|
+---+-----+
|1 |10 |
|2 |20 |
|3 |30 |
+---+-----+

Solution

1. Step 1: Understand the data and transformation

   The DataFrame starts with columns 'id' and 'value' with values (1,10), (2,20), (3,30). The withColumn adds 'double_value' by multiplying 'value' by 2.

2. Step 2: Calculate the new column values

   For each row, double_value = value * 2, so 10*2=20, 20*2=40, 30*2=60.

3. Final Answer:

   The DataFrame will show three columns with double_value as twice value: -> Option C

4. Quick Check:

   double_value = value * 2 [OK]

Hint: Multiply column with col('name') * number [OK]

Common Mistakes:

• Forgetting to import col function
• Using wrong column name
• Expecting original DataFrame without new column

4. Identify the error in this Spark code snippet:

from pyspark.sql.functions import col

df = spark.createDataFrame([(1, 'Alice'), (2, 'Bob')], ['id', 'name'])
df = df.withColumn('name_length', length(col('name')))
df.show()

medium

A. Syntax error in withColumn usage

B. Incorrect column name 'name'

C. DataFrame creation syntax is wrong

D. Missing import for length function

Solution

1. Step 1: Check imports for used functions

   The code uses length function but only imports col. The length function must be imported from pyspark.sql.functions.

2. Step 2: Verify other parts

   Column name 'name' exists, syntax for withColumn is correct, and DataFrame creation is valid.

3. Final Answer:

   Missing import for length function -> Option D

4. Quick Check:

   Import all functions you use [OK]

Hint: Always import functions like length explicitly [OK]

Common Mistakes:

• Forgetting to import needed functions
• Assuming all functions are auto-imported
• Typos in column names

5. You have a DataFrame df with columns id, score1, and score2. You want to create a new column max_score that contains the maximum value between score1 and score2 for each row. Which code snippet correctly achieves this?

hard

A. df = df.withColumn('max_score', max(col('score1'), col('score2')))

B. from pyspark.sql.functions import greatest

df = df.withColumn('max_score', greatest(col('score1'), col('score2')))

C. df = df.withColumn('max_score', col('score1') if col('score1') > col('score2') else col('score2'))

D. df = df.withColumn('max_score', col('score1').max(col('score2')))

Solution

1. Step 1: Identify the correct function for max between columns

   Spark provides the greatest function to get the maximum value among columns for each row.

2. Step 2: Check each option

   from pyspark.sql.functions import greatest df = df.withColumn('max_score', greatest(col('score1'), col('score2'))) uses greatest correctly with imported function. df = df.withColumn('max_score', max(col('score1'), col('score2'))) uses max which is an aggregate function, not for row-wise max. df = df.withColumn('max_score', col('score1') if col('score1') > col('score2') else col('score2')) tries Python if-else which does not work on columns. df = df.withColumn('max_score', col('score1').max(col('score2'))) uses invalid syntax.

3. Final Answer:

   Use greatest function with columns -> Option B

4. Quick Check:

   greatest(col1, col2) = max per row [OK]

Hint: Use greatest() for max of columns per row [OK]

Common Mistakes:

• Using max() instead of greatest() for columns
• Trying Python if-else on columns
• Wrong method chaining syntax