MapReduce 작업에서 감속 작업이 호출되지 않습니다.

Question

단어 수를 줄일 수있는 작업입니다. 내 자신의 InputFormat 있습니다.

JobExecutor :

val job = new Job(new Configuration()) 

job.setMapperClass(classOf[CountMapper]) 
job.setReducerClass(classOf[CountReducer]) 

job.setJobName("tarun-test-1") 
job.setInputFormatClass(classOf[MyInputFormat]) 
FileInputFormat.setInputPaths(job, new Path(args(0))) 
FileOutputFormat.setOutputPath(job, new Path(args(1))) 

job.setOutputKeyClass(classOf[Text]) 
job.setOutputValueClass(classOf[LongWritable]) 

job.setNumReduceTasks(1) 

println("status: " + job.waitForCompletion(true)) 

매퍼 :

class CountMapper extends Mapper[LongWritable, Text, Text, LongWritable] { 

    private val valueOut = new LongWritable(1L) 

    override def map(k: LongWritable, v: Text, context: Mapper[LongWritable, Text, Text, LongWritable]#Context): Unit = { 
     val str = v.toString 
     str.split(",").foreach(word => { 
      val keyOut = new Text(word.toLowerCase.trim) 
      context.write(keyOut, valueOut) 
     }) 
    } 
} 

감속기 :

class CountReducer extends Reducer[Text, LongWritable, Text, LongWritable] { 

    override def reduce(k: Text, values: Iterable[LongWritable], context: Reducer[Text, LongWritable, Text, LongWritable]#Context): Unit = { 
     println("Inside reduce method..") 
     val valItr = values.iterator() 
     var sum = 0L 
     while (valItr.hasNext) { 
      sum = sum + valItr.next().get() 
     } 

     context.write(k, new LongWritable(sum)) 
     println("done reducing.") 
    } 
} 

매퍼가 호출되고 있으며 RecordReader 제대로 로그를 기반으로 분할을 읽고 있습니다. 그러나 감속기가 호출되지 않습니다.

Answer 1

설정을 시도하십시오 : job.mapOutputKeyClass 및 job.MapOutputValueClass.