Hadoop MapReduce reducer未启动

Question

public class HoursJob {
    public static void main(String[] args) throws Exception {
        if (args.length != 2) {
          System.err.println("Usage: HoursJob <input path> <output path>");
          System.exit(-1);
        }

        Job job = Job.getInstance();
        job.setJarByClass(HoursJob.class);
        job.setJobName("Hours job");

        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        job.setMapperClass(HoursMapper.class);
        job.setReducerClass(HoursReducer.class);

        job.setMapOutputKeyClass(IntWritable.class);
        job.setMapOutputValueClass(CellWithTotalAmount.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.class);

        int ret = job.waitForCompletion(true) ? 0 : 1;
        System.exit(ret);
    }
}

public class HoursMapper 
        extends Mapper<LongWritable, Text, IntWritable, CellWithTotalAmount> {
    static double BEGIN_LONG = -74.913585;
    static double BEGIN_LAT = 41.474937;
    static double GRID_LENGTH = 0.011972;
    static double GRID_HEIGHT = 0.008983112;

    @Override
    public void map(LongWritable key, Text value, Mapper.Context context)
            throws IOException, InterruptedException {

        System.out.println("Hello from mapper.");
        String recordString = value.toString();
        try {
            DEBSFullRecord record = new DEBSFullRecord(recordString);
            Date pickupDate = record.getPickup();
            Calendar calendar = GregorianCalendar.getInstance();
            calendar.setTime(pickupDate);
            int pickupHour = calendar.get(Calendar.HOUR_OF_DAY);
            int cellX = (int)
                ((record.getPickupLongitude() - BEGIN_LONG) / GRID_LENGTH) + 1;
            int cellY = (int)
                ((BEGIN_LAT - record.getPickupLatitude()) / GRID_HEIGHT) + 1;

            CellWithTotalAmount hourInfo = 
                new CellWithTotalAmount(cellX, cellY, record.getTotal());
            context.write(new IntWritable(pickupHour), hourInfo);
        } catch (Exception ex) {
            System.out.println(
                "Cannot parse: " + recordString + "due to the " + ex);
        }
    }
}

减速器

public class HoursReducer 
        extends Reducer<IntWritable, CellWithTotalAmount, Text, NullWritable> {
    @Override
    public void reduce(IntWritable key, Iterable<CellWithTotalAmount> values, 
            Context context) throws IOException, InterruptedException {
        System.out.println("Hello from reducer.");
        int[][] cellRideCounters = getCellRideCounters(values);
        CellWithRideCount cellWithMostRides = 
            getCellWithMostRides(cellRideCounters);

        int[][] cellTotals = getCellTotals(values);
        CellWithTotalAmount cellWithGreatestTotal = 
            getCellWithGreatestTotal(cellTotals);

        String output = key + " "
            + cellWithMostRides.toString() + " "
            + cellWithGreatestTotal.toString();

        context.write(new Text(output), NullWritable.get());
    }

    //omitted for brevity
}

自定义可写类

public class CellWithTotalAmount implements Writable {
    public int cellX;
    public int cellY;
    public double totalAmount;

    public CellWithTotalAmount(int cellX, int cellY, double totalAmount) {
        this.cellX = cellX;
        this.cellY = cellY;
        this.totalAmount = totalAmount;
    }

    @Override
    public void write(DataOutput out) throws IOException {
        System.out.println("Writing CellWithTotalAmount");
        out.writeInt(cellX);
        out.writeInt(cellY);
        out.writeDouble(totalAmount);
    }

    @Override
    public void readFields(DataInput in) throws IOException {
        System.out.println("Reading CellWithTotalAmount");
        cellX = in.readInt();
        cellY = in.readInt();
        totalAmount = in.readDouble();
    }

    @Override
    public String toString() {
        return cellX + " " + cellY + " " + totalAmount;
    }
}

Answer 1

我认为在减少功能方面有很多例外，所以框架不能很好地完成工作

    public class HoursReducer 
            extends Reducer<IntWritable, CellWithTotalAmount, Text, NullWritable> {
        @Override
        public void reduce(IntWritable key, Iterable<CellWithTotalAmount> values, 
                Context context) throws IOException, InterruptedException {
            System.out.println("Hello from reducer.");
    try{
            int[][] cellRideCounters = getCellRideCounters(values);
       if(cellRideCounter[0].length>0){ // control it before executing it. more explanation is above
            CellWithRideCount cellWithMostRides = 
                getCellWithMostRides(cellRideCounters);



            int[][] cellTotals = getCellTotals(values);
            CellWithTotalAmount cellWithGreatestTotal = 
                getCellWithGreatestTotal(cellTotals);

            String output = key + " "
                + cellWithMostRides.toString() + " "
                + cellWithGreatestTotal.toString();

            context.write(new Text(output), NullWritable.get());


     }
   }catch(Exception e)

    e.printstack();
     return;
   {

  }


}

• 在reduce函数中添加try-catch以获取异常
• 。从catch中的函数返回

.还要在调用getCellWithMostRiders（..）之前添加if语句我想问题就在这里。按你想的填写if语句我做了一个猜测，然后按我的猜测填写它，如果它不适合你，你想怎么改就怎么改