Как улучшить загрузку тысяч крошечных файлов JSON в фрейм данных Spark?

Question

У меня есть около 30000 очень маленьких JSON-файлов, которые я пытаюсь загрузить в информационный фрейм Spark (из смонтированной корзины S3). Сообщается здесь и здесь , что могут быть проблемы с производительностью, и описывается как Hadoop Small Files Problem. В отличие от того, о чем ранее сообщалось, я не возвращаюсь в каталоги (так как все мои файлы JSON находятся в одной подпапке). Мой код для загрузки файлов JSON выглядит следующим образом.

val df = spark
  .read
  .option("multiline", "true")
  .json("/mnt/mybucket/myfolder/*.json")
  .cache

Пока моя работа "застряла". Я вижу 2 этапа.

• Задание 0, Этап 0: список конечных файлов и каталогов
• Задание 1, этап 1: val df = spark .read .option ("multiline", "...

Job 0, Stage 0 довольно быстро, менее 1 минуты. Job 1, Stage 1, однако, требуется вечность, чтобы даже появиться (потерянный счет времени, но между двумя мы говорим более 20 минут), и когда он появляется в пользовательском интерфейсе заданий, он, кажется, "застрял" ( Я все еще жду о любом прогрессе, о котором сообщат через 15 + минут). Интересно, что Job 0, Stage 0 имеет 200 задач (я вижу, что используются 7 исполнителей), а Job 1, Stage 1 имеет только 1 задачу (кажется, что используется только 1 узел / исполнитель! Что за пустая трата!).

Есть ли способ сделать этот, казалось бы, простой шаг загрузки 30 000 файлов быстрее или более быстрым?

Я подумал о том, чтобы просто «объединить» эти файлы в большие; Например, объедините 1000 файлов JSON в 30 больших (используя NDJSON ). Однако я скептически отношусь к этому подходу, поскольку объединение файлов (скажем, с использованием Python) само по себе может занять много времени (что-то вроде собственной команды linux ls в этом каталоге занимает очень много времени для возврата); Кроме того, этот подход может нанести ущерб цели сквозных кластерных вычислений (не очень элегантно).

Answer 1

Есть два HTTP-запроса на чтение, один HEAD, один GET; если все файлы хранятся в одном и том же каталоге, тогда стоимость листинга составляет всего один объект LIST / 5000, поэтому 6 вызовов списка. Вы заплатите ~ $ 25 за 30K HEAD & GET.

Если вы используете spark, чтобы составить список и сгенерировать запись для каждого отдельного файла, а также накладные расходы по планированию задачи для файла. Вы можете сделать трюк, когда создаете сам листинг (который вы делаете в .py), который становится входным RDD (то есть одна строка на файл), а map () становится чтением этого файла и выводом карты запись, представляющая один файл. пример скалы . Это устраняет накладные расходы, связанные с планированием зажигания, поскольку этот список входных данных будет разбит на большие части, которые будут передаваться работникам, оставляя только те вызовы HTTP HEAD / GET.

Чтобы это работало эффективно, используйте Jad-файлы Hadoop 2.8+ и выполните перечисление с помощью FileSystem.listFiles(Path, true) для единственного рекурсивного перечисления всего дерева каталогов под путем, поэтому используйте API S3 LIST в наиболее оптимальном варианте.

(Как только вы это сделаете, почему бы не опубликовать код где-нибудь для других?)

Answer 2

Объединение файлов JSON в символы новой строки, разделенные символом новой строки, гораздо более крупный (нацеленный на один или самое большее 10 файлов, а не 30), был бы единственным вариантом здесь.

Открытие 30K файлов на Python не будет медленнее, чем то, что вы уже делаете, просто не будет распространяться.

Кроме того, multiline=true был добавлен только в тех случаях, когда у вас уже есть действительно большой JSON-файл и это один массив или объект верхнего уровня, который хранится. До того, как эта опция существовала, "JSONLines" был единственным форматом, который Spark мог прочитать.

Наиболее последовательным решением здесь было бы исправить конвейер приема, который записывает все эти файлы, так что вы можете накапливать записи заранее, а затем выгружать большие пакеты. Или просто используйте Kafka вместо чтения данных из S3 (или любой аналогичной файловой системы)