Старый и новый Hadoop API

Все начинающие hadoop  разработчики рано или поздно обнаруживают, что все примеры готовых hadoop приложений написаны с использованием разных map и reduce функций, разного наследования классов от производного класса, с разным способом запуска задач.

Дело в том. что начиная с версии 0.20 у hadoop появился новый API, который несовместим со старым, хотя они и достаточно похожи друг на друга. Под старый API написано уже много кода, от которого не все еще готовы отказаться, поэтому сейчас два hadoop API существуют параллельно, что создает большую путаницу.

В разных версиях хадупа статусы старого и нового API различаются:

• в версии 0.20 новый API реализован частично, старый API рекомендован как основной
• в версиях 0.22 и 0.23 старый API помечен как устаревший (deprecated) и рекомендуется пользоваться новым
• однако в hadoop 1.x  новый API имеет статус "реализован частично", а старый - рекомендован для использования!

Я сам очень удивился, когда, читая доки по старому и новому API, обнаружил метки «deprecated» у старых классов в версиях 0.22 и 0.23 и не увидел ее в версии 1.01. Новый API не так понравился разработчикам как его старая версия? Как выяснилось из умной книжки, версия hadoop 1.0 была сделана из ветки hadoop 0.20 как самой стабильной и унаследовала от нее свое состояние дел. Поэтому в Hadoop 1.x рекомендуется писать свой код на старом API.

Разработчики хадупа иногда называют новый API «интерфейсом контекстных объектов» (context objects), так как в новом API все взаимодействие map-reduce приложений происходит через так называемый контекст. Новый API заточен на то, чтобы его было максимально просто дорабатывать в дальнейшем. Поэтому в нем многие интерфейсы были заменены абстрактными классами. В абстрактный класс гораздо проще добавить новый перегружаемый метод, сохранив при этом поддержку существующего кода.

В чем же разница между старым и новым API?

• Новый API живет в пакете пакете org.apache.hadoop.mapreduce,
  тогда как старый API находится в org.apache.hadoop.mapred

• Задачи теперь управляются через класс Job, а не JobClient

• В старом api в выходной директории создавались файлы
  

      part-00000
      part-00001
      part-00002
      ..........

  в новом выходные файлы называются
  

      part-m-00000
      part-m-00001
      part-m-00002
      ..........

  если они создаются после map стадии, или
  

      part-r-00000
      part-r-00001
      part-r-00002
      ..........

  если они получаются после операции reduce
• Коллекции ключей на входе метода reduce имеют тип Iterable <...> вместо Iterator <...>. В результате перебор значений в цикле становится более изящным и наглядным.
  Метод reduce старого API:

protected void reduce(KeyClass key,
                      Iterator<ValueClass> values, ...) {
    while (values.hasNext()) {
        ValueClass currentValue = clicksCounts.next ().get ();
        // ...
    }
}

тот же метод в новом стиле:

protected void reduce (KeyClass key, 
                       Iterable <ValueClass> values, ...) {
    for (ValueClass currentValue : values) {
        // ...
    }
}

• Новый hadoop API позволяет мапперам и редьюсерам контролировать процесс обработки входных значений через перегрузку метода run (). Оригинальный метод состоит из четырех строчек и его можно найти в исходном коде hadoop'а: 

public void run(Contextcontext) 
        throws IOException, InterruptedException
 {
    setup(context);
    while(context.nextKeyValue()) {
        map(context.getCurrentKey(), 
            context.getCurrentValue(), 
            context);
    }
    cleanup(context);
}

В том же исходном коде можно найти сложные примеры мапперов с перегруженным run'ом: ChainMapper и MultithreadMapper. В старом API это было возможно только для мапперов перегрузкой MapRunnable

• Методы map и reduce теперь выбрасывают не один, а два типа исключений. К существовавшему раньше исключению IOException, выбрасывающемуся по любому поводу добавилось еще InterrupdedException, которое вылетает из методов класса Context  при принудительной внешней остановке маппера и редьюсера. Его можно перехватывать и обрабатывать в методе run () для корректного завершения работы

• После введения API «контекстных объектов» сигнатуры методов map и reduce изменились. И это надо учитывать при переводе своих старых приложений на новый API

Для демонстрации разницы старого и нового API я написал простенькую hadoop-утилиту, которое определяет число кликов каждого из пользователей на выбранном сайте. Исходными данными для нее являются логи Апача, которые можно нагенерировать уже упоминавшимся генератором логов. Подробней про лог апача можно почитать в этой статье.
Проект выложен на github и состоит из трех классов: парсера апачовых логов и двух реализаций подсчета кликов пользователей на старом и на новом API. Собрать код можно ant'ом, предварительно указав верные пути до ваших библиотек hadoop'а. Задачу написанную на старом API можно запустить командой

hadoop jar find_clicks_count.jar \
    FindClicksCountOldApi input_dir1 input_dir2... output_dir

А с использованием нового программного интерфейса:

hadoop jar find_clicks_count.jar \
    FindClicksCountNewApi input_dir1 input_dir2... output_dir 

Информация про старый и новый API была взята из книжки Hadoop TheDefinitie Guide Тома Вайта.