How to validate and align schemas in ELT: step by step

João Barros 02 de September de 2026 5 min read

Validating and aligning schemas in ELT is essential to avoid load failures, quality issues and regressions when sources change. This guide shows how to detect schema differences, apply coercion/transformation and record changes so ELT loads are robust.

Prerequisites

Basic SQL knowledge and familiarity with a processing engine (e.g.: Spark, Synapse, BigQuery).
Access to an ELT environment with the ability to execute transformations (e.g.: Spark cluster or SQL warehouse).
Sample JSON/CSV files or source tables and a Delta/SQL target table.

Step 1: Map expected and actual schemas

Before loading, compare the expected schema (target table) with the actual source schema. This allows you to identify missing columns, incompatible types and extra columns.

-- Exemplo SQL para inspecionar esquema destino e fonte (Synapse/SQL genérico)
SELECT column_name, data_type
FROM information_schema.columns
WHERE table_name = 'destino_table'
ORDER BY ordinal_position;

-- Fonte: ler um ficheiro JSON com Spark para inferir esquema
val df = spark.read.option("multiline", true).json("/data/source/sample.json")
df.printSchema()

Step 2: Detect differences and coercion rules

Automate the comparison: create logic that detects missing columns, different types or nullable mismatches. Define coercion rules (e.g.: string->timestamp, int->long) and how to handle invalid values (null, default, error).

# Pseudocódigo em PySpark para comparar esquemas
from pyspark.sql.types import StructType

def schema_to_dict(schema: StructType):
    return {f.name: f.dataType.simpleString() for f in schema.fields}

src_schema = schema_to_dict(spark.read.json('/data/source/sample.json').schema)
dst_schema = schema_to_dict(spark.read.table('destino_table').schema)

# encontrar diferenças
missing = set(dst_schema) - set(src_schema)
diff_types = {col: (src_schema.get(col), dst_schema[col]) for col in src_schema if col in dst_schema and src_schema[col] != dst_schema[col]}
print('missing', missing)
print('diff_types', diff_types)

Step 3: Normalize and convert types in ELT

Transform the source to the target schema by applying casts, parsing and default values. Use robust functions to avoid failures (e.g.: try_cast, to_timestamp with fallback).

# Exemplo PySpark: aplicar casts e colunas por omissão
from pyspark.sql.functions import col, lit, to_timestamp

df = spark.read.json('/data/source/sample.json')

# garantir coluna 'user_id' como long e 'created_at' como timestamp
df2 = df.withColumn('user_id', col('user_id').cast('long')) \
         .withColumn('created_at', to_timestamp(col('created_at'), "yyyy-MM-dd'T'HH:mm:ss").alias('created_at'))

# adicionar colunas em falta com valor por omissão
for c in ['status', 'country']:
    if c not in df2.columns:
        df2 = df2.withColumn(c, lit(None).cast('string'))

df2.printSchema()

Step 4: Record divergences and create a compatibility report

For maintenance, record all divergences found with examples and counts. This helps track source changes and decide whether you need to change the target or apply additional transformations.

# Exemplo: contar valores que falharam no cast
bad_userid = df.filter(col('user_id').cast('long').isNull() & col('user_id').isNotNull()).count()
print(f'valores user_id com cast inválido: {bad_userid}')

# Escrever relatório simples numa tabela de auditoria
report = spark.createDataFrame([('missing_cols', ','.join(missing)), ('bad_userid', str(bad_userid))], ['issue', 'detail'])
report.write.mode('append').saveAsTable('audit_schema_issues')

Step 5: Apply idempotent loading after alignment

After normalizing and validating, load into the target table using an idempotent operation (e.g.: MERGE or controlled append) to avoid duplication and inconsistencies.

-- Exemplo MERGE em SQL (Delta Lake)
MERGE INTO destino_table AS d
USING staging_table AS s
ON d.id = s.id
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *;

-- Alternativa: overwrite particionado apenas para partição específica
INSERT OVERWRITE destino_table PARTITION(dt='2026-09-01') SELECT * FROM staging_table WHERE dt='2026-09-01';

Verify the result

Confirm validity by checking the final schema, counts and samples. Run queries to compare counts per partition and column-by-column validations (types, nulls, timestamp formats).

-- Verificar esquema final
SELECT column_name, data_type, is_nullable FROM information_schema.columns WHERE table_name='destino_table';

-- Contagens por partição e amostra
SELECT dt, COUNT(*) FROM destino_table GROUP BY dt ORDER BY dt DESC LIMIT 10;

-- Amostra de linhas com valores inválidos
SELECT * FROM destino_table WHERE TRY_CAST(user_id AS BIGINT) IS NULL AND user_id IS NOT NULL LIMIT 10;

Conclusion

Validating and aligning schemas in ELT reduces load failures and makes pipeline evolution easier. Next steps: automate these checks with CI/CD tests and alerts; consider maintaining a data dictionary and a schema versioning process. Tip: start by recording small divergence reports — they are gold for diagnosing source changes.