hadoop
Inleiding tot MapReduce

Syntaxis

• Om het voorbeeld uit te voeren, is de opdrachtsyntaxis:

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

• Gegevens naar HDFS kopiëren (van lokaal):

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

Opmerkingen

Word Count Program (in Java en Python)

Het programma voor het tellen van woorden lijkt op het programma "Hallo wereld" in MapReduce.

Hadoop MapReduce is een softwareframework voor het eenvoudig schrijven van toepassingen die enorme hoeveelheden gegevens (gegevensreeksen van meerdere terabytes) parallel verwerken op grote clusters (duizenden knooppunten) van hardware op een betrouwbare, fouttolerante manier.

Een MapReduce-taak splitst de ingevoerde gegevensset meestal in onafhankelijke blokken die door de kaarttaken op een volledig parallelle manier worden verwerkt. Het raamwerk sorteert de uitvoer van de kaarten, die vervolgens worden ingevoerd om de taken te verminderen. Gewoonlijk worden zowel de invoer als de uitvoer van de taak opgeslagen in een bestandssysteem. Het framework zorgt voor het plannen van taken, het bewaken ervan en voert de mislukte taken opnieuw uit.

Voorbeeld van woorden tellen:

WordCount-voorbeeld leest tekstbestanden en telt hoe vaak woorden voorkomen. De invoer is tekstbestand en de uitvoer is tekstbestand, waarvan elke regel een woord bevat en de telling van hoe vaak het voorkomt, gescheiden door een tab.

Elke mapper neemt een regel als invoer en verdeelt deze in woorden. Het zendt vervolgens een sleutel / waarde-paar van het woord uit en elk verloopstuk telt de tellingen voor elk woord en zendt een enkele sleutel / waarde uit met het woord en de som.

Als optimalisatie wordt het verloopstuk ook gebruikt als een combiner op de kaartuitgangen. Dit vermindert de hoeveelheid gegevens die over het netwerk worden verzonden door elk woord in één record te combineren.

Word Count Code:

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);
 }
        
}

Om het voorbeeld uit te voeren, is de opdrachtsyntaxis:

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

Alle bestanden in de invoermap (in de bovenliggende opdrachtregel genoemd) worden gelezen en de tellingen van woorden in de invoer worden naar de uitvoermap geschreven (hierboven out-dir genoemd). Er wordt verondersteld dat zowel invoer als uitvoer in HDFS zijn opgeslagen. Als uw invoer zich niet al in HDFS bevindt, maar zich ergens in een lokaal bestandssysteem bevindt, moet u de gegevens naar HDFS kopiëren met een opdracht als deze:

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

Voorbeeld van Word Count in Python:

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)

reducer.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)

Het bovenstaande programma kan worden uitgevoerd met behulp van cat filename.txt | python mapper.py | sort -k1,1 | python reducer.py