Объединение в цепочку нескольких заданий MapReduce в Hadoop
Во многих реальных ситуациях, где Вы применяете MapReduce, заключительные алгоритмы заканчивают тем, что были несколькими шагами MapReduce.
т.е. Map1, Reduce1, Map2, Reduce2, и так далее.
Таким образом, у Вас есть вывод от последнего, уменьшают, который необходим как вход для следующей карты.
Промежуточные данные - что-то, что Вы (в целом) не хотите сохранять, после того как конвейер был успешно завершен. Также, потому что эти промежуточные данные являются в целом некоторой структурой данных (как 'карта' или 'набор'), Вы не хотите излагать слишком много усилия в письменной форме и читающий эти пары "ключ-значение".
Каков рекомендуемый способ сделать это в Hadoop?
Существует ли (простой) пример, который показывает, как обработать эти промежуточные данные корректным способом, включая очистку позже?
4 ответа
Я думаю, что этот учебник в сети разработчиков Yahoo поможет вам в этом: Chaining Jobs
Вы используете JobClient.runJob(). Выходной путь данных из первого задания становится входным путем для вашего второго задания. Их нужно передать в качестве аргументов в задания с соответствующим кодом для их разбора и установки параметров для задания.
Я думаю, что вышеупомянутый метод может быть способом, который использовался в устаревшем API mapred, но он все равно должен работать. В новом API mapreduce будет аналогичный метод, но я не уверен, какой именно.
Что касается удаления промежуточных данных после завершения работы, вы можете сделать это в своем коде. Я уже делал это раньше, используя что-то вроде:
FileSystem.delete(Path f, boolean recursive);
Где путь - это местоположение данных в HDFS. Вам нужно убедиться, что вы удалите эти данные только после того, как они не понадобятся другим заданиям.
Вы можете использовать oozie для быстрой обработки ваших заданий MapReduce. http://issues.apache.org/jira/browse/HADOOP-5303
На самом деле существует несколько способов сделать это. Я остановлюсь на двух.
Один - через Riffle ( http://github.com/cwensel/riffle ) библиотеку аннотаций для определения зависимых вещей и "выполнения" их в порядке зависимости (топологическом).
Или вы можете использовать каскад (и MapReduceFlow) в Cascading ( http://www.cascading.org/ ). Будущая версия будет поддерживать аннотации Riffle, но сейчас это отлично работает с необработанными заданиями MR JobConf.
Вариант этого - вообще не управлять заданиями MR вручную, а разработать свое приложение, используя Cascading API. Тогда JobConf и цепочки заданий обрабатываются внутри приложения с помощью классов Cascading planner и Flow.
Таким образом, вы тратите свое время на решение проблемы, а не на механику управления заданиями Hadoop и т.д. Вы даже можете наложить сверху различные языки (например, clojure или jruby), чтобы еще больше упростить разработку и приложения. http://www.cascading.org/modules.html
Есть много способов сделать это.
(1) Каскадирование заданий
Создайте объект JobConf «job1» для первого задания и установите все параметры с «input» в качестве входного каталога и «temp» в качестве выходного каталога. Выполните это задание:
JobClient.run(job1).
Сразу под ним создайте объект JobConf "job2" для второго задания и установите все параметры с "temp" в качестве входного каталога и "output" в качестве выходного каталога. Выполните это задание:
JobClient.run(job2).
(2) Создайте два объекта JobConf и установите в них все параметры, как (1) , за исключением того, что вы не используете JobClient.run.
Затем создайте два объекта Job с настройками задания в качестве параметров:
Job job1=new Job(jobconf1);
Job job2=new Job(jobconf2);
Используя объект jobControl, вы указываете зависимости задания, а затем запускаете задания:
JobControl jbcntrl=new JobControl("jbcntrl");
jbcntrl.addJob(job1);
jbcntrl.addJob(job2);
job2.addDependingJob(job1);
jbcntrl.run();
(3) Если вам нужна структура, напоминающая Map + | Уменьшить | Map *, вы можете использовать классы ChainMapper и ChainReducer, которые поставляются с Hadoop версии 0.19 и новее.