Spark превратил 00123 в 123: явная схема CSV для идентификаторов

После загрузки CSV код клиента 00123 стал числом 123. Позже появился клиент A0123, и новая поставка начала определять колонку как строку. В результате одинаковый pipeline получал разные типы в зависимости от содержимого файла.

inferSchema выводит тип из наблюдаемых значений. Для числовой величины это бывает удобно, но идентификатор не становится числом только потому, что сегодня состоит из цифр. Ведущие нули могут быть частью ключа. Поэтому схему стабильного источника лучше задавать явно.

Опишем тип до чтения

Пример создаёт маленький временный CSV, читает его PySpark с заданной схемой и записывает обратно. После повторного чтения оба кода сохранятся. Для локального запуска нужны совместимые установленные PySpark и Java.

from pathlib import Path
from tempfile import TemporaryDirectory
from pyspark.sql import SparkSession, types as T

schema = T.StructType([
    T.StructField("customer_id", T.StringType(), False),
    T.StructField("qty", T.IntegerType(), False),
])
spark = SparkSession.builder.master("local[1]").appName("csv-contract").getOrCreate()
try:
    with TemporaryDirectory() as folder:
        source = Path(folder) / "input.csv"
        source.write_text("customer_id,qty\n00123,2\nA0123,3\n", encoding="utf-8")
        data = (spark.read.schema(schema).option("header", True)
                .option("mode", "FAILFAST").option("enforceSchema", False)
                .csv(str(source)))
        values = [(r.customer_id, r.qty) for r in data.orderBy("customer_id").collect()]
        assert values == [("00123", 2), ("A0123", 3)]
        target = str(Path(folder) / "output")
        data.write.mode("errorifexists").option("header", True).csv(target)
        restored = spark.read.schema(schema).option("header", True).csv(target)
        assert sorted(tuple(r) for r in restored.collect()) == values
        print(values)
finally:
    spark.stop()

Строковый тип назначен на входе, поэтому нули не исчезают на первом преобразовании. Превращение уже прочитанного числа в строку дало бы только «123»: исходная длина потеряна, и восстановить её без внешнего договора нельзя.

Режим FAILFAST останавливает чтение при повреждённой записи, а enforceSchema=false включает проверку соответствия заголовка схеме с учётом настроек чувствительности к регистру. Это полезно против незаметного изменения порядка или названий полей. Но такие параметры не заменяют проверку бизнес-значений.

nullable не является полной валидацией источника

Внешний формат может содержать пустые значения. Даже если поле схемы помечено обязательным, после чтения проверяйте customer_id.isNull(), допустимый формат и уникальность ключа. Для qty отдельно задайте положительный диапазон. Схема описывает представление, а качество поставки требует собственных условий.

Не применяйте универсальный lpad ко всем числовым кодам. Если источник использует переменную длину или буквы, дополнение нулями создаст новые значения. Нормализация должна следовать известному формату, а не исправлять последствия угадывания типа.

Проверьте следующий потребитель

При записи в Parquet тип сохранится в метаданных. В CSV типов нет, поэтому следующая система может снова угадать число. Передавайте схему рядом с файлом или используйте формат с типами. Если файл открывают в Excel, импортируйте столбец как текст, иначе проблема вернётся уже у аналитика.

В контрольный набор добавьте 00000, буквенный код, пустое значение и два разных кода, совпадающих после числового преобразования. Сверяйте сами ключи после полного прохода чтение–запись–чтение. Совпавшее количество строк не обнаружит повреждённые идентификаторы.

В Data Engineering Fast-Track вы отработаете схемы, контракты, Spark и хранение в Parquet/lakehouse. Посмотрите программу и соберите pipeline, в котором ключи сохраняют смысл от исходного файла до витрины.