Join PySpark Orders with Customer and Product Lookups

Introduction

An order knows its customer and product identifiers, while separate reference tables describe where the customer lives and which category the product belongs to. A join connects records through those keys so a report can answer questions such as revenue by region or product category.

You will enrich paid orders one lookup at a time, select unambiguous output columns, and reconcile totals. You will also compare INNER and LEFT JOIN when a customer is missing. The supplied customer and product keys are unique; each matching order should find one record in each lookup. Duplicate lookup keys could multiply joined rows and inflate revenue, so the intended relationship matters.

This independent lab prepares clean orders, two customers, three products, and a separate missing-match example. You should know typed reads, filters, decimal amounts, aggregation, and minimal Parquet output. No Hive, SQL, Hadoop, pandas, or cluster setup is required. Use the Terminal.

Add Customer Regions Through an Explicit Key

In this step, you will join the four paid orders to their unique customer records. Inspect the source relationship:

cd /home/labex/project

cat /home/labex/project/data/orders.csv

cat /home/labex/project/data/customers.csv

Several orders can share a customer_id, while the lookup contains one row for c1 and one for c2. Equal identifiers connect records; matching row positions would not be a valid rule. c1 supplies East and c2 supplies West.

DataFrame join accepts the other DataFrame, a matching condition, and a join type. inner keeps only matching records. DataFrame alias("o") and alias("c") give the inputs short reference names. col("o.customer_id") identifies the order-side column; col("c.customer_id") identifies the lookup-side column.

Both inputs contain a customer_id column. The explicit select chooses the order-side key and each desired field, avoiding ambiguous duplicate names. A DataFrame alias names an input reference; a column alias names a result field. Save and run this complete version, whose reader and amount calculation repeat earlier skills:

cat > /home/labex/project/enrich_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum as spark_sum
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType

spark = SparkSession.builder.appName("EnrichOrders").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),
])
orders = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")
paid = orders.filter(col("order_status") == "paid").withColumn("order_amount", col("quantity") * col("unit_price"))
customers = spark.read.option("header", True).schema("customer_id string, region string").csv("/home/labex/project/data/customers.csv")
products = spark.read.option("header", True).schema("product_id string, category string").csv("/home/labex/project/data/products.csv")

with_customers = (
    paid.alias("o")
    .join(customers.alias("c"), col("o.customer_id") == col("c.customer_id"), "inner")
    .select(
        col("o.order_id").alias("order_id"), col("o.customer_id").alias("customer_id"),
        col("o.product_id").alias("product_id"), col("o.quantity").alias("quantity"),
        col("o.unit_price").alias("unit_price"), col("o.order_status").alias("order_status"),
        col("o.order_date").alias("order_date"), col("o.order_amount").alias("order_amount"),
        col("c.region").alias("region"),
    )
)
with_customers.orderBy("order_id").show(truncate=False)
with_customers.write.mode("overwrite").parquet("/home/labex/project/output/orders-with-regions")
spark.stop()
PYTHON

bigdata-python /home/labex/project/enrich_orders.py > /home/labex/project/output/customer-join.txt

cat /home/labex/project/output/customer-join.txt

Expect four enriched paid rows. o1 and o4 should be East, o2 and o6 West; their amounts remain 20.00, 25.00, 50.00, and 30.00. Four paid inputs with complete unique matches produce four joined rows. The Parquet output preserves the typed fields and adds region without changing the raw CSV.

Add Product Categories and Reconcile Revenue

In this step, you will add a second lookup using product_id. Inspect the product mapping:

cat /home/labex/project/data/products.csv

p1 and p2 belong to Office; p3 belongs to Home. Joining by customer_id would be the wrong relationship for this table. Keep the intermediate enriched order fields and select only the category from the product lookup.

This full version retains the first join and adds the second. The explicit projection keeps one product_id rather than leaving two identically named keys. Then the familiar groupBy and aggregate sum calculate category revenue:

cat > /home/labex/project/enrich_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum as spark_sum
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType

spark = SparkSession.builder.appName("EnrichOrders").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),
])
orders = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")
paid = orders.filter(col("order_status") == "paid").withColumn("order_amount", col("quantity") * col("unit_price"))
customers = spark.read.option("header", True).schema("customer_id string, region string").csv("/home/labex/project/data/customers.csv")
products = spark.read.option("header", True).schema("product_id string, category string").csv("/home/labex/project/data/products.csv")

with_customers = (
    paid.alias("o")
    .join(customers.alias("c"), col("o.customer_id") == col("c.customer_id"), "inner")
    .select(
        col("o.order_id").alias("order_id"), col("o.customer_id").alias("customer_id"),
        col("o.product_id").alias("product_id"), col("o.quantity").alias("quantity"),
        col("o.unit_price").alias("unit_price"), col("o.order_status").alias("order_status"),
        col("o.order_date").alias("order_date"), col("o.order_amount").alias("order_amount"),
        col("c.region").alias("region"),
    )
)
enriched = (
    with_customers.alias("e")
    .join(products.alias("p"), col("e.product_id") == col("p.product_id"), "inner")
    .select(
        col("e.order_id").alias("order_id"), col("e.customer_id").alias("customer_id"),
        col("e.product_id").alias("product_id"), col("e.quantity").alias("quantity"),
        col("e.unit_price").alias("unit_price"), col("e.order_status").alias("order_status"),
        col("e.order_date").alias("order_date"), col("e.order_amount").alias("order_amount"),
        col("e.region").alias("region"), col("p.category").alias("category"),
    )
)
category_sales = enriched.groupBy("category").agg(spark_sum("order_amount").alias("revenue"))
enriched.write.mode("overwrite").parquet("/home/labex/project/output/enriched-orders")
category_sales.write.mode("overwrite").parquet("/home/labex/project/output/category-sales")
enriched.orderBy("order_id").show(truncate=False)
category_sales.orderBy("category").show(truncate=False)
spark.stop()
PYTHON

bigdata-python /home/labex/project/enrich_orders.py > /home/labex/project/output/product-join.txt

cat /home/labex/project/output/product-join.txt

The enriched result should still contain four paid orders, now with both region and category. The category report should show Home 30.00 and Office 95.00. Their total is the same 125.00 paid revenue.

The row count staying four and the reconciled revenue provide two useful checks. Neither should increase just because descriptive columns were added. This fixture's keys are unique and complete; the next step explores a missing match separately rather than changing the main sales totals.

Compare INNER and LEFT JOIN with a Missing Customer

In this step, you will observe what each join type does when the lookup cannot match. Inspect the independent example:

cat /home/labex/project/data/example-orders.csv

demo1 refers to c1, while demo2 refers to c_missing, which is absent from customers. Predict INNER JOIN first: only demo1 should remain. LEFT JOIN keeps every left-side example record and supplies NULL for fields that would have come from a missing right-side record.

Save a small separate program so the main enriched sales script and its data remain unchanged:

cat > /home/labex/project/compare_joins.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col

spark = SparkSession.builder.appName("CompareCustomerMatches").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
examples = spark.read.option("header", True).schema("order_id string, customer_id string").csv("/home/labex/project/data/example-orders.csv")
customers = spark.read.option("header", True).schema("customer_id string, region string").csv("/home/labex/project/data/customers.csv")
inner = examples.alias("o").join(customers.alias("c"), col("o.customer_id") == col("c.customer_id"), "inner").select(col("o.order_id").alias("order_id"), col("o.customer_id").alias("customer_id"), col("c.region").alias("region"))
left = examples.alias("o").join(customers.alias("c"), col("o.customer_id") == col("c.customer_id"), "left").select(col("o.order_id").alias("order_id"), col("o.customer_id").alias("customer_id"), col("c.region").alias("region"))
print("Inner matches:")
inner.orderBy("order_id").show(truncate=False)
print("Left matches:")
left.orderBy("order_id").show(truncate=False)
inner.write.mode("overwrite").parquet("/home/labex/project/output/example-inner")
left.write.mode("overwrite").parquet("/home/labex/project/output/example-left")
spark.stop()
PYTHON

bigdata-python /home/labex/project/compare_joins.py > /home/labex/project/output/join-comparison.txt

cat /home/labex/project/output/join-comparison.txt

Inner matches should contain demo1/c1/East. Left matches should contain that row plus demo2/c_missing/NULL. The left-side customer_id remains available even though the right-side region is missing. NULL is absence of a matched region, not a region literally named NULL.

INNER JOIN may be appropriate for a report whose complete matches are already established. LEFT JOIN can retain unresolved records for quality checks or a deliberate business policy. Choosing the join type changes which records survive; it is not merely a display preference.

Use the Enriched Fields for a Regional Comparison

In this step, you will apply already learned filtering and grouping to the main enriched dataset. All-category regional revenue should be East 70.00 and West 55.00. Predict the Office-only result: East stays 70.00, while West becomes 25.00 because its 30.00 Home order is excluded.

Run the complete main script with two additional derived reports. Its joins remain the same; the category condition changes only which enriched order rows enter the second aggregation:

cat > /home/labex/project/enrich_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum as spark_sum
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType

spark = SparkSession.builder.appName("EnrichOrders").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),
])
orders = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")
paid = orders.filter(col("order_status") == "paid").withColumn("order_amount", col("quantity") * col("unit_price"))
customers = spark.read.option("header", True).schema("customer_id string, region string").csv("/home/labex/project/data/customers.csv")
products = spark.read.option("header", True).schema("product_id string, category string").csv("/home/labex/project/data/products.csv")

with_customers = (
    paid.alias("o")
    .join(customers.alias("c"), col("o.customer_id") == col("c.customer_id"), "inner")
    .select(
        col("o.order_id").alias("order_id"), col("o.customer_id").alias("customer_id"),
        col("o.product_id").alias("product_id"), col("o.quantity").alias("quantity"),
        col("o.unit_price").alias("unit_price"), col("o.order_status").alias("order_status"),
        col("o.order_date").alias("order_date"), col("o.order_amount").alias("order_amount"),
        col("c.region").alias("region"),
    )
)
enriched = (
    with_customers.alias("e")
    .join(products.alias("p"), col("e.product_id") == col("p.product_id"), "inner")
    .select(
        col("e.order_id").alias("order_id"), col("e.customer_id").alias("customer_id"),
        col("e.product_id").alias("product_id"), col("e.quantity").alias("quantity"),
        col("e.unit_price").alias("unit_price"), col("e.order_status").alias("order_status"),
        col("e.order_date").alias("order_date"), col("e.order_amount").alias("order_amount"),
        col("e.region").alias("region"), col("p.category").alias("category"),
    )
)
category_sales = enriched.groupBy("category").agg(spark_sum("order_amount").alias("revenue"))
enriched.write.mode("overwrite").parquet("/home/labex/project/output/enriched-orders")
category_sales.write.mode("overwrite").parquet("/home/labex/project/output/category-sales")
enriched.orderBy("order_id").show(truncate=False)
category_sales.orderBy("category").show(truncate=False)
regional = enriched.groupBy("region").agg(spark_sum("order_amount").alias("revenue"))
office = enriched.filter(col("category") == "Office").groupBy("region").agg(spark_sum("order_amount").alias("revenue"))
regional.write.mode("overwrite").parquet("/home/labex/project/output/regional-sales")
office.write.mode("overwrite").parquet("/home/labex/project/output/office-regional-sales")
print("All categories by region:")
regional.orderBy("region").show(truncate=False)
print("Office category by region:")
office.orderBy("region").show(truncate=False)
spark.stop()
PYTHON

bigdata-python /home/labex/project/enrich_orders.py > /home/labex/project/output/regional-comparison.txt

cat /home/labex/project/output/regional-comparison.txt

Expect the predicted all-category and Office-only regional results. The Office regions add back to 95.00, matching the Office category subtotal from the preceding step. The overall regions add back to 125.00.

These reconciliations explain both the join relationship and the selected business subset. Your main enriched Parquet dataset still contains the four correctly typed paid orders, and the missing-match demonstration remains separate. Next, you will express comparable operations through Spark SQL over a DataFrame view.

Summary

You enriched orders through explicit unique customer and product keys, selected unambiguous output columns, and reconciled category and regional totals. INNER and LEFT JOIN produced deliberately different missing-match behavior. Next, you will compare Spark SQL and DataFrame expressions over the same structured data.