Как собрать минимальный ETL-конвейер на Apache Spark DataFrame API

Минимальный ETL в Spark DataFrame API — это короткий, но полный цикл: прочитать данные, преобразовать их и сохранить результат в формате, удобном для следующего этапа обработки. Для учебного конвейера обычно достаточно CSV или Parquet на входе, нескольких операций над DataFrame и записи результата обратно в Parquet.

Ниже — учебный пример на PySpark, собранный из документированных возможностей Spark: создание SparkSession, чтение CSV и Parquet, работа со схемой, обработка плохих строк, простые преобразования, подготовка к записи и повторное чтение сохранённого набора для проверки.

С чего начинается ETL и зачем нужен SparkSession

SparkSession — это точка входа в Spark SQL и DataFrame API. Через неё создают читатели данных, управляют настройками и запускают операции над DataFrame.

from pyspark.sql import SparkSession

spark = (
    SparkSession.builder
    .appName("minimal-etl")
    .getOrCreate()
)

В показанном конвейере переменная spark хранит SparkSession: через spark.read читаются файлы. В интерактивной оболочке PySpark такая сессия может быть создана автоматически; в самостоятельном скрипте её получают через builder.getOrCreate(). Это описание данного способа работы, а не утверждение, что у Spark нет других программных интерфейсов.

Чтение CSV и Parquet в DataFrame

CSV в Spark можно читать как из одного файла, так и из каталога файлов. Parquet тоже читается путём или каталогом. Разница в том, что CSV — текстовый формат, а Parquet — самодокументируемый колонночный формат, поэтому схема при чтении Parquet обычно уже известна и сохраняется автоматически.

Быстрый старт: schema inference

Если нужно быстро посмотреть содержимое CSV, можно включить inferSchema. Но документация Spark подчёркивает: это требует дополнительного прохода по данным. Для разового исследования это нормально, для стабильного ETL — не всегда выгодно.

quick_df = (
    spark.read
    .option("header", True)
    .option("inferSchema", True)
    .csv("/data/in/orders_csv")
)

Такой вариант удобен, когда структура понятна не до конца. Но у него есть минус: итоговые типы выводятся по содержимому файлов, а не по вашему контракту данных.

Надёжный вариант: явная схема

Для ETL обычно лучше задать схему явно. Тогда вы заранее фиксируете типы и делаете поведение конвейера более предсказуемым.

from pyspark.sql.types import (
    StructType, StructField, StringType, DoubleType
)

csv_schema = StructType([
    StructField("order_id", StringType(), True),
    StructField("event_date", StringType(), True),
    StructField("country", StringType(), True),
    StructField("amount", DoubleType(), True),
    StructField("discount", DoubleType(), True),
    StructField("_corrupt_record", StringType(), True)
])

raw_csv = (
    spark.read
    .option("header", True)
    .option("enforceSchema", False)
    .option("mode", "PERMISSIVE")
    .option("columnNameOfCorruptRecord", "_corrupt_record")
    .schema(csv_schema)
    .csv("/data/in/orders_csv")
)

Пара важных деталей:

• header=True говорит использовать первую строку как названия колонок.
• enforceSchema=False полезен, если вы хотите, чтобы Spark сверял схему с заголовками CSV. В документации отдельно отмечено, что при значении true схема принудительно применяется, а заголовки могут быть проигнорированы; для избежания неверных результатов рекомендуется отключать такое поведение.
• Если хотите сохранить исходную плохую строку, в пользовательской схеме должна быть строковая колонка с тем же именем, что и в columnNameOfCorruptRecord.

Parquet читается проще:

parquet_df = spark.read.parquet("/data/in/reference_parquet")

При повторном чтении Parquet Spark сохраняет схему набора. Документация также отмечает, что при чтении все колонки становятся nullable для совместимости.

Что делать с плохими входными данными

Для CSV у Spark есть документированные режимы обработки некорректных строк.

После чтения в режиме PERMISSIVE можно отделить хорошие строки от плохих:

bad_rows = raw_csv.filter("_corrupt_record is not null")
valid_rows = raw_csv.filter("_corrupt_record is null")

Здесь есть тонкость из документации Spark: CSV-парсер может анализировать только нужные колонки, и из-за этого одна и та же запись иногда выглядит «испорченной» или «нормальной» в зависимости от того, какие поля вы запросили. Даже простое count() может не выявить проблему, потому что для него не нужны колонки. Поэтому плохие строки лучше проверять на DataFrame, где действительно выбраны интересующие поля.

Ещё один важный случай — строки с переводом строки внутри кавычек. Если multiLine=False (это значение по умолчанию), перенос может разрезать одну логическую запись на две физические строки, и дальше режим mode будет работать уже с этим разбиением.

Повреждённые и лишние файлы

Для файловых источников Spark поддерживает ignoreCorruptFiles. Это относится уже не к плохим строкам внутри CSV, а к проблемным файлам как объектам чтения.

safe_parquet = (
    spark.read
    .option("ignoreCorruptFiles", "true")
    .parquet("/data/in/reference_parquet")
)

Такой режим позволяет продолжить чтение, если в каталоге оказался повреждённый файл. Но у него есть цена: часть данных будет пропущена. Поэтому после такой загрузки стоит отдельно проверить полноту набора.

Для CSV есть ещё одна документированная ловушка: если читать целую папку, нужно следить, чтобы там были только CSV. Лишние файлы могут привести к «неправильной» схеме и странным значениям, потому что Spark попытается прочитать всё содержимое каталога как CSV.

Базовые преобразования: select, filter, withColumn и null

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

from pyspark.sql.functions import col, lit, when

result = (
    valid_rows
    .filter(col("order_id").isNotNull())
    .filter(col("amount").isNotNull())
    .withColumn(
        "country",
        when(col("country").isNull(), lit("unknown")).otherwise(col("country"))
    )
    .withColumn(
        "discount",
        when(col("discount").isNull(), lit(0.0)).otherwise(col("discount"))
    )
    .withColumn("net_amount", col("amount") - col("discount"))
    .select("event_date", "country", "order_id", "net_amount")
)

Здесь используются три базовые идеи:

• select оставляет только поля, нужные следующему шагу.
• filter убирает строки, которые нельзя корректно обработать.
• withColumn пересчитывает существующую колонку или создаёт новую.

Работа с null в ETL важна не меньше самих вычислений. Если не заменить отсутствующую скидку на ноль, выражение для net_amount даст null. То есть вопрос не только в чистоте данных, но и в механике вычисления.

Что проверить перед записью

Если один и тот же результат вы хотите и посмотреть, и потом записать, есть смысл вызвать cache(). Spark SQL кэширует DataFrame в памяти в колонночном формате, а после первой материализации этот кэш можно увидеть на вкладке Storage в веб-интерфейсе Spark.

result = result.cache()

Сам по себе cache() ничего не материализует: кэш появится после действия, например записи или ограниченного просмотра данных.

Перед записью полезно вызвать explain(). Это не «магическая оптимизация», а способ увидеть план выполнения: откуда читаются данные, где стоят фильтры и не возник ли лишний обмен данными. Если доступен режим стоимости, можно использовать и explain(mode="cost") для оценки плана.

result.explain()

Ещё один практический вопрос — сколько файлов получится на выходе. В Spark число выходных файлов обычно связано с числом разделов DataFrame на момент записи. Если файлов слишком много, можно уменьшить количество разделов через coalesce(...). Если же перед записью нужно перераспределить данные, используют repartition(...). Это особенно актуально перед partitionBy, чтобы избежать чрезмерного дробления результата.

Запись в Parquet и пост-проверка результата

Для минимального ETL Parquet удобен тем, что сохраняет схему и хорошо подходит для последующего чтения Spark.

(
    result.write
    .mode("overwrite")
    .partitionBy("event_date")
    .parquet("/data/out/orders_parquet")
)

partitionBy раскладывает данные по каталогам вида event_date=.... Это похоже на схему партиционирования Hive и полезно, когда последующие чтения часто фильтруют набор по дате или другому ключу.

После записи разумно не считать задачу завершённой, а перечитать сохранённый набор как новый источник:

saved = spark.read.parquet("/data/out/orders_parquet")
saved.printSchema()

Spark умеет автоматически обнаруживать партиции по структуре каталогов built-in источников, поэтому колонка event_date будет восстановлена из путей. Можно проверить и отдельную партицию, прочитав конкретный каталог с её значением.

Типовые сбои и ограничения такого учебного ETL

• Несовпадение схемы и заголовков CSV. Если схема ожидает одни поля, а в файле другие заголовки, принудительное применение схемы может скрыть проблему. Поэтому header=True и enforceSchema=False — безопаснее для контроля соответствия.
• Смешанные файлы во входной папке. Если в каталоге лежат не только CSV, Spark может прочитать лишнее и получить неожиданные колонки или значения.
• Повреждённые файлы. Опция ignoreCorruptFiles помогает не падать на чтении, но одновременно означает, что часть входа будет пропущена.
• Большая выгрузка в driver. collect() возвращает все строки и подходит только тогда, когда весь результат помещается в память driver. toLocalIterator() ведёт себя иначе: документация оценивает его память размером крупнейшего раздела, а при prefetch — до двух крупнейших разделов; для Spark Connect параметр prefetchPartitions не действует. Итератор не равнозначен collect(), но большой раздел всё равно может исчерпать локальную память.

Если свести всё к одной практической мысли, минимальный ETL в Spark — это не только «прочитал и записал». Надёжность появляется там, где вы явно задаёте схему, понимаете поведение CSV-парсера, контролируете null и проверяете результат повторным чтением уже записанного Parquet-набора.