¿Cómo se implementa un Hadoop Mapper en Scala 2.9.0?

Question

Cuando migré a Scala 2.9.0 desde 2.8.1, todo el código fue funcional a excepción de los trazadores Hadoop. Porque tenía algunos objetos envolventes en el camino, me destilada hasta el siguiente ejemplo:

 
import org.apache.hadoop.mapreduce.{Mapper, Job} 


object MyJob { 
    def main(args:Array[String]) { 
    val job = new Job(new Configuration()) 
    job.setMapperClass(classOf[MyMapper]) 

    } 
} 

class MyMapper extends Mapper[LongWritable,Text,Text,Text] { 
    override def map(key: LongWritable, value: Text, context: Mapper[LongWritable,Text,Text,Text]#Context) { 

    } 
} 

Cuando ejecuto esto en 2.8.1, se corre bastante bien (y tengo un montón de código de producción en 2.8.1 . en 2.9.0 me sale el siguiente error de compilación:

error: type mismatch; 
found : java.lang.Class[MyMapper](classOf[MyMapper]) 
required: java.lang.Class[_ <: org.apache.hadoop.mapreduce.Mapper] 
job.setMapperClass(classOf[MyMapper]) 

la llamada no es cuando llamo setMapperClass en el objeto de trabajo Aquí está la definición de ese método:.

public void setMapperClass(java.lang.Class<? extends org.apache.hadoop.mapreduce.Mapper> cls) throws java.lang.IllegalStateException { /* compiled code */ } 

la definición o f La clase Mapper es la siguiente:

public class Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT> 

¿Alguien tiene una idea de lo que estoy haciendo mal? Me parece que el tipo es fundamentalmente correcto: MyMapper amplía Mapper, y el método quiere algo que extienda a Mapper. Y funciona muy bien en 2.8.1 ...

Answer 1

Por tonto que parezca, puede evitar el problema definiendo el asignador antes del trabajo. Las siguientes compilaciones:

import org.apache.hadoop._ 
import org.apache.hadoop.io._ 
import org.apache.hadoop.conf._ 
import org.apache.hadoop.mapreduce._ 

class MyMapper extends Mapper[LongWritable,Text,Text,Text] { 
    override def map(key: LongWritable, value: Text, context: Mapper[LongWritable,Text,Text,Text]#Context) { 
    } 
} 

object MyJob { 
    def main(args:Array[String]) { 
    val job = new Job(new Configuration()) 
    job.setMapperClass(classOf[MyMapper]) 
    } 
}