Хабрахабр

Тестирование и отладка MapReduce

В «Ростелекоме» мы используем Hadoop для хранения и обработки данных, загруженных из многочисленных источников с помощью java-приложений. Сейчас мы переехали на новую версию hadoop с Kerberos Authentication. При переезде столкнулись с рядом проблем, в том числе и с использованием YARN API. Работа Hadoop с Kerberos Authentication заслуживает отдельной статьи, а в этой мы поговорим об отладке Hadoop MapReduce.


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

Но как проанализировать вывод System.out.println("message") разбросанных по этим узлам? Можно использовать проверенный временем System.out.println("message").

Все, что пишется в stdout или stderr,
направляется в соответствующий файл журнала, который можно найти на веб-странице расширенной информации о задаче или в журнальных файлах. Мы можем выводить сообщения в стандартный поток ошибок.

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

Приложение Hadoop MapReduce можно отлаживать во всех трех режимах, в которых может работать Hadoop:
 

  • standalone
  • pseudo-distributed mode
  • fully distributed

Более подробно мы остановимся на первых двух.
 

Pseudo-distributed mode (псевдораспределенный режим)

Псевдораспределенный режим используется для имитации реального кластера. И может использоваться для тестирования в среде, максимально приближенной к продуктивной. В данном режиме все демоны Hadoop будут работать на одном узле!

Если у вас есть dev-сервер или другая песочница (например, Virtual Machine с настроенной средой разработки, такой как Hortonworks Sanbox с HDP), то можно отладить управляющую программу, используя средства удаленной отладки (remote debugging).

Ниже приведен пример. Для запуска отладки нужно задать значение переменной окружения: YARN_OPTS. Для удобства можно создать файл startWordCount.sh и добавить в него необходимые параметры для запуска приложения.

#!/bin/bash source /etc/hadoop/conf/yarn-env.sh
export YARN_OPTS='-agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=6000 $' yarn jar wordcount-0.0.1.jar ru.rtc.example.WordCount /input /output

Теперь, запустив скрипт `./startWordCount.sh`, мы увидим сообщение

Listening for transport dt_socket at address: 6000

Осталось настроить IDE для удаленной отладки (remote debugging). Я использую intellij IDEA. Перейдем в меню Run -> Edit Configurations… Добавим новую конфигурацию Remote.

Поставим breakpoint в main и запустим.

Все, теперь мы можем отлаживать программу как обычно.

Вы должны убедиться, что работаете с последней версией исходного кода. ВНИМАНИЕ. Если нет, то у вас могут быть различия в строках, в которых отладчик выполняет остановку.

В ранних версиях Hadoop поставлялся специальный класс, который позволял повторно запустить сбойное задание — isolationRunner. Данные, вызвавшие сбой, сохранялись на диск по адресу, указанному в переменной окружения Hadoop mapred.local.dir. К сожалению, в последних версиях Hadoop такой класс больше не поставляется.

Standalone (локальный запуск)

Standalone — это стандартный режим, в котором работает Hadoop. Он подходит для отладки там, где не используется HDFS. При такой отладке можно использовать ввод и вывод через локальную файловую систему. Standalone-режим обычно является самым быстрым режимом Hadoop, так как он использует локальную файловую систему для всех входных и выходных данных.

Счетчики определяются перечислением (enum) Java. Как упоминалось ранее, можно внедрить в код средство отладки, например, счетчики. Счетчик может пригодиться для оценки проблемы,
и может использоваться как дополнение к отладочному выводу. Имя перечисления определяет имя группы, а поля перечисления определяют имена счетчиков.

Объявление и использование счетчика:

package ru.rt.example; import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper; public class Map extends Mapper<LongWritable, Text, Text, IntWritable> { private Text word = new Text(); enum Word { TOTAL_WORD_COUNT, } @Override public void map(LongWritable key, Text value, Context context) { String[] stringArr = value.toString().split("\\s+"); for (String str : stringArr) { word.set(str); context.getCounter(Word.TOTAL_WORD_COUNT).increment(1); } }
}
}

Для инкремента счетчика нужно использовать метод increment(1).

...
context.getCounter(Word.TOTAL_WORD_COUNT).increment(1);
...

После успешного завершения MapReduce задача в конце выводит значения счетчиков.

Shuffle Errors BAD_ID=0 CONNECTION=0 IO_ERROR=0 WRONG_LENGTH=0 WRONG_MAP=0 WRONG_REDUCE=0 ru.rt.example.Map$Word TOTAL_WORD_COUNT=655

Ошибочные данные можно выводить в stderr или в stdout, или писать выходные данные в hdfs, используя класс MultipleOutputs для дальнейшего анализа. Полученные данные можно передавать на вход приложению в standalone режиме или при написании unit-тестов.

При написании модульных тестов мы проверяем, что на выходе функция выдает ожидаемый результат. В Hadoop есть библиотека MRUnit, которая используется совместно с фреймворками тестирования (например, JUnit). Для этого используется метод withMapper(), входные значения withInputValue() и ожидаемый результат withOutput() или withMultiOutput(), если используется множественный вывод. Мы используем класс MapDriver из пакета MRUnit, в свойствах которого устанавливаем тестируемый класс.

Вот наш тест.
 


package ru.rt.example; import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mrunit.mapreduce.MapDriver;
import org.apache.hadoop.mrunit.types.Pair;
import org.junit.Before;
import org.junit.Test; import java.io.IOException; public class TestWordCount { private MapDriver<Object, Text, Text, IntWritable> mapDriver; @Before public void setUp() { Map mapper = new Map(); mapDriver.setMapper(mapper) } @Test public void mapperTest() throws IOException { mapDriver.withInput(new LongWritable(0), new Text("msg1")); mapDriver.withOutput(new Pair<Text, IntWritable>(new Text("msg1"), new IntWritable(1))); mapDriver.runTest(); }
}

Fully distributed mode (полностью распределенный режим)

Как следует из названия, это режим, в котором используется вся мощность Hadoop. Запущенная программа MapReduce может работать на 1000 серверов. Всегда сложно отлаживать программу MapReduce, так как у вас есть Mappers, работающие на разных машинах с разными входными данными.

Заключение

Как оказалось, тестирование MapReduce не такая простая задача как кажется на первый взгляд.
Чтобы сэкономить время в поисках ошибки в MapReduce, я использовал все перечисленные методы и советую всем тоже их применять. Это особенно полезно в случае с большими инсталляциями, подобных таким, какие работают в «Ростелекоме».

Теги
Показать больше

Похожие статьи

Добавить комментарий

Ваш адрес email не будет опубликован. Обязательные поля помечены *

Кнопка «Наверх»
Закрыть