Definition
- transform(array, lambda) – maps each element to a new value, returning a new array of the same size.
- filter(array, lambda) – keeps only elements where the predicate is true, returning a smaller array.
- aggregate(array, start, merge, finish) – folds elements into a single value with an accumulator and optional finisher.
- exists(array, lambda) – returns true if any element matches the predicate.
- forall(array, lambda) – returns true only if every element matches.
- No row multiplication – all functions keep the array in one row, with no explode and no regroup shuffle.
Each National Bank customer row holds a list of transaction amounts. If you want to add a fee to every amount, find the big ones, total them, or check whether any is suspicious, you do not need to scatter the list into separate rows and gather them back. Higher-order functions do the work right inside the list while it stays in one row: transform adds the fee, filter keeps the big ones, aggregate totals them, exists checks for any suspicious amount, and forall checks they are all within limit.
Understanding the Concepts
Why were higher-order functions introduced?
Before Spark 2.4, the only way to transform array elements was to explode the array, apply the operation, then group back. This multiplied rows and added a shuffle.
from pyspark.sql import SparkSession, functions as F
spark = SparkSession.builder.appName("HOFDemo").getOrCreate()
data = [
(1001, "Rajesh", [50000, 30000, 10000]),
(1002, "Priya", [80000, 5000]),
(1003, "Pankaj", [200000, 150000, 75000]),
]
df = spark.createDataFrame(data, ["account_id", "holder", "amounts"])
# Old way: explode -> transform -> regroup (shuffle!)
old_way = df.withColumn("amt", F.explode("amounts")) \
.withColumn("amt_with_fee", F.col("amt") * 1.001) \
.groupBy("account_id", "holder") \
.agg(F.collect_list("amt_with_fee").alias("amounts_with_fee"))
Higher-order functions eliminate this entirely.
How does transform() work?
transform maps each element to a new value, returning a new array.
# Add a 0.1% processing fee to every transaction amount
result = df.withColumn(
"amounts_with_fee",
F.transform("amounts", lambda x: x * 1.001)
)
result.show(truncate=False)
+----------+------+--------------------+------------------------------+
|account_id|holder|amounts |amounts_with_fee |
+----------+------+--------------------+------------------------------+
|1001 |Rajesh|[50000, 30000, 10000]|[50050.0, 30030.0, 10010.0] |
|1002 |Priya |[80000, 5000] |[80080.0, 5005.0] |
|1003 |Pankaj|[200000, 150000, 75000]|[200200.0, 150150.0, 75075.0]|
+----------+------+--------------------+------------------------------+
One row per customer, no multiplication, no shuffle.
How does filter() work?
filter keeps only elements where the predicate is true.
# Keep only amounts above 50000
result = df.withColumn(
"large_amounts",
F.filter("amounts", lambda x: x > 50000)
)
result.select("account_id", "amounts", "large_amounts").show(truncate=False)
+----------+------------------------+-----------------+
|account_id|amounts |large_amounts |
+----------+------------------------+-----------------+
|1001 |[50000, 30000, 10000] |[] |
|1002 |[80000, 5000] |[80000] |
|1003 |[200000, 150000, 75000]|[200000, 150000, 75000]|
+----------+------------------------+-----------------+
How does aggregate() work?
aggregate folds the array into a single value using an accumulator.
# Sum all transaction amounts per customer
result = df.withColumn(
"total",
F.aggregate("amounts", F.lit(0).cast("long"), lambda acc, x: acc + x)
)
result.select("account_id", "amounts", "total").show(truncate=False)
+----------+------------------------+------+
|account_id|amounts |total |
+----------+------------------------+------+
|1001 |[50000, 30000, 10000] |90000 |
|1002 |[80000, 5000] |85000 |
|1003 |[200000, 150000, 75000]|425000|
+----------+------------------------+------+
The optional fourth argument is a finisher function applied to the final accumulator value.
How do exists() and forall() work?
exists returns true if any element matches. forall returns true only if every element matches.
# Flag accounts with any suspicious (>100000) transaction
result = df.withColumn(
"has_suspicious",
F.exists("amounts", lambda x: x > 100000)
).withColumn(
"all_positive",
F.forall("amounts", lambda x: x > 0)
)
result.select("account_id", "has_suspicious", "all_positive").show()
+----------+--------------+------------+
|account_id|has_suspicious|all_positive|
+----------+--------------+------------+
| 1001| false| true|
| 1002| false| true|
| 1003| true| true|
+----------+--------------+------------+
How do you use these in SQL?
In Spark SQL, higher-order functions use the arrow syntax.
SELECT account_id,
transform(amounts, x -> x * 1.001) AS amounts_with_fee,
filter(amounts, x -> x > 50000) AS large_amounts,
aggregate(amounts, 0L, (acc, x) -> acc + x) AS total,
exists(amounts, x -> x > 100000) AS has_suspicious,
forall(amounts, x -> x > 0) AS all_positive
FROM customer_transactions
How do they compose?
Higher-order functions chain freely in a single expression.
# Filter large amounts, add fee, then total
result = df.withColumn(
"large_total_with_fee",
F.aggregate(
F.transform(
F.filter("amounts", lambda x: x > 20000),
lambda x: x * 1.001
),
F.lit(0.0),
lambda acc, x: acc + x
)
)
result.select("account_id", "large_total_with_fee").show()
All of this stays columnar on the JVM with no Python boundary crossing and no row explosion.
When should you still use explode?
Use explode when you genuinely need each element as its own row – for example, to join elements against another table or to produce per-element output rows.
# Explode is correct here: join each transaction against a rules table
rules = spark.table("txn_rules") # one rule per amount range
flattened = df.withColumn("amt", F.explode("amounts"))
matched = flattened.join(rules, flattened.amt.between(rules.min_amt, rules.max_amt))
For everything else – maps, filters, checks, and simple reductions – prefer higher-order functions.
What should a senior engineer know?
- Higher-order functions run on the JVM with Catalyst optimization. They are faster than both explode-regroup and Python UDFs.
- They are available from Spark 2.4 in SQL and from Spark 3.1 with Python lambda syntax in PySpark.
- They compose freely, so complex array logic can be a single expression.
- For very complex element-wise logic that higher-order functions cannot express, fall back to
transformwith a UDF or to explode.
# Verify Spark version supports lambda syntax
print(spark.version) # 3.1+ for Python lambda support
Visual Overview
Higher-Order Functions vs Explode
──────────────────────────────────
Explode approach:
┌──────────┬───────────────┐
│ Customer │ [A, B, C] │ 1 row
└──────────┴───────────────┘
│ explode
▼
┌──────────┬───┐
│ Customer │ A │ 3 rows (multiplied)
│ Customer │ B │
│ Customer │ C │
└──────────┴───┘
│ transform + regroup (shuffle)
▼
┌──────────┬───────────────┐
│ Customer │ [A', B', C'] │ 1 row
└──────────┴───────────────┘
Higher-order function:
┌──────────┬───────────────┐
│ Customer │ [A, B, C] │ 1 row
└──────────┴───────────────┘
│ transform(lambda)
▼
┌──────────┬───────────────┐
│ Customer │ [A', B', C'] │ 1 row (no multiplication)
└──────────┴───────────────┘
Key Takeaway
Higher-order array functions, introduced in Spark 2.4, let you operate on array elements in place by passing a lambda, without ever exploding the array into rows. The five most useful are transform (maps each element), filter (keeps matching elements), aggregate (folds into one value), exists (any match?), and forall (all match?). They keep exactly one row per parent, with no row multiplication and no regrouping shuffle.
The reason to prefer them over explode for element-wise work is cost. Before these functions existed, the pattern was to explode the array, do the operation, and group back, which multiplied rows and added a shuffle. Higher-order functions apply the lambda columnar on the JVM and produce either a new array or a scalar, so Catalyst can optimize the entire expression.
At the National Bank of India this cleaned up feature engineering. For each customer's transaction list, adding a processing fee was a transform, pulling out large transactions was a filter, totalling was an aggregate, flagging suspicious amounts was an exists, and confirming all positive was a forall – all while keeping one row per customer. The guidance is to use higher-order functions for element-wise maps, checks, and simple reductions, and to reach for explode only when each element genuinely needs its own row.
Remember one sentence: higher-order array functions process elements in place with a lambda, keeping one row per parent with no explode, no multiplication, and no regroup shuffle.
---