Как оптимизировать фазу перетасовки/сортировки в задании hadoop

Question

Я занимаюсь подготовкой некоторых данных, используя одноуровневое задание hadoop. Картограф/сумматор в моей работе выводит много ключей (более 5 М или 6 М), и, очевидно, работа продолжается медленно или даже не срабатывает. Фаза отображения работает до 120 картографов, и есть только один редуктор (они автоматически определяются, и я не задал для них никаких значений). Я хочу оптимизировать работу, чтобы фаза перетасовки/сортировки выполнялась более эффективно. Я увеличил mapreduce.task.io.sort.mb до 300 м, но работа исчезла из-за того, что его значение было больше, чем куча карт. Затем я установил mapred.child.java.opts в -Xmx1024m, но он снова потерпел неудачу, потому что он не смог инициализировать выходной коллектор. Каковы наилучшие методы для этих сценариев?

Answer 1

По умолчанию mapreduce выбирает только один редуктор, который будет проблемой для большинства сценариев.

Либо вы можете изменить код, установив количество переходников на более высокое значение job.setNumReduceTasks(24);

или запустить hadoop jar <jarfilename> <class> -Dmapreduce.job.reduces=24 <parameters>

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

Answer 2

Для начала, поскольку вы используете кластер с одним узлом, вы не можете оптимизировать свою работу. У вас будет ограниченное количество контейнеров/слотов в кластере с одним узлом и количество обрабатываемых данных (от 5 до 6 миллионов ключей), ваши задания будут работать медленно и могут также не работать.

Я собираюсь ответить на этот вопрос для полностью распределенной настройки Hadoop. В книге «Hadoop The Definitive Guide» есть раздел («Shuffle and Sort»), который вы должны прочитать для настройки фазы Shuffle и Sort. На мой ответ в основном влияет содержание этого раздела, а также мой собственный опыт настройки настроек MapReduce.

Вы можете сделать следующее для достижения Перемешать и сортировку эффективности:

• Combiner: Использование объединителя уменьшит количество данных, передаваемых по каждому из редукторов, так как объединитель объединяет выход на сторона карты.
• Количество редукторов: Выберите оптимальное количество редукторов. Если размер данных огромен, то один редуктор не является хорошей идеей. Кроме того, установка количества редукторов на большое число не является хорошей идеей, так как количество редукторов также определяет количество разделов на стороне карты. Посмотрите на ссылку здесь: https://github.com/paulhoule/infovore/wiki/Choosing-the-number-of-reducers
• Когда начинать редукторы:; Вы можете контролировать, когда запускаются задачи сокращения. Это определяется конфигурацией mapreduce.job.reduce.slowstart.completedmaps в YARN. Он не запускает редукторы до тех пор, пока не будет достигнут определенный процент картографов. По умолчанию установлено значение «0,05» (это означает, что редукторы начинаются после завершения 5% картографов). Если редукторы запускаются раньше, то большинство редукторов простаивают до тех пор, пока все мапперы не будут завершены. Кроме того, редукторы могут потреблять слоты, которые в противном случае могли бы использоваться картографами для обработки. Контролируя это, вы можете оптимально использовать слоты карты/редуктора и улучшить время, затрачиваемое во время тасования.
• Compress Mapper Output: Рекомендуется сжать выходные данные карты (определяется конфигурацией: mapreduce.map.output.compress), так что меньшие данные записываются на диск и передаются на редукторы.
• Tune config "mapreduce.task.io.sort.mb": Увеличьте размер буфера, используемый мапперами во время сортировки. Это уменьшит количество разливов на диске.
• Tune config "mapreduce.reduce.input.buffer.percent": Если ваша задача уменьшения имеет меньшие требования к памяти, тогда это значение может быть установлено на высокий процент. Это означает, что большее количество кучи используется для сохранения выходов карты во время фазы уменьшения (после фазы перетасовки), тем самым уменьшая количество разливов на диск.
• Tune config "mapreduce.reduce.shuffle.parallelcopies": Количество потоков, используемых для копирования выходов карты на редукторы. Проверьте ссылку здесь: how to tune mapred.reduce.parallel.copies?

Ниже приведены другие параметры конфигурации, которые могут быть настроены для улучшения Перемешать и сортировку производительности фазы (см описания этих конфигураций здесь: https://hadoop.apache.org/docs/r2.4.1/hadoop-mapreduce-client/hadoop-mapreduce-client-core/mapred-default.xml):

• mapreduce.map.sort.spill.percent: Определяет порог для буфера памяти, используемого устройством сопоставления. Когда этот порог достигнут, содержимое буфера выливается на диск. Таким образом, это значение определяет количество разливов на диск
• mapreduce.task.io.sort.factor: Минимальное количество потоков, которые необходимо объединить сразу, во время сортировки. Итак, на стороне редуктора, если имеется 50 выходов отображения, и это значение равно 10, тогда будет 5 раундов слияния (в среднем 10 файлов для объединения раунда).
• mapreduce.shuffle.max.threads: Количество рабочих потоков для копирования выходов карты на редукторы.
• mapreduce.reduce.shuffle.input.buffer.percent: Сколько кучи должно использоваться для хранения вывода карты во время фазы перетасовки в редукторе. Этот параметр определяет объем вывода карты, который может храниться в памяти, прежде чем он будет выведен на диск.
• mapreduce.reduce.shuffle.merge.percent: Порог для начала процесса слияния и проливая на диск
• mapreduce.reduce.merge.inmem.threshold: Количество выходов карты, необходимые для запуска процесса слияния. Когда достигнут mapreduce.reduce.shuffle.merge.percent или mapreduce.reduce.merge.inmem.threshold, выходы карты объединяются и разливаются на диск.