java - Mapreduce链接作业因异常而失败

Question

我对hadoop非常陌生，在运行mapreduce作业时遇到此错误。我正在尝试计算一个人的平均水平，并尝试接受第一份工作的输入并传递给第二份工作来计算成绩。我了解问题所在，但无法弄清楚我在哪里做错了。

以下是异常(exception):

   15/07/02 23:53:36 WARN mapred.JobClient: Use GenericOptionsParser for parsing the arguments. Applications should implement Tool for the same.
    15/07/02 23:53:36 INFO input.FileInputFormat: Total input paths to process : 1
    15/07/02 23:53:38 INFO mapred.JobClient: Running job: job_201507022153_0026
    15/07/02 23:53:39 INFO mapred.JobClient:  map 0% reduce 0%
    15/07/02 23:53:44 INFO mapred.JobClient: Task Id : attempt_201507022153_0026_m_000000_0, Status : FAILED
    java.lang.ClassCastException: org.apache.hadoop.io.Text cannot be cast to org.apache.hadoop.io.DoubleWritable
        at com.hadoop.mrchain.Driver$Mapper2.map(Driver.java:1)
        at org.apache.hadoop.mapreduce.Mapper.run(Mapper.java:144)
        at org.apache.hadoop.mapred.MapTask.runNewMapper(MapTask.java:647)
        at org.apache.hadoop.mapred.MapTask.run(MapTask.java:323)
        at org.apache.hadoop.mapred.Child$4.run(Child.java:266)
        at java.security.AccessController.doPrivileged(Native Method)
        at javax.security.auth.Subject.doAs(Subject.java:396)
        at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1278)
        at org.apache.hadoop.mapred.Child.main(Child.java:260)

我的代码:

package com.hadoop.mrchain;

import java.io.IOException;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class Driver {
    /*
     * Mapper1
     */
    public static class Mapper1 extends
            Mapper<Object, Text, Text, DoubleWritable> {
        public void map(Object key, Text value, Context context)
                throws IOException, InterruptedException {
            StringTokenizer itr = new StringTokenizer(value.toString());
            String studentName = itr.nextToken();
            Double marks = Double.parseDouble(itr.nextToken());
            context.write(new Text(studentName), new DoubleWritable(marks));
        }
    }

    /*
     * Mapper1
     */
    public static class Mapper2 extends
            Mapper<Object, DoubleWritable, Text, DoubleWritable> {
        public void map(Object key, DoubleWritable value, Context context)
                throws IOException, InterruptedException {
            StringTokenizer itr = new StringTokenizer(value.toString());
            context.write(new Text(itr.nextToken()), new DoubleWritable(Double
                    .parseDouble(itr.nextToken().toString())));
        }
    }

    /*
     * Reducer1
     */
    public static class Reducer1 extends
            Reducer<Text, DoubleWritable, Text, DoubleWritable> {
        public void reduce(Text key, Iterable<DoubleWritable> values,
                Context context) throws IOException, InterruptedException {
            int sum = 0;
            int count = 0;
            for (DoubleWritable val : values) {
                sum += val.get();
                count++;
            }
            double avg = sum / count;
            context.write(key, new DoubleWritable(avg));

        }
    }

    /*
     * Reducer2
     */
    public static class Reducer2 extends
            Reducer<Text, DoubleWritable, Text, Text> {
        public void reduce(Text key, Iterable<DoubleWritable> values,
                Context context) throws IOException, InterruptedException {
            for (DoubleWritable val : values) {
                // double marks = Double.parseDouble(val.toString());
                int marks = ((Double) val.get()).intValue();
                if (marks >= 70) {
                    context.write(key, new Text("GradeA"));
                } else if (marks >= 60 && marks < 70) {
                    context.write(key, new Text("GradeB"));
                } else if (marks < 60 && marks >= 40) {
                    context.write(key, new Text("GradeC"));
                } else {
                    context.write(key, new Text("FAIL"));
                }
            }
        }
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        cleanFileSystem(conf, args);

        Job job1 = new Job(conf, "BATCH51-MRCHAIN-JOB1");
        job1.setJarByClass(Driver.class);

        job1.setMapperClass(Mapper1.class);
        job1.setCombinerClass(Reducer1.class);
        job1.setReducerClass(Reducer1.class);
        job1.setOutputKeyClass(Text.class);
        job1.setOutputValueClass(DoubleWritable.class);

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

        job1.waitForCompletion(true);

        // Job2
        Job job2 = new Job(conf, "BATCH51-MRCHAIN-JOB2");
        job2.setJarByClass(Driver.class);

        job2.setMapperClass(Mapper2.class);
        job2.setCombinerClass(Reducer2.class);
        job2.setReducerClass(Reducer2.class);
        // job2.setMapOutputKeyClass(Text.class);
        // job2.setMapOutputValueClass(DoubleWritable.class);
        job2.setOutputKeyClass(Text.class);
        job2.setOutputValueClass(Text.class);

        FileInputFormat.addInputPath(job2, new Path(args[1]));
        FileOutputFormat.setOutputPath(job2, new Path(args[2]));

        System.exit(job2.waitForCompletion(true) ? 0 : 1);
    }

    private static void cleanFileSystem(Configuration conf, String[] args)
            throws Exception {
        FileSystem fs = FileSystem.get(conf);
        if (fs.exists(new Path(args[1]))) {
            fs.delete(new Path(args[1]), true);
        }
        if (fs.exists(new Path(args[2]))) {
            fs.delete(new Path(args[2]), true);
        }
        // if (fs.exists(new Path(args[3]))) {
        // fs.delete(new Path(args[3]), true);
        // }
    }
}

输入样例:

hello 90
suresh 80
krishna 16
ramesh 55
santosh 82
anji 66
gopal 88
hello99
suresh 80
krishna 16
gopal 91
hello 91
suresh 80
krishna 86
ramesh 55
santosh 82
anji 66
gopal 95

Answer 1

它不能将几个字符串转换为 double ，例如，hello无法转换为 double 。您需要在映射器中更改逻辑以解决此问题。