问使用Java查找字符串的MapReduce

EN

Error: java.lang.ClassCastException: org.apache.hadoop.io.LongWritable cannot be cast to org.apache.hadoop.io.Text
        at searchaString.SearchDriver$searchMap.map(SearchDriver.java:1)
        at org.apache.hadoop.mapreduce.Mapper.run(Mapper.java:146)
        at org.apache.hadoop.mapred.MapTask.runNewMapper(MapTask.java:787)
        at org.apache.hadoop.mapred.MapTask.run(MapTask.java:341)
        at org.apache.hadoop.mapred.YarnChild$2.run(YarnChild.java:168)
        at java.security.AccessController.doPrivileged(Native Method)
        at javax.security.auth.Subject.doAs(Subject.java:415)
        at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1657)
        at org.apache.hadoop.mapred.YarnChild.main(YarnChild.java:162)

package samples.wordcount;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
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;
//import org.apache.hadoop.util.GenericOptionsParser;
//import org.apache.hadoop.mapred.lib.NLineInputFormat;
import java.io.IOException;
import java.util.Iterator;


public class WordCount {

    public static void main(String[] args) throws Exception {

        @SuppressWarnings("unused")
        JobClient jobC =new JobClient();

        Configuration conf = new Configuration();
        //String args[] = parser.getRemainingArgs();

        Job job = Job.getInstance(conf);
        job.setJobName("WordCount");


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

        job.setJarByClass(WordCount.class);

        job.setMapperClass(TokenizerMapper.class);
        job.setReducerClass(IntSumReducer.class);

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

        //job.setInputFormatClass(TextInputFormat.class);

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

        /*String MyWord = args[2];
        TokenizerMapper.find = MyWord;*/

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

    public static class TokenizerMapper extends Mapper<Text, Text, Text, IntWritable> {

        private final static IntWritable one = new IntWritable(1);
        //  private Text word = new Text();
        static String find="txt was not created";
        public int i;

        public void map(Text key, Text value,OutputCollector<Text, IntWritable> output,Reporter reporter) throws IOException, InterruptedException
        {
            String cleanLine = value.toString();        

            String[] cleanL =cleanLine.split("home");

            output.collect(new Text(cleanL[1]), one);

        }
    }

    public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> {



        public void reduce(Text key, Iterator<IntWritable> values, OutputCollector<Text, IntWritable> output,Reporter reporter)
                throws IOException, InterruptedException {

            int sum = 0;

            String wordText="txt was not created";

            while(values.hasNext()) {

                Boolean check = values.toString().contains("txt was not created");

                if(check)
                {
                    String[] cleanL =values.toString().split("\\.");

                    for(String w : cleanL)
                    {
                        if(w.length()>=wordText.length())

                        {
                            String wrd = w.substring(0,wordText.length()); 

                            if(wrd.equals(wordText))
                            {
                                IntWritable value=values.next();
                                sum += value.get();

                            }

                        }
                    }
                }
            }
            output.collect(key,new IntWritable(sum));
        }
    }
}