How-To: Настройка среды разработки для Hadoop MapReduce

Этот пост предназначен для тех, кто хочет быстро приступить к разработке базового приложения Hadoop MapReduce.

Мы увидим, как настроить базовое приложение MR для  WordCount использования Java, Maven и Eclipse и запустить базовую программу MR в локальном режиме, который легко отладить на ранней стадии.

Предполагая, что JDK 1.6+ уже установлен, и в Eclipse есть настройка для плагина Maven, и загрузка из репозитория maven по умолчанию не ограничена.

Постановка задачи  : подсчитать вхождение каждого слова, которое появляется во входном файле, с помощью MapReduce.

Шаг 1: Добавление зависимости

Создайте проект maven в eclipse и используйте следующий код в вашем  pom.xml .

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>
  <groupId>com.saurzcode.hadoop</groupId>
  <artifactId>MapReduce</artifactId>
  <version>0.0.1-SNAPSHOT</version>
  <packaging>jar</packaging>
  <dependencies>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-client</artifactId>
<version>2.2.0</version>
</dependency>
  </dependencies>
</project>

После сохранения необходимо загрузить все необходимые зависимости для запуска базовой программы Hadoop MapReduce.

Шаг 2: Программа Mapper

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

Наш класс mapper должен расширить класс Mapper и переопределить его метод map. Когда этот метод вызывается, параметр value метода будет содержать часть строк файла, которые должны быть обработаны, а выходной параметр используется для передачи экземпляров слова.

В кластерной установке реального мира этот код будет выполняться на нескольких узлах, которые будут использоваться множеством редукторов для дальнейшей обработки.

public class WordCountMapper extends
        Mapper<Object, Text, Text, IntWritable> {
    private final IntWritable ONE = new IntWritable(1);
    private Text word = new Text();
    public void map(Object key, Text value, Context context)
            throws IOException, InterruptedException {
        String line = value.toString();
        StringTokenizer tokenizer = new StringTokenizer(line);
        while(tokenizer.hasMoreTokens()) {
            word.set(tokenizer.nextToken());
            context.write(word, ONE);
        }
    }
}

Шаг 3: Программа Редуктора

Наш редуктор расширяет класс Reducer и реализует логику для суммирования каждого вхождения токена слова, полученного от mappers. Выходные данные от Reducers будут отправляться в  выходную  папку в виде текстового файла (по умолчанию или в соответствии с настройкой в ​​программе драйвера для формата вывода), называемого как part- r-00000  вместе с  файлом _SUCCESS . 

public class WordCountReducer extends
        Reducer<Text, IntWritable, Text, IntWritable> {
    public void reduce(Text text, Iterable<IntWritable> values, Context context)
            throws IOException, InterruptedException {
        int sum = 0;
        for (IntWritable value : values) {
            sum += value.get();
        }
        context.write(text, new IntWritable(sum));
    }
}

public class WordCount {
    public static void main(String[] args) throws IOException,
            InterruptedException, ClassNotFoundException {
        Path inputPath = new Path(args[0]);
        Path outputDir = new Path(args[1]);
        // Create configuration
        Configuration conf = new Configuration(true);
        // Create job
        Job job = new Job(conf, "WordCount");
        job.setJarByClass(WordCountMapper.class);
        // Setup MapReduce
        job.setMapperClass(WordCountMapper.class);
        job.setReducerClass(WordCountReducer.class);
        job.setNumReduceTasks(1);
        // Specify key / value
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        // Input
        FileInputFormat.addInputPath(job, inputPath);
        job.setInputFormatClass(TextInputFormat.class);
        // Output
        FileOutputFormat.setOutputPath(job, outputDir);
        job.setOutputFormatClass(TextOutputFormat.class);
        // Delete output if exists
        FileSystem hdfs = FileSystem.get(conf);
        if (hdfs.exists(outputDir))
            hdfs.delete(outputDir, true);
        // Execute job
        int code = job.waitForCompletion(true) ? 0 : 1;
        System.exit(code);
    }
}

foo bar is foo count
count foo for saurzcode

foo 3
bar  1
is  1
count 2
for 1
saurzcode 1

java com.saurzcode.mapreduce.WordCount input/inputfile.txt output

output/
_SUCCESS
part-r-00000