Learn Apache Spark - Joins, Aggregations, & Window Functions

#Introduction

Now that you've mastered DataFrames and basic SQL, episode 7 moves into the analytical operations most used in the real world: joins, aggregations, and window functions. These three are the everyday language of data analysts and data engineers.

In this episode we cover all the available join types, groupBy aggregations and their advanced variants, window functions for sequence-ordered analysis, and performance implications — because joins and shuffle are the two most common reasons queries slow down.

#Join Types in Spark

#Inner Join and the Family of Joins

Spark supports all the standard SQL join types. The clearest example uses PySpark:

Inner join of two DataFrames

pelanggan = spark.createDataFrame([(1, "budi"), (2, "sari")], ["id", "nama"])
pesanan = spark.createDataFrame([(1, "p1"), (1, "p2"), (3, "p3")], ["id", "produk"])
 
inner = pelanggan.join(pesanan, "id", "inner")
inner.show()

pelanggan.join(pesanan, "id", "inner") only returns the rows whose id matches on both sides. The available join types:

• inner: only matching rows.
• left outer: all left rows, with nulls on the right where there's no match.
• right outer: all right rows, with nulls on the left where there's no match.
• full outer: the union of all rows from both sides.
• left semi: left rows that have a match — right columns aren't included.
• left anti: left rows that have no match — useful for detecting missing data.

left semi → "data that has a match" (filter)
left anti → "data that has no match" (data quality check)

left_anti, for example, is very practical for finding orders whose customer id doesn't exist in the customer table.

#Cross Join and Caution

A cross join produces the product of all rows from both sides. Because the result can explode exponentially, Spark requires explicit confirmation before running it.

#GroupBy and Aggregations

#Basic Aggregations and Advanced Grouping

groupBy groups rows by a key and then applies an aggregation function:

Aggregation with groupBy

from pyspark.sql import functions as F
 
penjualan = spark.read.option("header", True).csv("data/penjualan.csv")
hasil = penjualan \
    .groupBy("kota", "kategori") \
    .agg(
        F.sum("jumlah").alias("total"),
        F.avg("harga").alias("harga_rata"),
        F.countDistinct("produk").alias("produk_unik"),
    )
hasil.show()

F.sum, F.avg, and F.countDistinct are built-in aggregation functions. groupBy can take multiple columns for deeper granularity — city per category analysis, for example.

#Rollup, Cube, and Grouping Sets

For multidimensional analysis, Spark provides three variants:

• rollup: hierarchical aggregation from the total down to the most detailed groups.
• cube: all key combinations, including subtotals.
• grouping sets: an explicitly chosen set of groups.

rollup(a, b) → (a,b), (a), (total)
cube(a, b)   → (a,b), (a), (b), (total)

cube produces more rows because it covers all combinations. This is useful for reports that need to view data from various angles without writing several separate queries.

#Rolling Aggregations with Windows

For aggregations that move over an ordered window (for example the average of the last 7 days), combine groupBy with the window functions in the next section.

#Window Functions

#Window: Aggregation Without Collapsing Rows

Unlike groupBy, which collapses rows, a window function computes aggregate values while keeping every row. Define the window with partitionBy and orderBy:

Row number per partition

from pyspark.sql.window import Window
 
w = Window.partitionBy("kota").orderBy(F.col("total").desc())
 
hasil = penjualan.withColumn("peringkat", F.row_number().over(w))
hasil.filter(F.col("peringkat") == 1).show()

F.row_number().over(w) assigns a sequential number within each city partition ordered by the largest total. The result: the top 1 sale per city — without losing any other columns.

#Important Window Functions

The most frequently used functions:

• row_number(): sequential number per partition.
• rank() and dense_rank(): ranking with handling for equal values.
• lag() and lead(): fetch the values of the previous and next rows.
• sum(), avg() with a window: rolling aggregations.

Change in values between periods

w2 = Window.partitionBy("kota").orderBy("bulan")
 
hasil2 = penjualan.withColumn("bulan_lalu", F.lag("total").over(w2)) \
                  .withColumn("delta", F.col("total") - F.col("bulan_lalu"))

F.lag("total").over(w2) fetches the total value from the previous month within the same city — the basic pattern for time-series analysis like growth calculations.

#Time-Based Analytics with Window Grouping

For aggregations over a time range (for example per hour), use the F.window function:

Aggregation per time window

from pyspark.sql import functions as F
 
event = spark.read.json("data/event.json")
per_jam = event \
    .groupBy(F.window("waktu", "1 hour"), "produk") \
    .count()
per_jam.show()

F.window("waktu", "1 hour") groups time into hourly windows — a pattern that will reappear in Structured Streaming in episode 10.

#Performance Implications of Joins and Shuffle

#Why Joins Are Expensive

Most joins require shuffle: data is regrouped into partitions by the join key and sent between executors over the network. The larger the data, the larger the cost. There are several strategies to reduce this cost:

• Broadcast join: if one side is small, send a copy to every executor so no shuffle is needed at all.
• Sort-merge join: the default strategy for large data, using sort and merge.
• Bucketing: prepare data with bucketing on the join key so the join becomes local.

broadcast (small side) → bucket join (prepared) → sort-merge (default)

The strategies and when to use them will be covered fully in episode 9. What you should remember now: every join and groupBy that triggers a shuffle is a prime optimization candidate.

#Conclusion

Episode 7 equips you with the three core analytical operations: joins with eight types from inner to anti, groupBy aggregations with rollup and cube, and window functions for sequential and time-based analysis. You also understand why joins and shuffle are the main sources of cost.

Key takeaways:

• Spark supports inner, outer, semi, anti, and cross joins.
• left_anti is a tool for detecting data missing from a reference table.
• rollup and cube add subtotal dimensions to aggregations.
• Window functions compute aggregations without collapsing rows.
• Every large join has the potential for a large shuffle — anticipate it from the start.

In the next episode, episode 8, we'll discuss data sources and storage — connecting Spark to HDFS, S3, JDBC, and the file system, reading and writing Parquet, Avro, ORC, JSON, and CSV, and strategies for partitioning, bucketing, file layout, and schema evolution.