Introduction
A correctly typed order table is a starting point, not yet an answer. A colleague may want just a few columns, only paid orders, and an amount calculated from quantity and unit price. You will express those decisions as Spark DataFrame operations while keeping the original records available.
A column expression describes a calculation or condition for Spark to apply to rows. You will select fields, combine conditions, add a decimal amount, and give a result column a readable name. These operations create new DataFrames; actions such as show request actual results. The distinction helps explain Spark's lazy execution without needing scheduling internals yet.
This lab independently supplies six clean retail orders. You should understand the explicit CSV schema from the previous lab and basic Python. The full reader and SparkSession are included in each runnable version, so no earlier process or file is needed. Use the Terminal and the prepared bigdata-python environment.
Select the Columns Needed for a Question
In this step, you will narrow the table's columns while retaining all six order records. Start with the source:
cd /home/labex/project
cat /home/labex/project/data/orders.csv
select chooses which columns a new DataFrame exposes and in what order. It does not filter rows. The complete script below repeats the familiar explicit reader, then selects the order identifier, quantity, price, and status. The date and customer fields remain in df, even though selected does not expose them.
Save the script with a quoted here-document and run it. > writes the file or replaces the previous version; the final PYTHON marker ends the source text:
cat > /home/labex/project/transform_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType
spark = SparkSession.builder.appName("TransformOrders").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
order_schema = StructType([
StructField("order_id", StringType(), True),
StructField("customer_id", StringType(), True),
StructField("product_id", StringType(), True),
StructField("quantity", IntegerType(), True),
StructField("unit_price", DecimalType(10, 2), True),
StructField("order_status", StringType(), True),
StructField("order_date", DateType(), True),
])
df = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")
selected = df.select("order_id", "quantity", "unit_price", "order_status")
selected.orderBy("order_id").show(truncate=False)
spark.stop()
PYTHON
bigdata-python /home/labex/project/transform_orders.py > /home/labex/project/output/selected-orders.txt
cat /home/labex/project/output/selected-orders.txt
Expect all six order IDs with four displayed columns. orderBy makes the display predictable, while show triggers the work needed to return rows. This is a column selection, sometimes called a projection. No source file was changed and no orders were removed.
Combine Conditions to Select Rows
In this step, you will keep paid orders with a quantity of at least two. col("order_status") refers to a Spark column; comparing it with == "paid" builds a condition for Spark, rather than a single Python True or False. >= builds the quantity condition.
Use & to combine Spark column conditions with AND semantics. Put parentheses around each comparison because Python operator precedence matters. Python's ordinary and is not the operator for combining Spark Column objects.
Predict the qualifying IDs: o1, o4, and o6 meet both conditions. o2 is paid but has quantity one. Replace and run the complete script:
cat > /home/labex/project/transform_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType
spark = SparkSession.builder.appName("TransformOrders").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
order_schema = StructType([
StructField("order_id", StringType(), True),
StructField("customer_id", StringType(), True),
StructField("product_id", StringType(), True),
StructField("quantity", IntegerType(), True),
StructField("unit_price", DecimalType(10, 2), True),
StructField("order_status", StringType(), True),
StructField("order_date", DateType(), True),
])
df = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")
eligible = df.filter((col("order_status") == "paid") & (col("quantity") >= 2))
eligible.select("order_id", "quantity", "unit_price", "order_status").orderBy("order_id").show(truncate=False)
spark.stop()
PYTHON
bigdata-python /home/labex/project/transform_orders.py > /home/labex/project/output/eligible-orders.txt
cat /home/labex/project/output/eligible-orders.txt
The result should contain only those three IDs, each with quantity 2. filter changes the selected rows; select controls the displayed columns. You can combine both operations while df continues to describe the complete six-row source.
This step's quantity threshold is a demonstration condition, not a general definition of sales. The next calculation will include all four paid orders, including o2, because each paid order contributes revenue.
Calculate and Name Decimal Order Amounts
In this step, you will compute each paid order's amount as quantity multiplied by decimal unit price. withColumn adds a named column to a new DataFrame, or replaces that column if its name already exists. Here order_amount is calculated using Spark's built-in multiplication expression.
The following select retains the order ID and gives the calculated column the display name amount using alias. An alias changes the result's column name; it does not edit the CSV header. All four paid orders belong in this calculation; pending and refunded rows are excluded, without treating refunds as negative revenue in this simplified rule.
Run this complete version:
cat > /home/labex/project/transform_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType
spark = SparkSession.builder.appName("TransformOrders").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
order_schema = StructType([
StructField("order_id", StringType(), True),
StructField("customer_id", StringType(), True),
StructField("product_id", StringType(), True),
StructField("quantity", IntegerType(), True),
StructField("unit_price", DecimalType(10, 2), True),
StructField("order_status", StringType(), True),
StructField("order_date", DateType(), True),
])
df = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")
paid_result = (
df.filter(col("order_status") == "paid")
.withColumn("order_amount", col("quantity") * col("unit_price"))
.select("order_id", col("order_amount").alias("amount"))
)
print("Paid orders:")
paid_result.orderBy("order_id").show(truncate=False)
paid_result.printSchema()
spark.stop()
PYTHON
bigdata-python /home/labex/project/transform_orders.py > /home/labex/project/output/paid-amounts.txt
cat /home/labex/project/output/paid-amounts.txt
Expect o1/20.00, o2/25.00, o4/50.00, and o6/30.00. Extra trailing zeroes may appear. The schema should identify amount as a decimal type. Spark may widen decimal precision during multiplication; the important point is that the calculation retains decimal arithmetic rather than converting prices to binary floating point.
filter, withColumn, select, and orderBy describe transformations and return new DataFrames. Defining this chain does not immediately materialize every intermediate table. show is an action asking Spark to execute the required work and display an answer. printSchema describes column structure. Later you will inspect the actual execution plan behind such a chain.
Change One Status and Predict Its Effect
In this step, you will retain the paid result and add a separate pending-order practice view. Predict its output before running: only o3 is pending, with quantity one and price 10.00, so its amount is 10.00.
The new pending_result uses the same expression chain with only the status value changed. It starts from df, not from the already filtered paid_result, so pending records are still available:
cat > /home/labex/project/transform_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType
spark = SparkSession.builder.appName("TransformOrders").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
order_schema = StructType([
StructField("order_id", StringType(), True),
StructField("customer_id", StringType(), True),
StructField("product_id", StringType(), True),
StructField("quantity", IntegerType(), True),
StructField("unit_price", DecimalType(10, 2), True),
StructField("order_status", StringType(), True),
StructField("order_date", DateType(), True),
])
df = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")
paid_result = (
df.filter(col("order_status") == "paid")
.withColumn("order_amount", col("quantity") * col("unit_price"))
.select("order_id", col("order_amount").alias("amount"))
)
print("Paid orders:")
paid_result.orderBy("order_id").show(truncate=False)
paid_result.printSchema()
pending_result = (
df.filter(col("order_status") == "pending")
.withColumn("order_amount", col("quantity") * col("unit_price"))
.select("order_id", col("order_amount").alias("amount"))
)
print("Pending practice:")
pending_result.orderBy("order_id").show(truncate=False)
spark.stop()
PYTHON
bigdata-python /home/labex/project/transform_orders.py > /home/labex/project/output/status-comparison.txt
cat /home/labex/project/output/status-comparison.txt
Under Paid orders, the same four paid amounts should remain. Under Pending practice, expect only o3/10.00. Filtering the paid view for pending status would return no rows because those records were already excluded; starting from the appropriate source matters.
Both chains are evaluated by Spark's built-in expressions. You did not collect the general dataset into a Python list and loop over it. This keeps the processing description suitable for Spark's execution engine. Your final script retains the paid calculation and the small practice comparison, while the original CSV still contains all six orders.
Summary
You selected columns, combined row conditions with Spark expressions, calculated decimal order amounts, and named output columns. Comparing paid and pending views showed why transformations should start from the appropriate source and how actions request results. Next, you will apply familiar reading and selection ideas to JSON input.
