Different result in each run (pyspark)

I have a data frame as a result of multiple joins. And I want to investigate for duplicates. But each time when I investigate it the data frame looks different. In particular, the following command leads to different IDs but the number of results stays constant.

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, IntegerType, StringType
import pyspark.sql.functions as f
from pyspark.sql.functions import lit
# Create a Spark session
spark = SparkSession.builder.appName("CreateDataFrame").getOrCreate()
# User input for number of rows
n_a = 10
n_a_c = 5
n_a_c_d = 3
n_a_c_e = 4
# Define the schema for the DataFrame
schema_a = StructType([StructField("id1", StringType(), True)])
schema_a_b = StructType(
[
StructField("id1", StringType(), True),
StructField("id2", StringType(), True),
StructField("extra", StringType(), True),
]
)
schema_a_c = StructType(
[
StructField("id1", StringType(), True),
StructField("id3", StringType(), True),
]
)
schema_a_c_d = StructType(
[
StructField("id3", StringType(), True),
StructField("id4", StringType(), True),
]
)
schema_a_c_e = StructType(
[
StructField("id3", StringType(), True),
StructField("id5", StringType(), True),
]
)
# Create a list of rows with increasing integer values for "id1" and a constant value of "1" for "id2"
rows_a = [(str(i),) for i in range(1, n_a + 1)]
rows_a_integers = [str(i) for i in range(1, n_a + 1)]
rows_a_b = [(str(i), str(1), "A") for i in range(1, n_a + 1)]
def get_2d_list(ids_part_1: list, n_new_ids: int):
rows = [
[
(str(i), str(i) + "_" + str(j))
for i in ids_part_1
for j in range(1, n_new_ids + 1)
]
]
return [item for sublist in rows for item in sublist]
rows_a_c = get_2d_list(ids_part_1=rows_a_integers, n_new_ids=n_a_c)
rows_a_c_d = get_2d_list(ids_part_1=[i[1] for i in rows_a_c], n_new_ids=n_a_c_d)
rows_a_c_e = get_2d_list(ids_part_1=[i[1] for i in rows_a_c], n_new_ids=n_a_c_e)
# Create the DataFrame
df_a = spark.createDataFrame(rows_a, schema_a)
df_a_b = spark.createDataFrame(rows_a_b, schema_a_b)
df_a_c = spark.createDataFrame(rows_a_c, schema_a_c)
df_a_c_d = spark.createDataFrame(rows_a_c_d, schema_a_c_d)
df_a_c_e = spark.createDataFrame(rows_a_c_e, schema_a_c_e)
# Join everything
df_join = (
df_a.join(df_a_b, on="id1")
.join(df_a_c, on="id1")
.join(df_a_c_d, on="id3")
.join(df_a_c_e, on="id3")
)
# Nested structure
# show
df_nested = df_join.withColumn("id3", f.struct(f.col("id3")))
for i, index in enumerate([(5, 3), (4, 3), (3, None)]):
remaining_columns = list(set(df_nested.columns).difference(set([f"id{index[0]}"])))
df_nested = (
df_nested.groupby(*remaining_columns)
.agg(f.collect_list(f.col(f"id{index[0]}")).alias(f"id{index[0]}_tmp"))
.drop(f"id{index[0]}")
.withColumnRenamed(
f"id{index[0]}_tmp",
f"id{index[0]}",
)
)
if index[1]:
df_nested = df_nested.withColumn(
f"id{index[1]}",
f.struct(
f.col(f"id{index[1]}.*"),
f.col(f"id{index[0]}"),
).alias(f"id{index[1]}"),
).drop(f"id{index[0]}")
# Investigate for duplicates in id3 (should be unique)
df_test = df_nested.select("id2", "extra", f.explode(f.col("id3")["id3"]).alias("id3"))
for i in range(5):
df_test.groupby("id3").count().filter(f.col("count") > 1).show()

The last command prints in one of two my case different results. Sometimes:

+---+-----+
|id3|count|
+---+-----+
|6_4|    2|
+---+-----+

And sometimes

+---+-----+
|id3|count|
+---+-----+
|9_3|    2|
+---+-----+

If it helps I use Databricks Runtime Version 11.3 LTS (includes Apache Spark 3.3.0, Scala 2.12)

Moreover, there can be no duplicate to my understanding based on the design of the code. The found duplicate seems to be a bug!?

Maybe as a potential proof that the join does not result in any duplicates:

df_join.groupby("id3", "id4", "id5").count().filter(f.col("count") > 1).show()

is empty

The way the column "id3" is constructed is random, so for each execution you will get different results, you need to define an orderBy() to get the same results, so after adding an orderBy() on that column like below:

df_nested = df_join.withColumn("id3", f.struct(f.col("id3"))).orderBy("id3")

Now you will always get the same results, for multiple executions.

Remember, spark evaluation is lazy, so the Dag will be re-constructed for each action, in this case it's the show().

So if your code is not deterministic, it will give different outputs each time.