Spark DataFrame processed the same partition twice, which can be reproduced with very simple code.

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("llm hard negs records") \
    .master(f"local[8]") \
    .getOrCreate()


def process_partition(index, partition):
    results = []
    s = 0

    for _ in partition:
        row = {"Result": "cool"}
        results.append(row)
        s += 1

    print(str(index) + "cool" + str(s))
    return results


data = list(range(2000))

results_rdd = spark.sparkContext.parallelize(data).repartition(8).mapPartitionsWithIndex(process_partition)
results_df = results_rdd.toDF(["Query"])

output_path = "/tmp/bc_inputs6"
results_df.write.json(output_path, mode="overwrite")