import java.io.*;
import java.util.*;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapred.*;
import org.apache.hadoop.util.*;

public class MapReduceTripLength {

  /**
   * Calculates the distance in kilometers, input in radians
   */
  private static double sphereDistance(double lat1, double lon1, double lat2, double lon2) {
    double d2r = Math.PI/180; // Conversion factor from degrees to radians
    lat1 *= d2r;
    lon1 *= d2r;
    lat2 *= d2r;
    lon2 *= d2r;
    double R = 6371.009; // kilometers
    double dPhi = lat1 - lat2;
    double dLambda = lon1 - lon2;
    double phi_m = (lat1 + lat2) / 2;
    double D = R * Math.sqrt(Math.pow(dPhi, 2) + Math.pow(Math.cos(phi_m) * dLambda, 2));
    return D;
  }

  public static class Map extends MapReduceBase implements Mapper<LongWritable, Text, NullWritable, Double> {
    public void map(LongWritable key, Text value, OutputCollector<NullWritable, Double> output, Reporter reporter) throws IOException {
      String line = value.toString();
      String[] trip = line.split("\\s+");
      if (trip.length == 7) {
        // Only need the positions but only valid trips count
        try {
          double lat1 = Double.parseDouble(trip[2]); // <start pos (lat)>
          double lon1 = Double.parseDouble(trip[3]); // <end pos (lat)>
          double lat2 = Double.parseDouble(trip[5]); // <start pos (long)>
          double lon2 = Double.parseDouble(trip[6]); // <end pos (long)>

          output.collect(NullWritable.get(), sphereDistance(lat1, lon1, lat2, lon2));
        } catch (Exception e) {
          // Ignore exception on the assumption that the data in the file was
          // invalid on this line (bad practice)
        }
      }
    }
  }

  public static void main(String[] args) throws Exception {
    JobConf conf = new JobConf(MapReduceTripLength.class);
    conf.setJobName("MapReduceTripLength--toonn");
    conf.setOutputKeyClass(NullWritable.class);
    conf.setOutputValueClass(DoubleWritable.class);
    conf.setMapperClass(Map.class);
    conf.setNumReduceTasks(0);
    conf.setInputFormat(TextInputFormat.class);
    conf.setOutputFormat(TextOutputFormat.class);
    FileInputFormat.setInputPaths(conf, new Path(args[0]));
    FileOutputFormat.setOutputPath(conf, new Path(args[1]));
    JobClient.runJob(conf);
  }
}

@@ -161,20 +161,120 @@
 p.mapred
+uce.*;%0Aimport org.apache.hadoop.mapreduce.lib.input.*;%0Aimport org.apache.hadoop.mapreduce.lib.output
 .*;%0A
-
 import o
@@ -933,33 +933,8 @@
 nds 
-MapReduceBase implements 
 Mapp
@@ -1034,71 +1034,23 @@
 ue, 
-OutputCollector%3CNullWritable, Double%3E output, Reporter reporter
+Context context
 ) th
@@ -1563,22 +1563,21 @@
     
-output.collect
+context.write
 (Nul
@@ -1886,23 +1886,29 @@
 n %7B%0A    
-Job
 Conf
+iguration
  conf = 
@@ -1915,16 +1915,62 @@
 new 
-JobConf(
+Configuration();%0A    Job job = Job.getInstance(conf, %22
 MapR
@@ -1988,38 +1988,41 @@
 ngth
-.class
+--toonn%22
 );%0A    
-conf
+job
 .setJ
-obName(%22
+arByClass(
 MapR
@@ -2040,27 +2040,24 @@
 ngth
---toonn%22
+.class
 );%0A    
-conf
+job
 .set
@@ -2088,36 +2088,35 @@
 ble.class);%0A    
-conf
+job
 .setOutputValueC
@@ -2143,28 +2143,27 @@
 class);%0A    
-conf
+job
 .setMapperCl
@@ -2178,28 +2178,27 @@
 class);%0A    
-conf
+job
 .setNumReduc
@@ -2216,99 +2216,57 @@
     
-conf.setInputFormat(TextInputFormat.class);%0A    conf.setOutputFormat(TextOutputFormat.class
+FileInputFormat.setMaxInputSplitSize(job, 1000000
 );%0A 
@@ -2284,19 +2284,19 @@
 tFormat.
-set
+add
 InputPat
@@ -2300,14 +2300,12 @@
 Path
-s(conf
+(job
 , ne
@@ -2357,20 +2357,19 @@
 putPath(
-conf
+job
 , new Pa
@@ -2390,29 +2390,55 @@
     
-JobClient.runJob(conf
+System.exit(job.waitForCompletion(true) ? 0 : 1
 );%0A 

Switch from mapred to mapreduce