Spark writes parquet with partitionBy throws FileAlreadyExistsException in its own temporary working space

It seems to be a simple task

What I am trying to do is to read from a csv file and turn it into a hive partitioned parquet dataset.

master = "local[*]"
app_name = "convert_to_parquet"
spark = (
    SparkSession.builder
    .appName(app_name)
    .master(master)
    .getOrCreate()
)

csv_path = "<csv-path>"
in_df = spark.read.option("inferSchema", "true").option("header", "true").csv(csv_path)
out_df = in_df.selectExpr("trim(OBJECTID) AS ID",
                          "trim(NAME) AS NAME",
                          "trim(CITY) AS CITY",
                          "trim(STATE) AS STATE", "X", "Y")
out_path = "<out-dir>"
# make sure output directory doesn't exist before writing
shutil.rmtree(out_path, ignore_errors=True)

out_df.write.partitionBy(["CITY"]).parquet(out_path)

I got error:

23/02/24 01:54:22 ERROR Utils: Aborting task                        (0 + 1) / 1]
org.apache.hadoop.fs.FileAlreadyExistsException: File already exists: file: <out-dir>/_temporary/0/_temporary/attempt_202302240154215533456461598462887_0002_m_000000_2/CITY=Apex/part-00000-0de20945-1012-4a7a-b183-c4235717a0a2.c000.snappy.parquet
        at org.apache.hadoop.fs.RawLocalFileSystem.create(RawLocalFileSystem.java:421)
        at org.apache.hadoop.fs.RawLocalFileSystem.create(RawLocalFileSystem.java:459)
        ...

The baffling part is that FileAlreadyExistsException is caused by Spark's temporary working space.

I am using the local mode of spark and the local filesystem (not HDFS), if this information matters.

What can go wrong here?

Alright, after several rounds of excruciating trial & error, I find that this issue is caused by letter case!

The dataset has city names of different case versions coexisted under the column City, like "APEX" and "Apex" or "MORRISVILLE" and "Morrisville".

Inside the spark engine, which is case sensitive, they are treated as different partitions. But probably because the parquet file writer is case insensitive towards the output path (derived out of the value from columns used for partition), what regarded as different partitions are somehow given the same output path, hence the issue.

One exactly same question was asked 6 years ago, yet issue of this level of detail can still hardly be covered by documentation.

https://stackoverflow.com/questions/38597401/spark-case-sensitive-partitionby-column

Solution:

out_df = in_df.selectExpr("trim(OBJECTID) AS ID",
                          "trim(NAME) AS NAME",
                          "upper(trim(CITY)) AS CITY", # unify cases
                          "trim(POSTALCODE) AS POSTALCODE",
                          "trim(STATE) AS STATE", "X", "Y")