Trouver le moyen de nombres à l'aide de MapReduce

Utiliser tous les 3 Mapper/Combiner/Réducteur pour résoudre le problème.
Consultez ci-dessous le lien pour remplir le code & explication

http://alchemistviews.blogspot.com/2013/08/calculate-average-in-map-reduce-using.html

OriginalL'auteur Pankaj Khattar

La moyenne arithmétique est une fonction d'agrégation qui n'est pas distributive mais algébrique. Selon Han et coll. une fonction d'agrégation est distributive si:

[...] il peut être calculée [...] comme suit. Supposons que [..] les données sont partitionnées en n ensembles. Nous appliquons la fonction de chaque partition, résultant en n valeurs agrégées. Si le résultat obtenu en appliquant la fonction de la n valeurs d'agrégation est le même que celui obtenu en appliquant la fonction de l'ensemble des données (sans partitionnement), la fonction peut être calculée de manière distribuée.

Ou, en d'autres termes, il doit être associative et commutative. Une fonction d'agrégation est cependant algébrique selon Han et coll. si:

[...] elle peut être calculée par la algébrique de la fonction avec le m arguments (où m est bornée entier positif), dont chacune est obtenue par l'application d'une distribution fonction d'agrégation.

Pour la moyenne arithmétique, c'est juste moyenne = somme/comte. Évidemment, vous devez effectuer un nombre de plus. Mais à l'aide d'un compteur global à cet effet semble être un abus. Le API décrit org.apache.hadoop.mapreduce.Counter comme suit:

Un nommé compteur qui suit la progression d'un travail map/reduce.

Compteurs devraient être généralement utilisés pour les statistiques sur les emplois de toute façon, mais pas dans les calculs lors du traitement des données lui-même.

Donc tout ce que vous avez à faire à l'intérieur d'une partition est à ajouter vos numéros en place et le suivi de leur nombre et la somme (sum, count); une approche simple peut être une chaîne de caractères comme <sum><separator><count>.

Dans le mapper le comte sera toujours 1 et la somme est la valeur brute elle-même. Afin de réduire les fichiers de carte déjà vous pouvez utiliser le viseur et traiter les agrégats comme (sum_1 + ... + sum_n, count_1 + ... + count_n). Ceci doit être répété dans le réducteur et fini par le calcul final somme/comte. Gardez à l'esprit que cette approche est indépendante de la clé utilisée!

Enfin, voici un exemple simple à l'aide de la crue statistiques de la criminalité de la LAPD qui devrait calcule la "moyenne du temps de la criminalité" à Los Angeles:

public class Driver extends Configured implements Tool {
enum Counters {
DISCARDED_ENTRY
}
public static void main(String[] args) throws Exception {
ToolRunner.run(new Driver(), args);
}
public int run(String[] args) throws Exception {
Configuration configuration = getConf();
Job job = Job.getInstance(configuration);
job.setJarByClass(Driver.class);
job.setMapperClass(Mapper.class);
job.setMapOutputKeyClass(LongWritable.class);
job.setMapOutputValueClass(Text.class);
job.setCombinerClass(Combiner.class);
job.setReducerClass(Reducer.class);
job.setOutputKeyClass(LongWritable.class);
job.setOutputValueClass(Text.class);
FileInputFormat.addInputPath(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
return job.waitForCompletion(true) ? 0 : -1;
}
}
public class Mapper extends org.apache.hadoop.mapreduce.Mapper<
LongWritable,
Text,
LongWritable,
Text
> {
@Override
protected void map(
LongWritable key,
Text value,
org.apache.hadoop.mapreduce.Mapper<
LongWritable,
Text,
LongWritable,
Text
>.Context context
) throws IOException, InterruptedException {
//parse the CSV line
ArrayList<String> values = this.parse(value.toString());
//validate the parsed values
if (this.isValid(values)) {
//fetch the third and the fourth column
String time = values.get(3);
String year = values.get(2)
.substring(values.get(2).length() - 4);
//convert time to minutes (e.g. 1542 -> 942)
int minutes = Integer.parseInt(time.substring(0, 2))
* 60 + Integer.parseInt(time.substring(2,4));
//create the aggregate atom (a/n)
//with a = time in minutes and n = 1
context.write(
new LongWritable(Integer.parseInt(year)),
new Text(Integer.toString(minutes) + ":1")
);
} else {
//invalid line format, so we increment a counter
context.getCounter(Driver.Counters.DISCARDED_ENTRY)
.increment(1);
}
}
protected boolean isValid(ArrayList<String> values) {
return values.size() > 3 
&& values.get(2).length() == 10 
&& values.get(3).length() == 4;
}
protected ArrayList<String> parse(String line) {
ArrayList<String> values = new ArrayList<>();
String current = "";
boolean escaping = false;
for (int i = 0; i < line.length(); i++){
char c = line.charAt(i);
if (c == '"') {
escaping = !escaping;
} else if (c == ',' && !escaping) {
values.add(current);
current = "";
} else {
current += c;
}
}
values.add(current);
return values;
}
}
public class Combiner extends org.apache.hadoop.mapreduce.Reducer<
LongWritable,
Text,
LongWritable,
Text
> {
@Override
protected void reduce(
LongWritable key,
Iterable<Text> values,
Context context
) throws IOException, InterruptedException {
Long n = 0l;
Long a = 0l;
Iterator<Text> iterator = values.iterator();
//calculate intermediate aggregates
while (iterator.hasNext()) {
String[] atom = iterator.next().toString().split(":");
a += Long.parseLong(atom[0]);
n += Long.parseLong(atom[1]);
}
context.write(key, new Text(Long.toString(a) + ":" + Long.toString(n)));
}
}
public class Reducer extends org.apache.hadoop.mapreduce.Reducer<
LongWritable,
Text,
LongWritable,
Text
> {
@Override
protected void reduce(
LongWritable key, 
Iterable<Text> values, 
Context context
) throws IOException, InterruptedException {
Long n = 0l;
Long a = 0l;
Iterator<Text> iterator = values.iterator();
//calculate the finale aggregate
while (iterator.hasNext()) {
String[] atom = iterator.next().toString().split(":");
a += Long.parseLong(atom[0]);
n += Long.parseLong(atom[1]);
}
//cut of seconds
int average = Math.round(a / n);
//convert the average minutes back to time
context.write(
key,
new Text(
Integer.toString(average / 60) 
+ ":" + Integer.toString(average % 60)
)
);
}
}

OriginalL'auteur witrin