DataFrame Filtering, Projection, Aggregation, and Conditional Logic

Start with the required result

Imagine an order feed with these columns: order_id, region, status, units, and unit_price_cents. A reporting team wants one summary row per region with:

• the number of valid paid orders;
• total revenue, calculated as units multiplied by unit price;
• revenue from “large” orders, whose totals are at least 3,000 cents.

Only paid orders with positive units qualify. The final report should include only regions with at least 3,000 cents in total revenue.

Before coding, separate two kinds of conditions. Order eligibility is decided on individual input rows, so it belongs before aggregation. Regional eligibility depends on a completed total, so it belongs after aggregation. That distinction will determine where the two filters go.

The following short demonstration shows the DataFrame vocabulary for selecting columns, naming expressions, adding a column, and filtering rows.

Filter rows, then project what you need

Here is a small, self-contained input you can run in a PySpark session or notebook. Prices are stored in integer cents, so the arithmetic in this example does not depend on floating-point currency calculations.

from pyspark.sql import functions as F

orders = spark.createDataFrame(
    [
        (101, "East",  "paid",      2, 1500),
        (102, "West",  "paid",      1, 2000),
        (103, "East",  "cancelled", 8, 1000),
        (104, "East",  "paid",      3, 1000),
        (105, "West",  "paid",      1,  500),
        (106, "West",  "paid",      4,  750),
        (107, "East",  "pending",   5,  900),
        (108, "East",  "paid",      0,  900),
        (109, "South", "paid",      1,  300),
    ],
    "order_id INT, region STRING, status STRING, "
    "units INT, unit_price_cents INT"
)

Express both eligibility rules in a single filter:

qualified = orders.filter(
    (F.col("status") == "paid") & (F.col("units") > 0)
)

F.col("status") == "paid" constructs a Column expression; it does not compare one Python value with another. The & combines two Column conditions. Parenthesize each comparison because Python’s operator precedence can otherwise make a compound predicate behave unexpectedly or fail. Use | for OR, and ~ to negate a Column condition. where(...) is an alternative name for filter(...).

This filter removes the cancelled and pending orders and the paid order with zero units. It retains six orders, including the small South order. South must remain for now: whether its region meets the reporting threshold cannot be known from one order row.

Next, projection means choosing the columns in the resulting DataFrame. select can also calculate and name expressions:

projected = qualified.select(
    F.col("order_id"),
    F.col("region"),
    (F.col("units") * F.col("unit_price_cents"))
        .alias("order_total_cents")
)

This produces just the fields needed downstream. The multiplication is evaluated for each qualifying row, and alias gives the expression a stable name for later steps. If instead you wanted to retain all existing columns and add order_total_cents, you could use withColumn("order_total_cents", ...). Neither method changes qualified in place; each returns a new DataFrame.

Notice the order of operations in the code: filter while status and units are available, then project them away once they are no longer needed. That makes the intended data contract clear, even though Spark may optimize how the operations are physically executed.

Express the business rule with when

Suppose the reporting team also wants every qualifying order labeled:

• large if its total is at least 3,000 cents;
• medium if it is at least 1,000 but below 3,000 cents;
• small otherwise.

Read the official PySpark reference for the behavior of when, particularly what happens when no condition matches.

Build the label from the projected order total:

enriched = projected.withColumn(
    "order_band",
    F.when(F.col("order_total_cents") >= 3000, "large")
     .when(F.col("order_total_cents") >= 1000, "medium")
     .otherwise("small")
)

The conditions are tested in order, and the first matching branch supplies the value. Order matters: if the >= 1000 branch came first, an order worth 3,000 cents would be labeled medium. An order worth exactly 3,000 is large, because the requirement says at least 3,000.

otherwise("small") makes the fallback explicit. Without otherwise, an unmatched row receives a null result. That may be correct for some requirements, but it is not the rule here.

There is an important distinction between this expression and a filter. The when expression keeps the order and assigns it a category. A filter removes the order when its condition is not satisfied.

Aggregate at the reporting grain

The requested output grain is one row per region. Grouping establishes that grain; aggregation calculates metrics over the qualifying orders in each group. The Azure Databricks PySpark guide shows groupBy(...).agg(...) and a chained transformation pipeline.

Now complete the report:

regional_report = (
    enriched
    .groupBy("region")
    .agg(
        F.count("*").alias("order_count"),
        F.sum("order_total_cents").alias("revenue_cents"),
        F.sum(
            F.when(
                F.col("order_band") == "large",
                F.col("order_total_cents")
            ).otherwise(F.lit(0))
        ).alias("large_revenue_cents")
    )
    .filter(F.col("revenue_cents") >= 3000)
    .orderBy("region")
)

regional_report.show()

The three aggregate expressions answer different questions:

• F.count("*") counts qualifying rows in each region. By contrast, F.count("some_column") counts non-null values of that column; it is not a distinct-customer count.
• F.sum("order_total_cents") adds every qualifying order total in the region.
• The final F.sum(...) is a conditional aggregation. Each large order contributes its total; every other order contributes zero.

The filter on revenue_cents happens after agg because it tests a regional total. In SQL terms, the paid-and-positive-units filter serves the role of WHERE, while this later regional-total filter serves the role of HAVING. Spark also needs the alias revenue_cents to exist before this DataFrame filter can reference it.

The expected report is:

South has one qualifying order worth 300 cents, so it is removed only by the post-aggregation filter. West’s 2,000-cent order counts toward total revenue but not large-order revenue. The final orderBy makes the displayed row order predictable; it is not required to calculate the metrics.

Explain the pipeline as well as the code

In an interview, a working snippet is stronger when you can state its grain and boundary conditions. For this solution, you could explain that input filtering excludes invalid orders, projection computes an order-level monetary value, when labels each order, and groupBy("region").agg(...) produces one summary row per region. The last filter applies the reporting threshold to those summary rows—not to individual orders.

This also reconnects the code to the previous lesson. Filtering, projection, and order-level conditional expressions are ordinarily partition-local. Grouping by region usually requires a shuffle so contributions from different partitions can be combined. The optional global orderBy may require further redistribution. regional_report remains a planned DataFrame until an action such as show() requests its results.

The key implementation habit is to decide what one row represents at each point. Before groupBy, a row represents an order; after it, a row represents a region. That single distinction prevents many misplaced-filter and incorrect-count errors.

Conclusion

You have built a complete DataFrame transformation from a reporting requirement: filter eligible input rows, project and derive the needed fields, apply ordered conditional rules, aggregate at the required grain, and filter the finished summaries. Keep row-level rules and group-level rules separate, and make threshold boundaries and fallback values explicit.

Next, you will work with window functions, which calculate results across related rows without collapsing each group to a single summary row.