Joins

The default join keeps only rows where the key matches on both sides. In Scala you pass the two DataFrames and a join condition; here we match each order's user_id to a person's id to attach the buyer's name.

def join(right: Dataset[_], joinExprs: Column): DataFrame               // defaults to inner
def join(right: Dataset[_], joinExprs: Column, joinType: String): DataFrame
// joinType: inner | left | right | full | left_semi | left_anti | cross

people.join(orders, orders("user_id") === people("id"))
  .select($"name", $"product", $"amount")
  .orderBy($"name")
  .show(false)

Outer joins preserve unmatched rows, filling the other side with nulls. left keeps every left row, right keeps every right row, and full (outer) keeps everything from both. A left join is how you keep all customers even those who never ordered.

people.join(orders, cond, "left")     // all people, nulls where no order
people.join(orders, cond, "right")    // all orders, nulls where no person
people.join(orders, cond, "full")     // everything from both sides

people.join(orders, orders("user_id") === people("id"), "left")
  .select($"name", $"product", $"amount")
  .orderBy($"name")
  .show(false)

These joins filter the left side by existence and return only left columns — no columns from the right, no row multiplication. left_semi keeps left rows that have a match; left_anti keeps left rows that don't. Think of them as IN and NOT IN against another table.

people.join(orders, cond, "left_semi")   // people who HAVE ordered (like WHERE EXISTS)
people.join(orders, cond, "left_anti")   // people who have NOT ordered (like NOT EXISTS)

// customers who have never placed an order
people.join(orders, orders("user_id") === people("id"), "left_anti")
  .select($"name", $"country")
  .show(false)

The join condition is just a Column expression, so you can combine several equalities with &&, or add range conditions — not only equality. Here we enrich orders with the buyer's name but only for high-value orders (amount ≥ 25), all in the join condition.

// multiple keys:
a.join(b, a("k1") === b("k1") && a("k2") === b("k2"))
// non-equi condition (range / inequality) — a theta join:
a.join(b, a("id") === b("fk") && b("amount") >= 25)

people.join(
  orders,
  orders("user_id") === people("id") && orders("amount") >= 25
).select($"name", $"product", $"amount")
 .orderBy($"amount".desc)
 .show(false)

When one side is small (a lookup or dimension table), wrap it in broadcast(...). Spark ships a full copy to every executor so the big side never has to shuffle across the network — often a 10× speedup. This is the single most impactful manual join optimization.

import org.apache.spark.sql.functions.broadcast
bigDf.join(broadcast(smallDf), "id")   // hint: broadcast the small side
// auto-broadcast kicks in under spark.sql.autoBroadcastJoinThreshold (default 10MB)

import org.apache.spark.sql.functions.broadcast

// people is tiny → broadcast it to avoid shuffling orders
orders.join(broadcast(people), orders("user_id") === people("id"))
  .select($"product", $"name")
  .orderBy($"product")
  .show(false)

Join two DataFrames that share a column name and the result has two columns with the same name — referencing it then throws an ambiguity error. Three ways out, in order of preference.

// 1. join on the Seq of names → Spark keeps ONE key column
a.join(b, Seq("id"))
// 2. rename before the join
b.withColumnRenamed("id", "b_id")
// 3. qualify with the source DataFrame after the join
joined.select(a("id"), b("amount"))

// BAD: both frames have `id` → ambiguous reference
// people.join(orders, people("id") === orders("id"))  // then $"id" is ambiguous

// GOOD: join on the shared name so only one `id` survives
people.join(orders.withColumnRenamed("id", "order_id"),
            orders("user_id") === people("id"))
  .select($"id", $"name", $"order_id", $"product")
  .orderBy($"order_id").show(3, false)