I got a piece of Java code using Hadoop to calculate min, max, average and variance on a large dataset made of (index value) couples separated by a newline:
0 11839923.64831265
1 5710431.90800272
It's compiled locally and run on a remote distributed HDFS instance by a sh script.
My main concerns are:
if the output is collected in a different order, the thing just stops working, and instead of returning one result for each key, it prints the same key over and over, filling the terminal
it creates a new
Text
instance every time, which looks really inefficient, but when I tired to use a single shared constant, it stopped working. Maybe using anEnum
would do fine, but I don't feel like changing it until I fixed the previous point.it's using a
Scanner
inside the mapper to treat multi-line inputs properly, but only single-line input is showing up. Does Hadoop guarantee each mapper only receives one-line inputs, or is the remote setup that makes it so?it uses the "old" approach of extending the
MapReduceBase
class and implementing theMapper
/Reducer
interface. I've read that, with the new 2.0 APIs it's sufficient to extend oneMapper
orReducer
class. Yet, I can't find any migration doc with simple migration doc, and the official WordCount example tutorial is stuck at r1.2.1. EDIT: found a reference for this. And another one here.
Here's the code:
package org.myorg;
import java.io.IOException;
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.io.BufferedWriter;
import java.io.OutputStreamWriter;
import java.util.*;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapred.*;
import org.apache.hadoop.util.*;
public class calcAll {
public static class Map extends MapReduceBase implements Mapper<LongWritable, Text, Text, DoubleWritable> {
public void map(LongWritable key, Text value, OutputCollector<Text, DoubleWritable> output, Reporter reporter) throws IOException {
// this will work even if we receive more than 1 line
Scanner scanner = new Scanner(value.toString());
String line;
String[] tokens;
double observation;
while (scanner.hasNext()) {
line = scanner.nextLine();
tokens = line.split("\\s+");
observation = Double.parseDouble(tokens[1]);
output.collect(new Text("values"), new DoubleWritable(observation));
}
}
}
public static class Combine extends MapReduceBase implements Reducer<Text, DoubleWritable, Text, DoubleWritable> {
public void reduce(Text key, Iterator<DoubleWritable> values, OutputCollector<Text, DoubleWritable> output, Reporter reporter) throws IOException {
double count = 0d; // should be an int, but anyway...
double min = Double.MAX_VALUE;
double max = Double.MIN_VALUE;
double sum = 0d;
double sumSquared = 0d;
double value;
while (values.hasNext()) {
++count;
value = values.next().get();
min = Math.min(min, value);
max = Math.max(max, value);
sum += value;
sumSquared += value * value;
}
// keep in alphabetical order or KABOOM!
output.collect(new Text("count"), new DoubleWritable(count));
output.collect(new Text("max"), new DoubleWritable(max));
output.collect(new Text("min"), new DoubleWritable(min));
output.collect(new Text("sum"), new DoubleWritable(sum));
output.collect(new Text("sumSquared"), new DoubleWritable(sumSquared));
}
}
public static class Reduce extends MapReduceBase implements Reducer<Text, DoubleWritable, Text, DoubleWritable> {
public void reduce(Text key, Iterator<DoubleWritable> values, OutputCollector<Text, DoubleWritable> output, Reporter reporter) throws IOException {
if (key.equals(new Text("count"))) {
double count = 0d;
double value;
while (values.hasNext()) {
value = values.next().get();
count += value;
}
output.collect(new Text("count"), new DoubleWritable(count));
}
if (key.equals(new Text("max"))) {
double max = Double.MIN_VALUE;
double value;
while (values.hasNext()) {
value = values.next().get();
max = Math.max(max, value);
}
output.collect(new Text("max"), new DoubleWritable(max));
}
if (key.equals(new Text("min"))) {
double min = Double.MAX_VALUE;
double value;
while (values.hasNext()) {
value = values.next().get();
min = Math.min(min, value);
}
output.collect(new Text("min"), new DoubleWritable(min));
}
if (key.equals(new Text("sum"))) {
double sum = 0d;
double value;
while (values.hasNext()) {
value = values.next().get();
sum += value;
}
output.collect(new Text("sum"), new DoubleWritable(sum));
}
if (key.equals(new Text("sumSquared"))) {
double sumSquared = 0d;
double value;
while (values.hasNext()) {
value = values.next().get();
sumSquared += value;
}
output.collect(new Text("sumSquared"), new DoubleWritable(sumSquared));
}
}
}
public static boolean applySecondPass(Path in, Path out) {
double count = 0d, max = 0d, min = 0d, sum = 0d, sumSquared = 0d;
try (FileSystem fs = FileSystem.get(new Configuration());
BufferedReader br = new BufferedReader(new InputStreamReader(fs.open(in)));) {
String line;
String[] words;
line = br.readLine();
while (line != null) {
words = line.split("\\s+");
switch (words[0]) {
case "count":
count = Double.parseDouble(words[1]);
break;
case "max":
max = Double.parseDouble(words[1]);
break;
case "min":
min = Double.parseDouble(words[1]);
break;
case "sum":
sum = Double.parseDouble(words[1]);
break;
case "sumSquared":
sumSquared = Double.parseDouble(words[1]);
break;
}
line = br.readLine();
}
} catch (Exception e) {
e.printStackTrace();
return false;
}
double avg = sum / count;
// (Sum_sqr - (Sum*Sum)/n)/(n - 1)
double variance = (sumSquared - (sum * sum) / count) / (count - 1);
try (FileSystem fs = FileSystem.get(new Configuration());
BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(fs.create(out, true)));) {
String line;
line = "avg\t" + String.valueOf(avg) + System.lineSeparator();
bw.write(line);
line = "variance\t" + String.valueOf(variance) + System.lineSeparator();
bw.write(line);
} catch (Exception e) {
e.printStackTrace();
return false;
}
return true;
}
public static void main(String[] args) throws Exception {
JobConf conf = new JobConf(calcAll.class);
conf.setJobName("calcAll");
conf.setOutputKeyClass(Text.class);
conf.setOutputValueClass(DoubleWritable.class);
conf.setMapperClass(Map.class);
conf.setCombinerClass(Combine.class);
conf.setReducerClass(Reduce.class);
conf.setInputFormat(TextInputFormat.class);
conf.setOutputFormat(TextOutputFormat.class);
FileInputFormat.setInputPaths(conf, new Path(args[0]));
Path out1 = new Path(args[1]);
FileOutputFormat.setOutputPath(conf, out1);
JobClient.runJob(conf); // blocking call
// the output is a set of files, merge them before continuing
Path out1Merged = new Path(args[2]);
Configuration config = new Configuration();
try {
FileSystem hdfs = FileSystem.get(config);
FileUtil.copyMerge(hdfs, out1, hdfs, out1Merged, false, config, null);
} catch (IOException e) {
e.printStackTrace();
System.exit(1);
}
// calculate on job output
boolean success = applySecondPass(out1Merged, new Path(args[3]));
System.out.println("Second pass successful? " + success);
System.exit(success ? 1 : 0);
}
}