How can you read data from multiple sources (e.g., CSV, JSON, Parquet) and combine them into a single DataFrame, handling potential schema differences?

Read each source with its own reader and an explicit schema, map every source onto one target schema (same names, same types, a column for the source), then combine them with unionByName, using allowMissingColumns=True for columns that exist in only some sources.

from pyspark.sql import functions as F

csv_df = (spark.read.option("header", True)
            .schema("id INT, name STRING, amount DOUBLE")
            .csv("/landing/orders_csv/"))

json_df = (spark.read
            .schema("id LONG, customer_name STRING, amount DOUBLE, channel STRING")
            .json("/landing/orders_json/"))

parquet_df = spark.read.parquet("/landing/orders_parquet/")   # schema in the files

def to_target(df, source, renames=None):
    for old, new in (renames or {}).items():
        df = df.withColumnRenamed(old, new)
    return (df.withColumn("id", F.col("id").cast("long"))
              .withColumn("amount", F.col("amount").cast("decimal(12,2)"))
              .withColumn("source", F.lit(source)))

combined = (to_target(csv_df, "csv")
    .unionByName(to_target(json_df, "json", {"customer_name": "name"}), allowMissingColumns=True)
    .unionByName(to_target(parquet_df, "parquet"), allowMissingColumns=True))

Handling the schema differences

• Different names: rename to the target names (withColumnRenamed, or withColumnsRenamed with a dictionary) before the union.
• Missing columns: unionByName(..., allowMissingColumns=True) fills them with null.
• Different types: cast explicitly to the target type. unionByName will widen compatible types (int and long become long) but fails or gives surprising results for incompatible ones. Remember that ANSI mode is on by default since Spark 4, so a cast that can't succeed raises an error; use try_cast where bad values are expected.
• Column order: unionByName matches by name; plain union matches by position and silently mixes up columns.
• Nested fields: align structs field by field, or flatten them first.

Why explicit schemas Schema inference reads extra data, may guess differently from day to day (for example, a zip code 01234 inferred as the number 1234), and differs between formats. Explicit schemas make the union predictable.

Bad records With the default PERMISSIVE mode, a malformed CSV or JSON line becomes a row of nulls (a CSV line with too few fields counts as malformed). Add a _corrupt_record column to the schema to keep the raw text, and filter those rows into a quarantine table before the union.

Parquet files with different schemas Within one Parquet source, option("mergeSchema", "true") combines compatible schemas across files.

In production Land each source as is (with its own schema), then build the combined table in a separate step, ideally as a Delta or Iceberg table with schema enforcement, so a new column or changed type in one source is caught instead of silently changing the combined data. Auto Loader and declarative pipelines on Databricks add schema evolution and rescued-data columns for this.