Перетасовка, или Shuffle, — это операция перемещения данных по узлам в результате выполнения операций соединения или агрегации Apache Spark. Её можно наблюдать при использовании преобразований join, distinct, groupBy, orderBy и др.
Стоимость использования shuffle
Перетасовка (shuffle) — это ещё и дорогостоящая операция, ведь требуется копировать данные по исполнителям, а также перетасовка включает:
- операции ввода-вывода,
- сериализацию и десериализацию данных,
- перемещение по сети.
Для организации данных для перетасовки Spark генерирует набор задач отображений (map) и набор задач сверток (reduce). Это организация пришла из MapReduce и не связана с операциями Spark’а map и reduce.
Результаты каждой map-задачи хранятся в памяти, пока они не понадобятся. Затем они в отсортированном порядке в заданной партиции будут записаны в файл. Для сверток задачи читаются из соответствующих сортированных блоках.
Некоторые операции перетасовки расходуют много памяти в куче, поскольку там хранятся структуры данных, которые необходимы как до, так и после размещения записей. Например, reduceByKey и aggregateByKey создают такие структуры данных на стороне отображений, а ByKey — на стороне сверток. Если памяти не хватает, то Spark запишет таблицы на диск, что приводит к медленной операции ввода-вывода на диске с последующим вызовом сборщика мусора.
Перетасовка также генерирует болшое количество промежуточных файлов на диске. Начиная с Spark 1.3, эти файлы сохраняются до тех пор, пока соответствующие RDD становится больше не нужен, они затем уничтожаются сборщиком мусора. Хранение файлов позволяет не проводить повторную перетасовку, ведь данные могут ещё понадобятся. Сборка мусора может не происходить ещё долгое время, если ссылки на RDD существуют. Это означает, что длительные задачи будут потреблять много дискового пространства.
Как работает перетасовка
Рассмотрим вызов метода reduceByKey. Он генерирует новый RDD, в котором все значения для одного ключа объединены в кортеж в виде ключа и результата функции свертки над значениями, которые ассоциированы с ключом. Проблема заключается в том, что все значения ключа могут и не оставаться на той же самой партиции или даже той же самой машине. Но они должны быть размещены вместе для подсчета результата.
В Spark данные, в большинстве случаев, не распределяются по партициям для того, чтобы быть в нужном месте для некоторой операции. Во время вычислений одна задача будет выполняться в одной партиции. Так, чтобы организовать все данные, необходимые для reduceByKey, Spark произведет вычисления везде и всюду. Следовательно, он прочитает из всех партиций для того, чтобы найти все значения каждого ключа, затем перенесет все эти значения по партициям для вычисления результата для каждого ключа. Все эти переносы и называются перетасовкой.
Несмотря на то, что элементы в каждой партиции новых “перетасовочных” данных определены, так же как и порядок самих партиций, то порядок элементов — нет.