Spark fails reading a Parquet folder: column type mismatch, incompatible schemas and unreadable files

#Symptom

A job that reads a Parquet landing folder or a Hive-style external table fails during the scan, not in your transformation code. Typical messages (names vary by Spark version; check your build's output):

• Spark 3.x: Parquet column cannot be converted in file .../part-00012.parquet. Column: [amount], Expected: decimal(18,2), Found: DOUBLE, often wrapping SchemaColumnConvertNotSupportedException
• Newer releases: [FAILED_READ_FILE.PARQUET_COLUMN_DATA_TYPE_MISMATCH] ... Expected Spark type ..., actual Parquet type ...
• With schema merging on: [CANNOT_MERGE_INCOMPATIBLE_DATA_TYPE] Failed to merge incompatible data types ...
• Decimal problems: [PARQUET_CONVERSION_FAILURE.DECIMAL] or .WITHOUT_DECIMAL_METADATA
• Empty or missing folder: [UNABLE_TO_INFER_SCHEMA] Unable to infer schema for Parquet. It must be specified manually.
• Truncated or non-Parquet file: ... is not a Parquet file, or FAILED_READ_FILE.CANNOT_READ_FILE_FOOTER
• Old dates or timestamps: INCONSISTENT_BEHAVIOR_CROSS_VERSION / SparkUpgradeException about rebasing ancient dates

The cause is almost always the same. Parquet stores a schema in every file, and a folder has no schema of its own. When one producer run writes amount as DOUBLE and the next writes DECIMAL(18,2), the folder now holds two schemas. Spark finds out only when it opens the file that doesn't match.

Note

If the table is Delta, don't read its folder with spark.read.parquet. Delta keeps the schema in its transaction log and handles this differently. See the Delta schema mismatch post instead.

#Why it worked yesterday

With mergeSchema off (the default), Spark infers the schema from the footer of a single file and assumes every other file matches. That gives you three confusing behaviours:

1. The job succeeds or fails depending on which file Spark samples. After new files land, the sampled file can change.
2. Partition pruning hides the problem. A query on dt=2026-11-01 works, and one on dt=2026-11-02 fails.
3. A Hive or catalog table definition fixes the expected type. Every file that disagrees with it fails, however many files there are.

#Step 1: find the files that disagree

Don't guess. List every file's schema and group by schema. With PyArrow this reads only footers, so it's quick even for thousands of files:

import pyarrow.parquet as pq
import pyarrow.fs as pafs
from collections import defaultdict

# For ADLS/S3 use the matching pyarrow filesystem or fsspec; local/mounted path shown here
fs = pafs.LocalFileSystem()
root = "/mnt/landing/sales/orders"

files = [f.path for f in fs.get_file_info(pafs.FileSelector(root, recursive=True))
         if f.path.endswith(".parquet")]

groups = defaultdict(list)
bad = []
for path in files:
    try:
        schema = pq.read_schema(path, filesystem=fs)
        key = tuple((f.name, str(f.type)) for f in schema)
        groups[key].append(path)
    except Exception as e:          # truncated / not Parquet
        bad.append((path, str(e)[:120]))

for i, (key, paths) in enumerate(groups.items()):
    print(f"schema #{i}: {len(paths)} files, e.g. {paths[0]}")
    print("   ", key)
print("unreadable:", len(bad), bad[:5])

On a cluster without direct filesystem access, use Spark for the same check. It's slower, so run it on one partition first:

paths = [r.path for r in (spark.read.format("binaryFile")
         .option("recursiveFileLookup", "true")
         .option("pathGlobFilter", "*.parquet")
         .load("abfss://landing@acct.dfs.core.windows.net/sales/orders/dt=2026-11-02")
         .select("path").collect())]

for p in paths:
    try:
        print(p, spark.read.parquet(p).schema.simpleString())
    except Exception as e:
        print(p, "UNREADABLE", str(e)[:120])

Compare the schema groups with the file modification times. The changeover point usually lines up with a deploy of the producing job or a source system change.

#Step 2: match the error to the fix

#What mergeSchema does and doesn't do

df = spark.read.option("mergeSchema", "true").parquet(root)

Schema merging combines the column sets from every file, so it fixes added columns. It does not reconcile incompatible types, so you'll get CANNOT_MERGE_INCOMPATIBLE_DATA_TYPE. It also opens every footer on each read, which is slow on large folders. Use it for a known additive change, not as a permanent setting.

Caution

spark.sql.files.ignoreCorruptFiles=true and the ignoreCorruptFiles read option make Spark skip files it can't read. The job goes green and those rows are silently missing. Use it only for a one-off read where you have already listed and quarantined the bad files, and never leave it on in a production job.

#Step 3: read each schema group and rewrite clean

Don't overwrite the source folder in place. Write a clean copy, check it, then switch readers over.

from functools import reduce
from pyspark.sql import functions as F

target_types = {"order_id": "bigint", "amount": "decimal(18,2)", "order_ts": "timestamp"}

def normalise(df):
    for col, typ in target_types.items():
        df = df.withColumn(col, F.col(col).cast(typ)) if col in df.columns \
             else df.withColumn(col, F.lit(None).cast(typ))
    return df.select(*target_types.keys(), *[c for c in df.columns if c not in target_types])

# groups: {schema_key: [paths]} from Step 1 (use full abfss:// or s3:// URIs)
parts = [normalise(spark.read.parquet(*paths)) for paths in groups.values()]
clean = reduce(lambda a, b: a.unionByName(b, allowMissingColumns=True), parts)

(clean.write.mode("overwrite")
      .partitionBy("dt")
      .parquet("abfss://curated@acct.dfs.core.windows.net/sales/orders_v2"))

Each spark.read.parquet(*paths) gets files with a single schema, so the scan can't hit a mismatch. The cast happens after the read, where you control it.

Check before cutting over:

src_rows = sum(spark.read.parquet(*p).count() for p in groups.values())
dst = spark.read.parquet("abfss://curated@acct.dfs.core.windows.net/sales/orders_v2")
print(src_rows, dst.count())
dst.where(F.col("amount").isNull()).count()   # casts that failed become null

Warning

cast returns null for values it can't convert, such as non-numeric strings to int, and it doesn't raise an error (unless ANSI mode is on). Compare null counts per column before and after. A jump means data was lost in the cast, not just reformatted.

For a Hive external table, point the table at the new location (or swap folders) once the counts match:

ALTER TABLE sales.orders SET LOCATION 'abfss://curated@acct.dfs.core.windows.net/sales/orders_v2';
MSCK REPAIR TABLE sales.orders;

#Stop the bleeding while you fix it

If the morning load has to run before the cleanup is finished, read with an explicit schema and limit the scan to partitions you've confirmed are consistent:

schema = "order_id bigint, amount decimal(18,2), order_ts timestamp, dt string"
df = spark.read.schema(schema).parquet(f"{root}/dt=2026-11-03")

An explicit schema removes the sampling lottery, but files whose physical type disagrees will still fail. That's what you want, because they fail loudly rather than load wrong data.

#Prevent it next time

• Land raw data into a table format (Delta or Iceberg) instead of loose Parquet. The table schema is enforced on write, so a bad producer fails at write time, not three jobs later.
• Have producers write an explicit schema, not one inferred from the source, and agree on types in a data contract (especially decimal precision and int vs bigint).
• Add a footer check (the Step 1 script) as a gate in the ingestion job. Fail the run and quarantine the files when a new schema group appears.
• Keep a quarantine folder and an alert. A rejected file should open a ticket for the producer, not wait for an engineer to find it.
• Don't leave mergeSchema or ignoreCorruptFiles switched on in production jobs.