Недавно я в очередной раз услышал:

“CSV — это популярный формат хранения данных, имеющий встроенную поддержку в Apache Spark…”

Нууу, на счет “популярный” — согласен, “имеющий встроенную поддержку” — согласен, но на счет “хранения данных” — категорически не согласен. Подобные фразы могут не только сбить с толку окружающих, но и привести к значительным непродуктивным затратам времени (и памяти данных). Давайте разберемся.

В этой статье я расскажу о массовом чтении (то есть методе .load() Apache Spark) в рамках Structured API и реализации pyspark (сомневаюсь, что есть отличия в работе в биндингах Scala/Java).

Итак, давайте освежим в памяти несколько общих моментов:

• spark использует «ленивые вычисления» (lazy evaluation), означающие, что трансформации (transformations) вычисляются только тогда, когда действие (action) требует, чтобы результат был возвращен управляющей программе.

• .load() — это трансформация, поэтому не следует ожидать какой-либо значительной дисковой активности во время выполнения .load() (значительность является важной, далее мы разберем это чуть подробнее).

• реальная загрузка данных (имеется в виду процесс чтения с диска) происходит, когда мы выполняем .save() или любое другое действие.

Пока все звучит достаточно безобидно, не так ли?

“…практика как критерий истины...”

Теперь посмотрим, что происходит на практике (для простоты я вставил фрагменты из Jupyter notebook)

ШАГ 1. Выполните простое трансформацию чтения CSV 

ШАГ 2. Распечатайте полученную схему датафрейма с помощью .printSchema()

ШАГ 3. Сохраните датафрейм в файл (сейчас формат не имеет значения - пусть это снова будет CSV)

Пока ничего особенного не произошло. Файл маленький, и задержку трудно заметить на первом шаге (во время выполнения .load()). Что ж, нам в помощь есть графический интерфейс Spark, и вот что показывает наше маленькое приложение:

Теперь начинается “магия”... У нас есть 3 джоба, что странно: ведь количество джобов обычно соответствует количеству действий, а у нас было одно действие (.save ()), поэтому мы ожидаем, что у нас будет один джоб. Странно, не правда ли? Чтобы развеять магию, давайте снова пройдем шаг за шагом, и после каждого шага будем наблюдать за Spark GUI.

Выполните первый шаг повторно, и вот как будет выглядеть Spark GUI:

Теперь не может быть никаких сомнений, что появилось два новых джоба (с идентификаторами 3 и 4) без каких-либо действий в нашем коде, следовательно скорее всего это является результатом нашей трансформации .load(). Давайте копнем глубже и посмотрим метрики дискового ввода-вывода этих джобов:

WEB UI Spark с информацией об этапах выполнения

Интересно — файл считывается уже три раза (размер файла 37KB), какую трансформацию мы осуществили.

Давайте продолжим с нашим пошаговым разбором, повторно выполнив printSchema(): новых джобов в Spark WEB UI нет (что вполне естественно).

Повторно выполняем .save():

Как и предполагалось — еще один новый джоб: было выполнено действие, следовательно мы получили другой джоб. А как насчет метрик дискового ввода-вывода:

Ага, еще одно считывание - мы читали файл уже 4 раза, хорошо, не так ли? Такой простой “конвейер” (загрузка, сохранение) привел к неэффективному вводу-выводу с диска.

Как вам такой формат “хранения данных”? — Мне не очень.

Дело в CSV или, может быть, это у меня какие-то проблемы в конфигурации Spark? Давайте выясним — почему бы нам не повторить те же шаги для другого файлового формата, например возьмем ORC (также имеющий встроенную поддержку Apache Spark).

Я не буду перегружать вас скриншотами и расписывать шаги (шаги не меняются, разница лишь в том, что я загружаю те же данные из файла ORC) - однако результаты будут следующими:

• нет джобов после трансформации .load() (шаг 1)

• получаем схему и никаких джобов после выполнения .printSchema() (шаг 2)

• всего один джоб и одно чтение файла после выполнения .save() (шаг 3)

Итак, мы выяснили, что предыдущее странное поведение связанно с CSV, и что формат ORC ведет себя так, как ожидалось.

И вот мы подошли к сути, которую я хотел выразить этой статьей:

Никогда не храните данные, которые вы планируете обрабатывать с помощью Apache Spark, в формате CSV. Можно использовать CSV в качестве транзитного формата (например, во время операций экспорта или импорта, где это может быть необходимо), для хранения используйте форматы файлов ORC или Parquet.

Я напишу еще одну статью о Parquet и ORC, а пока по своему опыту могу сказать, что они приблизительно на одном уровне (с точки зрения производительности и хранения).

Небольшой дисклеймер и пояснения

Опытные разработчики Spark уже увидели некоторые “несоответствия” в том, что я сказал выше. Позвольте мне немного объяснить, где я “скосил углы”, чтобы лучше выразить суть. Приведенные ниже пояснения не меняют сути дела, но они необходимы, чтобы укрепить “доверие” к сказанному выше.

Вывод схемы

Вывод схемы в Spark платный:

Ожидается, что при использовании вывода схемы (.option (“inferSchema”, “true”)) данные файла будут пройдены единожды. Но почему чтение данных файла произошло дважды (см. скриншоты выше)?

И как Spark удается вывести схему из ORC файла без каких-либо джобов? Это пример “белой магии” в Spark.

Играет ли здесь какую-то роль вывод схемы? — Да, вы можете быть уверены, что вы, например, можете выполнить .sum() на значениях столбцов в случае вывода схемы и CSV, и если данные беспорядочны, Spark будет по умолчанию использовать StringType и не позволит вам суммировать их. Это важно, но очень дорого.

Я потратил дополнительное время и повторил шаги 1–3 для файла CSV без вывода схемы — файл читался всего дважды. Почему? — см. ниже.

Влияние размера файла

Мой файл был относительно небольшим (37KB), что объясняет, почему на шаге 1 (.load()) он был прочитан один раз — здесь всего лишь один первый блок. Для больших файлов — и я их тоже протестировал — поведение то же самое: без вывода схемы .load() создает джоб для чтения только первого блока данных файла.

Я предполагаю, что это необходимо, чтобы узнать количество столбцов в создаваемом .load() датафрейме.

Конец исследования

Эти последние заметки завершают мое небольшое исследование о неэффективности CSV, следите за обновлениями — темы следующих статей: ввод и вывод — важная часть инфраструктуры Apache Spark и очень энергозатратная рабочая рутина любого датаинженера. Ввод/вывод надо делать правильно!

Материал подготовлен в рамках курса «Spark Developer».

Всех желающих приглашаем на открытый урок «Приземление данных с помощью Apache Flink». На вебинаре рассмотрим проблемы чтения и записи данных из Apache Kafka, познакомимся с Apache Flink и посмотрим на стенде, как можно эти проблемы решить.
>> РЕГИСТРАЦИЯ