MapReduce - Partitioner
Un partitioner funziona come una condizione nell'elaborazione di un set di dati di input. La fase di partizione avviene dopo la fase Mappa e prima della fase Riduci.
Il numero di partizionatori è uguale al numero di riduttori. Ciò significa che un partizionatore dividerà i dati in base al numero di riduttori. Pertanto, i dati passati da un singolo partitioner vengono elaborati da un singolo Reducer.
Partitioner
Un partizionatore partiziona le coppie chiave-valore degli output intermedi della mappa. Partiziona i dati utilizzando una condizione definita dall'utente, che funziona come una funzione hash. Il numero totale di partizioni è uguale al numero di attività di Reducer per il lavoro. Facciamo un esempio per capire come funziona il partitioner.
Implementazione del partizionatore MapReduce
Per comodità, supponiamo di avere una piccola tabella chiamata Employee con i seguenti dati. Useremo questi dati di esempio come set di dati di input per dimostrare come funziona il partitioner.
Id | Nome | Età | Genere | Stipendio |
---|---|---|---|---|
1201 | gopal | 45 | Maschio | 50.000 |
1202 | manisha | 40 | Femmina | 50.000 |
1203 | khalil | 34 | Maschio | 30.000 |
1204 | prasanth | 30 | Maschio | 30.000 |
1205 | kiran | 20 | Maschio | 40.000 |
1206 | laxmi | 25 | Femmina | 35.000 |
1207 | bhavya | 20 | Femmina | 15.000 |
1208 | reshma | 19 | Femmina | 15.000 |
1209 | kranthi | 22 | Maschio | 22.000 |
1210 | Satish | 24 | Maschio | 25.000 |
1211 | Krishna | 25 | Maschio | 25.000 |
1212 | Arshad | 28 | Maschio | 20.000 |
1213 | lavanya | 18 | Femmina | 8.000 |
Dobbiamo scrivere una domanda per elaborare il set di dati di input per trovare il dipendente più salariato in base al sesso in diverse fasce di età (ad esempio, sotto i 20 anni, tra i 21 ei 30 anni, sopra i 30).
Dati in ingresso
I dati di cui sopra vengono salvati come input.txt nella directory "/ home / hadoop / hadoopPartitioner" e fornito come input.
1201 | gopal | 45 | Maschio | 50000 |
1202 | manisha | 40 | Femmina | 51000 |
1203 | khaleel | 34 | Maschio | 30000 |
1204 | prasanth | 30 | Maschio | 31000 |
1205 | kiran | 20 | Maschio | 40000 |
1206 | laxmi | 25 | Femmina | 35000 |
1207 | bhavya | 20 | Femmina | 15000 |
1208 | reshma | 19 | Femmina | 14000 |
1209 | kranthi | 22 | Maschio | 22000 |
1210 | Satish | 24 | Maschio | 25000 |
1211 | Krishna | 25 | Maschio | 26000 |
1212 | Arshad | 28 | Maschio | 20000 |
1213 | lavanya | 18 | Femmina | 8000 |
Sulla base dell'input fornito, di seguito viene fornita la spiegazione algoritmica del programma.
Attività sulla mappa
L'attività di mappa accetta le coppie chiave-valore come input mentre abbiamo i dati di testo in un file di testo. L'input per questa attività di mappa è il seguente:
Input - La chiave sarebbe un modello come "qualsiasi chiave speciale + nome file + numero di riga" (esempio: chiave = @ input1) e il valore sarebbe i dati in quella riga (esempio: valore = 1201 \ t gopal \ t 45 \ t Maschio \ t 50000).
Method - Il funzionamento di questa attività di mappa è il seguente:
Leggi il value (record data), che viene fornito come valore di input dall'elenco di argomenti in una stringa.
Utilizzando la funzione split, separare il sesso e memorizzare in una variabile stringa.
String[] str = value.toString().split("\t", -3);
String gender=str[3];
Invia le informazioni sul sesso e i dati del record value come coppia chiave-valore di output dall'attività di mappa al file partition task.
context.write(new Text(gender), new Text(value));
Ripetere tutti i passaggi precedenti per tutti i record nel file di testo.
Output - Otterrai i dati sul sesso e il valore dei dati del record come coppie chiave-valore.
Attività partizionatore
L'attività di partizionamento accetta le coppie chiave-valore dall'attività di mappa come input. La partizione implica la divisione dei dati in segmenti. In base ai criteri condizionali delle partizioni forniti, i dati accoppiati valore-chiave di input possono essere suddivisi in tre parti in base ai criteri di età.
Input - Tutti i dati in una raccolta di coppie chiave-valore.
chiave = valore del campo Sesso nel record.
valore = valore dei dati dell'intero record di quel genere.
Method - Il processo della logica di partizione viene eseguito come segue.
- Leggere il valore del campo età dalla coppia chiave-valore di input.
String[] str = value.toString().split("\t");
int age = Integer.parseInt(str[2]);
Verificare il valore dell'età con le seguenti condizioni.
- Età inferiore o uguale a 20 anni
- Età maggiore di 20 e minore o uguale a 30.
- Età superiore a 30 anni.
if(age<=20)
{
return 0;
}
else if(age>20 && age<=30)
{
return 1 % numReduceTasks;
}
else
{
return 2 % numReduceTasks;
}
Output- Tutti i dati delle coppie chiave-valore sono segmentati in tre raccolte di coppie chiave-valore. Il Reducer funziona individualmente su ogni collezione.
Riduci le attività
Il numero di attività del partizionatore è uguale al numero di attività del riduttore. Qui abbiamo tre attività di partizionamento e quindi abbiamo tre attività di Reducer da eseguire.
Input - Il Reducer verrà eseguito tre volte con una raccolta diversa di coppie chiave-valore.
chiave = valore del campo sesso nel record.
valore = tutti i dati del record di quel genere.
Method - La seguente logica verrà applicata a ciascuna raccolta.
- Leggi il valore del campo Stipendio di ogni record.
String [] str = val.toString().split("\t", -3);
Note: str[4] have the salary field value.
Controlla lo stipendio con la variabile max. Se str [4] è il salario massimo, assegna str [4] al massimo, altrimenti salta il passaggio.
if(Integer.parseInt(str[4])>max)
{
max=Integer.parseInt(str[4]);
}
Ripeti i passaggi 1 e 2 per ogni ritiro delle chiavi (maschio e femmina sono le collezioni delle chiavi). Dopo aver eseguito questi tre passaggi, troverai uno stipendio massimo dalla consegna delle chiavi maschile e uno stipendio massimo dalla consegna delle chiavi femminile.
context.write(new Text(key), new IntWritable(max));
Output- Infine, otterrai una serie di dati sulla coppia chiave-valore in tre raccolte di diversi gruppi di età. Contiene rispettivamente lo stipendio massimo della raccolta Maschile e lo stipendio massimo della raccolta Femminile in ciascuna fascia di età.
Dopo aver eseguito le attività Mappa, Partitioner e Riduci, le tre raccolte di dati di coppie chiave-valore vengono archiviate in tre file diversi come output.
Tutte e tre le attività vengono trattate come lavori MapReduce. I seguenti requisiti e specifiche di questi lavori devono essere specificati nelle Configurazioni:
- Nome del lavoro
- Formati di input e output di chiavi e valori
- Classi individuali per attività di mappatura, riduzione e partizionamento
Configuration conf = getConf();
//Create Job
Job job = new Job(conf, "topsal");
job.setJarByClass(PartitionerExample.class);
// File Input and Output paths
FileInputFormat.setInputPaths(job, new Path(arg[0]));
FileOutputFormat.setOutputPath(job,new Path(arg[1]));
//Set Mapper class and Output format for key-value pair.
job.setMapperClass(MapClass.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(Text.class);
//set partitioner statement
job.setPartitionerClass(CaderPartitioner.class);
//Set Reducer class and Input/Output format for key-value pair.
job.setReducerClass(ReduceClass.class);
//Number of Reducer tasks.
job.setNumReduceTasks(3);
//Input and Output format for data
job.setInputFormatClass(TextInputFormat.class);
job.setOutputFormatClass(TextOutputFormat.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
Programma di esempio
Il seguente programma mostra come implementare i partizionatori per i criteri specificati in un programma MapReduce.
package partitionerexample;
import java.io.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.mapreduce.lib.input.*;
import org.apache.hadoop.mapreduce.lib.output.*;
import org.apache.hadoop.util.*;
public class PartitionerExample extends Configured implements Tool
{
//Map class
public static class MapClass extends Mapper<LongWritable,Text,Text,Text>
{
public void map(LongWritable key, Text value, Context context)
{
try{
String[] str = value.toString().split("\t", -3);
String gender=str[3];
context.write(new Text(gender), new Text(value));
}
catch(Exception e)
{
System.out.println(e.getMessage());
}
}
}
//Reducer class
public static class ReduceClass extends Reducer<Text,Text,Text,IntWritable>
{
public int max = -1;
public void reduce(Text key, Iterable <Text> values, Context context) throws IOException, InterruptedException
{
max = -1;
for (Text val : values)
{
String [] str = val.toString().split("\t", -3);
if(Integer.parseInt(str[4])>max)
max=Integer.parseInt(str[4]);
}
context.write(new Text(key), new IntWritable(max));
}
}
//Partitioner class
public static class CaderPartitioner extends
Partitioner < Text, Text >
{
@Override
public int getPartition(Text key, Text value, int numReduceTasks)
{
String[] str = value.toString().split("\t");
int age = Integer.parseInt(str[2]);
if(numReduceTasks == 0)
{
return 0;
}
if(age<=20)
{
return 0;
}
else if(age>20 && age<=30)
{
return 1 % numReduceTasks;
}
else
{
return 2 % numReduceTasks;
}
}
}
@Override
public int run(String[] arg) throws Exception
{
Configuration conf = getConf();
Job job = new Job(conf, "topsal");
job.setJarByClass(PartitionerExample.class);
FileInputFormat.setInputPaths(job, new Path(arg[0]));
FileOutputFormat.setOutputPath(job,new Path(arg[1]));
job.setMapperClass(MapClass.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(Text.class);
//set partitioner statement
job.setPartitionerClass(CaderPartitioner.class);
job.setReducerClass(ReduceClass.class);
job.setNumReduceTasks(3);
job.setInputFormatClass(TextInputFormat.class);
job.setOutputFormatClass(TextOutputFormat.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
System.exit(job.waitForCompletion(true)? 0 : 1);
return 0;
}
public static void main(String ar[]) throws Exception
{
int res = ToolRunner.run(new Configuration(), new PartitionerExample(),ar);
System.exit(0);
}
}
Salva il codice sopra come PartitionerExample.javain "/ home / hadoop / hadoopPartitioner". Di seguito viene fornita la compilazione e l'esecuzione del programma.
Compilazione ed esecuzione
Supponiamo di essere nella directory home dell'utente Hadoop (ad esempio, / home / hadoop).
Seguire i passaggi indicati di seguito per compilare ed eseguire il programma sopra.
Step 1- Scarica Hadoop-core-1.2.1.jar, che viene utilizzato per compilare ed eseguire il programma MapReduce. Puoi scaricare il jar da mvnrepository.com .
Supponiamo che la cartella scaricata sia "/ home / hadoop / hadoopPartitioner"
Step 2 - I seguenti comandi vengono utilizzati per compilare il programma PartitionerExample.java e creare un vaso per il programma.
$ javac -classpath hadoop-core-1.2.1.jar -d ProcessUnits.java
$ jar -cvf PartitionerExample.jar -C .
Step 3 - Utilizzare il comando seguente per creare una directory di input in HDFS.
$HADOOP_HOME/bin/hadoop fs -mkdir input_dir
Step 4 - Utilizzare il seguente comando per copiare il file di input denominato input.txt nella directory di input di HDFS.
$HADOOP_HOME/bin/hadoop fs -put /home/hadoop/hadoopPartitioner/input.txt input_dir
Step 5 - Utilizzare il seguente comando per verificare i file nella directory di input.
$HADOOP_HOME/bin/hadoop fs -ls input_dir/
Step 6 - Utilizzare il seguente comando per eseguire l'applicazione Top salary prendendo i file di input dalla directory di input.
$HADOOP_HOME/bin/hadoop jar PartitionerExample.jar partitionerexample.PartitionerExample input_dir/input.txt output_dir
Attendi qualche istante finché il file non viene eseguito. Dopo l'esecuzione, l'output contiene una serie di suddivisioni di input, attività di mappatura e attività di riduzione.
15/02/04 15:19:51 INFO mapreduce.Job: Job job_1423027269044_0021 completed successfully
15/02/04 15:19:52 INFO mapreduce.Job: Counters: 49
File System Counters
FILE: Number of bytes read=467
FILE: Number of bytes written=426777
FILE: Number of read operations=0
FILE: Number of large read operations=0
FILE: Number of write operations=0
HDFS: Number of bytes read=480
HDFS: Number of bytes written=72
HDFS: Number of read operations=12
HDFS: Number of large read operations=0
HDFS: Number of write operations=6
Job Counters
Launched map tasks=1
Launched reduce tasks=3
Data-local map tasks=1
Total time spent by all maps in occupied slots (ms)=8212
Total time spent by all reduces in occupied slots (ms)=59858
Total time spent by all map tasks (ms)=8212
Total time spent by all reduce tasks (ms)=59858
Total vcore-seconds taken by all map tasks=8212
Total vcore-seconds taken by all reduce tasks=59858
Total megabyte-seconds taken by all map tasks=8409088
Total megabyte-seconds taken by all reduce tasks=61294592
Map-Reduce Framework
Map input records=13
Map output records=13
Map output bytes=423
Map output materialized bytes=467
Input split bytes=119
Combine input records=0
Combine output records=0
Reduce input groups=6
Reduce shuffle bytes=467
Reduce input records=13
Reduce output records=6
Spilled Records=26
Shuffled Maps =3
Failed Shuffles=0
Merged Map outputs=3
GC time elapsed (ms)=224
CPU time spent (ms)=3690
Physical memory (bytes) snapshot=553816064
Virtual memory (bytes) snapshot=3441266688
Total committed heap usage (bytes)=334102528
Shuffle Errors
BAD_ID=0
CONNECTION=0
IO_ERROR=0
WRONG_LENGTH=0
WRONG_MAP=0
WRONG_REDUCE=0
File Input Format Counters
Bytes Read=361
File Output Format Counters
Bytes Written=72
Step 7 - Utilizzare il seguente comando per verificare i file risultanti nella cartella di output.
$HADOOP_HOME/bin/hadoop fs -ls output_dir/
Troverai l'output in tre file perché stai usando tre partitioner e tre Reducers nel tuo programma.
Step 8 - Usa il seguente comando per vedere l'output in Part-00000file. Questo file è generato da HDFS.
$HADOOP_HOME/bin/hadoop fs -cat output_dir/part-00000
Output in Part-00000
Female 15000
Male 40000
Usa il seguente comando per vedere l'output in Part-00001 file.
$HADOOP_HOME/bin/hadoop fs -cat output_dir/part-00001
Output in Part-00001
Female 35000
Male 31000
Usa il seguente comando per vedere l'output in Part-00002 file.
$HADOOP_HOME/bin/hadoop fs -cat output_dir/part-00002
Output in Part-00002
Female 51000
Male 50000