hadoop
Wprowadzenie do MapReduce

Składnia

• Aby uruchomić przykład, składnia polecenia to:

  bin/hadoop jar hadoop-*-examples.jar wordcount [-m <#maps>] [-r <#reducers>] <in-dir> <out-dir>
  

• Aby skopiować dane do HDFS (z lokalnego):

  bin/hadoop dfs -mkdir <hdfs-dir> //not required in hadoop 0.17.2 and later
  bin/hadoop dfs -copyFromLocal <local-dir> <hdfs-dir>
  

Uwagi

Program Word Count (w Javie i Python)

Program liczenia słów jest podobny do programu „Hello World” w MapReduce.

Hadoop MapReduce to platforma programowa do łatwego pisania aplikacji, które przetwarzają równolegle duże ilości danych (zestawy danych terabajtowych) na dużych klastrach (tysiące węzłów) sprzętu towarowego w niezawodny, odporny na uszkodzenia sposób.

Zadanie MapReduce zwykle dzieli zestaw danych wejściowych na niezależne porcje, które są przetwarzane przez zadania mapy w całkowicie równoległy sposób. Struktura sortuje dane wyjściowe map, które są następnie wprowadzane do zadań redukcji. Zazwyczaj zarówno dane wejściowe, jak i wyjściowe zadania są przechowywane w systemie plików. Struktura zajmuje się planowaniem zadań, monitorowaniem ich i ponownym wykonywaniem zadań zakończonych niepowodzeniem.

Przykład liczby słów:

Przykład WordCount czyta pliki tekstowe i liczy, jak często występują słowa. Dane wejściowe to pliki tekstowe, a dane wyjściowe to pliki tekstowe, z których każdy wiersz zawiera słowo i liczbę ich występowania, oddzielone tabulatorem.

Każdy twórca map pobiera wiersz jako dane wejściowe i dzieli go na słowa. Następnie emituje parę klucz / wartość słowa, a każdy reduktor sumuje liczby dla każdego słowa i emituje pojedynczy klucz / wartość ze słowem i sumą.

Jako optymalizacja reduktor jest również używany jako sumator na wyjściach mapy. Zmniejsza to ilość danych przesyłanych przez sieć, łącząc każde słowo w jeden rekord.

Kod liczby słów:

package org.myorg;
        
import java.io.IOException;
import java.util.*;
        
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
        
public class WordCount {
        
 public static class Map extends Mapper<LongWritable, Text, Text, IntWritable> {
    private final static IntWritable one = new IntWritable(1);
    private Text word = new Text();
        
    public void map(LongWritable 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);
        }
    }
 } 
        
 public static class Reduce extends Reducer<Text, IntWritable, Text, IntWritable> {

    public void reduce(Text key, Iterable<IntWritable> values, Context context) 
      throws IOException, InterruptedException {
        int sum = 0;
        for (IntWritable val : values) {
            sum += val.get();
        }
        context.write(key, new IntWritable(sum));
    }
 }
        
 public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();
        
        Job job = new Job(conf, "wordcount");
    
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(IntWritable.class);
        
    job.setMapperClass(Map.class);
    job.setReducerClass(Reduce.class);
        
    job.setInputFormatClass(TextInputFormat.class);
    job.setOutputFormatClass(TextOutputFormat.class);
        
    FileInputFormat.addInputPath(job, new Path(args[0]));
    FileOutputFormat.setOutputPath(job, new Path(args[1]));
        
    job.waitForCompletion(true);
 }
        
}

Aby uruchomić przykład, składnia polecenia to:

bin/hadoop jar hadoop-*-examples.jar wordcount [-m <#maps>] [-r <#reducers>] <in-dir> <out-dir>

Wszystkie pliki w katalogu wejściowym (zwanym in-dir w wierszu poleceń powyżej) są odczytywane, a liczba słów na wejściu jest zapisywana w katalogu wyjściowym (zwanym powyżej out-dir). Zakłada się, że zarówno dane wejściowe, jak i wyjściowe są przechowywane w HDFS. Jeśli dane wejściowe nie są już w HDFS, ale znajdują się gdzieś w lokalnym systemie plików, musisz skopiować dane do HDFS za pomocą polecenia takiego:

bin/hadoop dfs -mkdir <hdfs-dir> //not required in hadoop 0.17.2 and later
bin/hadoop dfs -copyFromLocal <local-dir> <hdfs-dir>

Przykład liczby słów w Pythonie:

mapper.py

import sys 
for line in sys.stdin: 
    # remove leading and trailing whitespace 
    line = line.strip() 
    # split the line into words 
    words = line.split() 
    # increase counters 
    for word in words: 
        print '%s\t%s' % (word, 1)

reduktor.py

import sys
current_word = None
current_count = 0
word = None
for line in sys.stdin:
    # remove leading and trailing whitespaces
    line = line.strip()
    # parse the input we got from mapper.py
    word, count = line.split('\t', 1)
    # convert count (currently a string) to int
    try:
        count = int(count)
    except ValueError:
        # count was not a number, so silently
        # ignore/discard this line
        continue
    if current_word == word:
        current_count += count
    else:
        if current_word:
            print '%s\t%s' % (current_word, current_count)
        current_count = count
        current_word = word
if current_word == word:
    print '%s\t%s' % (current_word, current_count)

Powyższy program można uruchomić za pomocą cat filename.txt | python mapper.py | sort -k1,1 | python reducer.py