Non-deterministic functions in Spark

• Non-deterministic function – a function that can return different results across invocations even for the same input, because it depends on randomness, state, partitioning, or ordering.
• Examples – rand, randn, uuid, monotonically_increasing_id, shuffle, input_file_name, spark_partition_id, and order-sensitive collect_list.
• Optimizer impact – Catalyst marks these non-deterministic and will not push them below filters, will not collapse duplicate invocations, and must handle re-execution carefully.
• Re-execution risk – Spark can recompute a lost partition or recompute an uncached DataFrame referenced twice, producing different values each time.
• Fix – cache or materialize to freeze non-deterministic values, seed rand for repeatability, and sort collect_list for deterministic order.

What is a non-deterministic function in Spark?

A deterministic function returns the same output for the same input every time, which lets Catalyst move, combine, duplicate, and reorder it safely. A non-deterministic function breaks that assumption. Each invocation may produce a different result, so the optimizer must treat it with care.

from pyspark.sql import SparkSession
from pyspark.sql.functions import rand, col, monotonically_increasing_id

spark = SparkSession.builder.appName("NonDetDemo").getOrCreate()

# rand() returns different values every time it is evaluated
df = spark.range(5).withColumn("random_val", rand())
df.show()
# Each run produces different random_val values

+---+-------------------+
| id|         random_val|
+---+-------------------+
|  0|0.41371264720975787|
|  1|0.73117192818966221|
|  2|0.12324568901234567|
|  3|0.98765432109876543|
|  4|0.55544332211009988|
+---+-------------------+

Which functions are non-deterministic?

Spark explicitly tracks a deterministic flag on every expression. The following built-in functions are non-deterministic:

| Function | Why Non-Deterministic |
|---|---|
| rand(), randn() | Draw from a random number generator |
| uuid() | Generates a fresh unique identifier each call |
| monotonically_increasing_id() | Values depend on partition layout, not stable across runs |
| shuffle(array) | Randomly reorders array elements |
| input_file_name() | Returns the physical file path, depends on data location |
| spark_partition_id() | Returns the partition number, depends on partitioning |
| collect_list() | Element order depends on execution and partitioning |

from pyspark.sql.functions import uuid, spark_partition_id, input_file_name

# Each of these changes between runs
df_ids = spark.range(3) \
    .withColumn("uid", uuid()) \
    .withColumn("part_id", spark_partition_id()) \
    .withColumn("mono_id", monotonically_increasing_id())
df_ids.show(truncate=False)

How does Catalyst treat non-deterministic functions differently?

Catalyst applies three restrictions to non-deterministic expressions:

• No predicate pushdown – Catalyst will not push a non-deterministic expression below a filter, because a value computed before versus after the filter could differ. This can cause more data to be read than expected.
• No common-subexpression elimination – two separate rand() calls are not collapsed into one, because each should produce an independent random value.
• No safe re-execution assumption – Spark cannot assume a re-run yields the same values, which matters when it recomputes a lost partition or recomputes an uncached DataFrame.

# Predicate pushdown is skipped around non-deterministic functions
df_with_rand = spark.read.parquet("/data/transactions") \
    .withColumn("sample_flag", rand() < 0.01)

# The filter on sample_flag cannot be pushed to the Parquet reader
sampled = df_with_rand.filter(col("sample_flag"))
sampled.explain(True)
# The physical plan shows a Filter AFTER the full scan, not pushed down

What happens when you reference a non-deterministic DataFrame twice?

If you create a DataFrame with a non-deterministic column and reference it in two places without caching, Spark recomputes the column independently for each reference, potentially producing different values.

# BUG: two references to the same uncached DataFrame disagree
df = spark.read.parquet("/data/trades") \
    .withColumn("surrogate_id", monotonically_increasing_id())

# These two actions may see different surrogate_id values!
df.select("surrogate_id", "trade_id").write.parquet("/output/mapping")
df.select("surrogate_id", "amount").write.parquet("/output/amounts")
# Joining mapping and amounts on surrogate_id can match wrong rows

# FIX: cache to freeze the values
df.cache()
df.select("surrogate_id", "trade_id").write.parquet("/output/mapping")
df.select("surrogate_id", "amount").write.parquet("/output/amounts")
df.unpersist()

How do you make non-deterministic results stable?

There are several strategies to freeze or stabilize non-deterministic values:

from pyspark.sql.functions import sha2, concat_ws

# Strategy 1: Cache the DataFrame
df_cached = df.withColumn("rid", rand(seed=42)).cache()
df_cached.count()  # Materialize the cache

# Strategy 2: Write-then-read to persist to storage
df.withColumn("rid", rand(seed=42)).write.parquet("/tmp/frozen")
df_frozen = spark.read.parquet("/tmp/frozen")

# Strategy 3: Use a deterministic hash instead of monotonically_increasing_id
df_stable = spark.read.parquet("/data/trades") \
    .withColumn("stable_id",
        sha2(concat_ws("|", col("trade_id"), col("account_id"),
             col("trade_date")), 256))

How do you handle collect_list ordering?

The collect_list function is non-deterministic because the order of collected elements depends on execution and partitioning. When order matters, impose it explicitly.

from pyspark.sql.functions import collect_list, sort_array
from pyspark.sql.window import Window

# WRONG: collect_list order is not guaranteed
txn_list = spark.read.parquet("/data/transactions") \
    .groupBy("account_id") \
    .agg(collect_list("amount").alias("amounts"))
# The amounts array may be in different order on each run

# CORRECT: sort the collected array
txn_list_sorted = spark.read.parquet("/data/transactions") \
    .groupBy("account_id") \
    .agg(sort_array(collect_list("amount")).alias("amounts"))

# ALTERNATIVE: use a window function to control order before collecting
w = Window.partitionBy("account_id").orderBy("txn_timestamp")
ordered_txns = spark.read.parquet("/data/transactions") \
    .withColumn("row_num", monotonically_increasing_id())

What are common mistakes with non-deterministic functions?

The most frequent mistakes involve treating non-deterministic results as stable identifiers.

# WRONG: using monotonically_increasing_id as a join key across DataFrames
df1 = spark.read.parquet("/data/trades") \
    .withColumn("id", monotonically_increasing_id())
df2 = spark.read.parquet("/data/trades") \
    .withColumn("id", monotonically_increasing_id())
joined = df1.join(df2, "id")  # May match wrong rows!

# CORRECT: use a deterministic key from the data itself
df1 = spark.read.parquet("/data/trades")
df2 = spark.read.parquet("/data/trades")
joined = df1.join(df2, "trade_id")  # Stable, deterministic key

What should a senior engineer know?

• Seeded rand – rand(seed=42) is repeatable within a single run but still non-deterministic across stage retries after failures.
• Stage retries – if a stage fails and is retried, non-deterministic columns in the retried tasks may produce different values than the original tasks, causing data inconsistency.
• Performance impact – non-deterministic expressions can prevent predicate pushdown, which means Spark reads more data from the source than necessary.
• Tungsten codegen – non-deterministic expressions are evaluated per-row and cannot be hoisted out of loops during whole-stage code generation.
• Production rule – at the National Bank of India, we banned monotonically_increasing_id as a join key and required deterministic hashes of stable columns for surrogate keys.