Transforming Batch Data with PySpark, SQL and KQL

What it is

Transformation turns raw data into tables people can use. DP-700 expects the same handful of operations in three languages: PySpark in notebooks, T-SQL (or Spark SQL) in warehouses and lakehouses, and KQL in Eventhouses.

The operations are denormalizing (joining lookup tables into a wide table), grouping and aggregating, and cleaning duplicate, missing and late-arriving data.

How it works

Denormalize data

Denormalizing joins related tables so reports need fewer joins. Use a left join from the main table so rows without a match are kept.

from pyspark.sql import functions as F

sales_flat = (spark.read.table("silver_sales")
    .join(spark.read.table("silver_products"), "product_id", "left")
    .join(spark.read.table("silver_customers"), "customer_id", "left")
    .select("order_id", "order_date", "product_name", "category",
            "customer_name", "city", "amount"))

SELECT s.OrderID, s.OrderDate, p.ProductName, p.Category, c.CustomerName, s.Amount
FROM dbo.Sales AS s
LEFT JOIN dbo.Products AS p ON s.ProductID = p.ProductID
LEFT JOIN dbo.Customers AS c ON s.CustomerID = c.CustomerID;

Sales
| join kind=leftouter Products on ProductId
| project OrderId, OrderDate, ProductName, Category, Amount

In KQL, the default join kind is innerunique, which deduplicates the left side's keys, so state kind=leftouter or kind=inner explicitly. For a small dimension table, the lookup operator is a simpler, efficient alternative.

Group and aggregate data

daily = (sales_flat
    .groupBy("order_date", "category")
    .agg(F.sum("amount").alias("revenue"),
         F.countDistinct("customer_name").alias("customers")))
daily.write.mode("overwrite").format("delta").saveAsTable("gold_daily_sales")

CREATE TABLE gold.DailySales AS
SELECT OrderDate, Category, SUM(Amount) AS Revenue, COUNT(DISTINCT CustomerID) AS Customers
FROM dbo.SalesFlat
GROUP BY OrderDate, Category;

Sales
| summarize Revenue = sum(Amount), Orders = dcount(OrderId) by Category, bin(OrderDate, 1d)

CREATE TABLE AS SELECT (CTAS) in a warehouse creates and fills a table in one statement. KQL's dcount() is an estimate, fast on large data.

Handle duplicate data

from pyspark.sql.window import Window

w = Window.partitionBy("order_id").orderBy(F.col("modified_at").desc())
latest = (spark.read.table("bronze_orders")
    .withColumn("rn", F.row_number().over(w))
    .where("rn = 1")
    .drop("rn"))

dropDuplicates keeps an arbitrary row, so use the window pattern when you must keep the most recent version.

Handle missing data

clean = (latest
    .dropna(subset=["customer_id"])                      # rows unusable without the key
    .fillna({"quantity": 0, "region": "Unknown"}))       # sensible defaults

In T-SQL use COALESCE(Region, 'Unknown') or ISNULL; in KQL use coalesce(), isnull() and isempty(). Decide per column: drop rows missing a required key, default descriptive columns, and leave measures null rather than inventing values.

Handle late-arriving data

Using the event time (when it happened) rather than the load time keeps late rows in the right period.

A worked example

Order rows arrive several times as they are updated, some lack a region, and some reference products not yet in the product table.

A notebook keeps the latest row per order_id with row_number(), fills a missing region with "Unknown", and left joins products so unmatched orders are kept with a null product name. A daily job merges the result into silver and rebuilds the last three days of the gold aggregate, so late corrections are picked up.

How the exam asks about this

• "Keep only the most recent version of each order in KQL." summarize arg_max(ModifiedDate, *) by OrderId.
• "Remove exact duplicate rows in a DataFrame." dropDuplicates().
• "Create and fill an aggregated warehouse table in one statement." CREATE TABLE AS SELECT.