Как обеспечить подкласс в Mapper и Reducer Hadoop?

Question

У меня есть дочерний (дочерний) класс, который распространяется от класса super (parent). Я хочу способ обеспечить общий тип для ввода значения Mapper, так что я могу обеспечить как ребенка и родителя, как допустимые значения, как это:

общественный статический класс MyMapper расширяет Mapper < ..., MyParentClass, ..., ...>

Я хочу, чтобы MyChildClass, который простирается от MyParentClass, также действителен.

Однако, когда я бег программы, если значение является дочерним классом я получаю исключение:

типа несоответствия значения от карты: ожидается MyParentClass, получил MyChildClass

Как я могу включить как дочерний, так и родительский классы должны быть действительными ввод/вывод значение в/из картографа?

Update:

package hipi.examples.dumphib; 

import hipi.image.FloatImage; 
import hipi.image.ImageHeader; 
import hipi.imagebundle.mapreduce.ImageBundleInputFormat; 
import hipi.util.ByteUtils; 

import org.apache.hadoop.conf.Configuration; 
import org.apache.hadoop.conf.Configured; 
import org.apache.hadoop.fs.FileSystem; 
import org.apache.hadoop.fs.Path; 
import org.apache.hadoop.io.IntWritable; 
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; 
import org.apache.hadoop.util.Tool; 
import org.apache.hadoop.util.ToolRunner; 

import java.io.IOException; 
import java.util.Iterator; 

public class DumpHib extends Configured implements Tool { 

    public static class DumpHibMapper extends Mapper<ImageHeader, FloatImage, IntWritable, Text> { 

    @Override 
    public void map(ImageHeader key, FloatImage value, Context context) throws IOException, InterruptedException { 

     int imageWidth = value.getWidth(); 
     int imageHeight = value.getHeight(); 

     String outputStr = null; 

     if (key == null) { 
    outputStr = "Failed to read image header."; 
     } else if (value == null) { 
    outputStr = "Failed to decode image data."; 
     } else { 
    String camera = key.getEXIFInformation("Model"); 
    String hexHash = ByteUtils.asHex(ByteUtils.FloatArraytoByteArray(value.getData())); 
    outputStr = imageWidth + "x" + imageHeight + "\t(" + hexHash + ")\t " + camera; 
     } 

     context.write(new IntWritable(1), new Text(outputStr)); 
    } 

    } 

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

    @Override 
    public void reduce(IntWritable key, Iterable<Text> values, Context context) throws IOException, InterruptedException { 
     for (Text value : values) { 
    context.write(key, value); 
     } 
    } 

    } 

    public int run(String[] args) throws Exception { 

    if (args.length < 2) { 
     System.out.println("Usage: dumphib <input HIB> <output directory>"); 
     System.exit(0); 
    } 

    Configuration conf = new Configuration(); 

    Job job = Job.getInstance(conf, "dumphib"); 

    job.setJarByClass(DumpHib.class); 
    job.setMapperClass(DumpHibMapper.class); 
    job.setReducerClass(DumpHibReducer.class); 

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

    String inputPath = args[0]; 
    String outputPath = args[1]; 

    removeDir(outputPath, conf); 

    FileInputFormat.setInputPaths(job, new Path(inputPath)); 
    FileOutputFormat.setOutputPath(job, new Path(outputPath)); 

    job.setNumReduceTasks(1); 

    return job.waitForCompletion(true) ? 0 : 1; 

    } 

    private static void removeDir(String path, Configuration conf) throws IOException { 
    Path output_path = new Path(path); 
    FileSystem fs = FileSystem.get(conf); 
    if (fs.exists(output_path)) { 
     fs.delete(output_path, true); 
    } 
    } 

    public static void main(String[] args) throws Exception { 
    int res = ToolRunner.run(new DumpHib(), args); 
    System.exit(res); 
    } 

} 

FloatImage супер класс и у меня есть ChildFloatImage класс, который наследуется от него. Когда ChildFloatImage возвращается с RecordReader, он бросает предыдущее исключение.

Answer 1

Решение, я последовал , заключается в создании класса container/wrapper, который делегирует все необходимые функции на оригинальный объект следующим образом:

public class FloatImageContainer implements Writable, RawComparator<BinaryComparable> { 

    private FloatImage floatImage; 

    public FloatImage getFloatImage() { 
     return floatImage; 
    } 

    public void setFloatImage(FloatImage floatImage) { 
     this.floatImage = floatImage; 
    } 

    public FloatImageContainer() { 
     this.floatImage = new FloatImage(); 
    } 

    public FloatImageContainer(FloatImage floatImage) { 
     this.floatImage = floatImage; 
    } 

    @Override 
    public int compare(BinaryComparable o1, BinaryComparable o2) { 
     // TODO Auto-generated method stub 
     return floatImage.compare(o1, o2); 
    } 

    @Override 
    public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) { 
     // TODO Auto-generated method stub 
     return floatImage.compare(b1, s1, l1, b2, s2, l2); 
    } 

    @Override 
    public void write(DataOutput out) throws IOException { 
     // TODO Auto-generated method stub 
     floatImage.write(out); 
    } 

    @Override 
    public void readFields(DataInput in) throws IOException { 
     // TODO Auto-generated method stub 
     floatImage.readFields(in); 
    } 

} 

И в Mapper:

public static class MyMapper extends Mapper<..., FloatImageContainer, ..., ...> { 

В этом случае как FloatImage и ChildFloatImage может быть воплощен в FloatImageContainer и вы избавиться от проблем с перемещением в Hadoop, потому что есть только один класс, используемый напрямую FloatImageContainer, который не является родителем/дочерним.

Answer 2

фон

Причина этого заключается в том, что тип стирание делает невозможным для Java, чтобы (во время выполнения) убедитесь, что ваш MyMapper фактически расширяет правильный тип (в терминах параметров универсального типа на Mapper).

Java в основном составляет:

List<String> list = new ArrayList<String>(); 
list.add("Hi"); 
String x = list.get(0); 

в

List list = new ArrayList(); 
list.add("Hi"); 
String x = (String) list.get(0); 

Кредиты для этого примера идти here.

Таким образом, вы вводите MyMapper где Java хочет видеть Mapper<A, B, C, D> специфического A, B, C и D - не возможно во время выполнения. Поэтому мы вынуждены проверять время компиляции.

Решение

Вы можете сделать следующее для всех пользовательских подклассов:

job.setMapperClass(DumpHibMapper.class); 

использованием java.lang.Class#asSubclass

и делать это вместо:

job.setMapperClass(DumpHibMapper.class.asSubclass(Mapper.class));