В «Ростелекоме» мы используем 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_OPTS}'

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). При написании модульных тестов мы проверяем, что на выходе функция выдает ожидаемый результат. Мы используем класс MapDriver из пакета MRUnit, в свойствах которого устанавливаем тестируемый класс. Для этого используется метод withMapper(), входные значения withInputValue() и ожидаемый результат withOutput() или withMultiOutput(), если используется множественный вывод.

Вот наш тест.
 

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, я использовал все перечисленные методы и советую всем тоже их применять. Это особенно полезно в случае с большими инсталляциями, подобных таким, какие работают в «Ростелекоме».