Overwriting What You Read
Concepts covered: Spark UI Diagnosis
A DataFrame read from a path holds the list of files found there and the recipe to apply to them. The rows stay on disk until a job opens those files, which is safe as long as the files stay put, and dangerous the day a job writes over the path its own input points at.
Spark blocks the direct case
This job lowercases every email and writes the result back where it came from:
users = spark.read.parquet("/data/users/")
fixed = users.withColumn("email", F.lower("email"))
fixed.write.mode("overwrite").parquet("/data/users/")It fails when the write is called, before any task runs, with an AnalysisException: Cannot overwrite a path that is also being read from. The check compares the target path with every path the plan reads. Nothing is deleted.
Why the guard exists
Overwrite mode empties the target first, then runs the job that produces the new files. With the guard out of the way, the steps would go like this:
| Step | What Spark does | Files in /data/users/ |
|---|---|---|
| read.parquet | records the files it found: part-0 and part-1 | 2 |
| withColumn | adds lower(email) to the recipe | 2 |
| write, before any task | deletes everything at the target path | 0 |
| write, tasks start | each task opens part-0 or part-1 to read its rows | 0 |
Calling users.count() before the write would change nothing. An action computes its answer and keeps no rows, the re-execution rule from the beginner lessons, so the write's job goes back to the files regardless.
When the guard cannot see the path
The guard reads the plan, so it catches only a read the plan still names. Convert the DataFrame to an RDD and back, a common move when the fix is a Python function over rows, and the new plan starts from an anonymous RDD with no path in it. Here clean_row is a Python function that lowercases a row's email:
users = spark.read.parquet("/data/users/")
fixed = spark.createDataFrame(users.rdd.map(clean_row), users.schema)
fixed.write.mode("overwrite").parquet("/data/users/")The RDD still reads /data/users/ lazily, but the plan the guard checks names no file, so the write goes ahead. The old files are deleted, the tasks try to open them, and each fails with an error saying a file does not exist and that the underlying files may have been updated. The job fails, and the path it set out to fix holds no data.
With spark.sql.files.ignoreMissingFiles set to true, which some teams enable to ride out files vanishing during a read, the outcome is worse: the tasks skip the deleted files instead of failing, the job reports success, and the table now has 0 rows.
Write somewhere new, then switch
The fix leaves the input untouched until the output is complete. Write the result to a new path, check it, and move readers over only after the check passes:
fixed.write.mode("overwrite").parquet("/data/users_2024_03_15/")
check = spark.read.parquet("/data/users_2024_03_15/")
assert check.count() == spark.read.parquet("/data/users/").count()Readers find the current version through 1 indirection, such as a table definition or a config value holding the path, and you update that pointer after the check. The old path stays readable until you delete it, so a bad run rolls back by pointing the indirection at yesterday's directory.
- Do
- Write each run to a new path and move readers over once the output is checked
- Treat a job whose input and output paths match as a bug, even when Spark lets it run
- Keep ignoreMissingFiles off in jobs that overwrite, so a missing input fails the run
- Avoid
- Count the input before the write and assume its rows are now held in memory
- Round-trip through .rdd and then overwrite the path you read from
- Overwrite in place to save storage; a single failed run costs you the whole table
clean_row lowercases the email field. What happens when the last line runs?
users = spark.read.parquet("/data/users/")
fixed = spark.createDataFrame(users.rdd.map(clean_row), users.schema)
print(users.count())
fixed.write.mode("overwrite").parquet("/data/users/")- The guard finds no path in the plan, overwrite deletes the old files, and the tasks fail on files that are gone
- Spark raises an AnalysisException at the write, because fixed still reads /data/users/ underneath
- The write succeeds, because users.count() already loaded every row of the table into executor memory
- The write succeeds, but the new files hold only the rows the earlier count() happened to read
About this lesson
This section is part of the Lazy Evaluation: Intermediate lesson on DataDriven, the home of data engineering. Lessons cover SQL, Python, Spark, data modeling and pipeline architecture, and each section pairs its explanation with worked examples and practice you run as you go.