如何在Hadoop的Mapper和Reducer中提供子类？

Question

我有一个从超级（父级）类扩展的子（子）类。我想办法提供对Mapper的输入值一般类型，这样我可以提供两个孩子和家长作为有效值是这样的：

公共静态类MyMapper扩展映射< ...，MyParentClass，...，...>

我希望从MyParentClass扩展的MyChildClass也是有效的。

然而，当我运行程序，如果该值是一个子类我得到一个例外：从地图值

类型不匹配：预计MyParentClass，收到MyChildClass

如何启用输入/输出值到映射器的值是否为有效的值？

更新：

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类，从它延伸。当从RecordReader返回ChildFloatImage时，它抛出以前的异常。

Answer 1

的解决方案，我也跟着，是创建一个委托所有的容器/包装类所需要的功能的原始对象，如下所示：

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); 
    } 

} 

而在制图员：

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

在这种情况下，无论是FloatImage和ChildFloatImage可以封装FloatImageContainer和你摆脱Hadoop中的inheretance问题，因为只有一个类直接使用FloatImageContainer它不是任何父/子的。

Answer 2

背景

这样做的原因是类型擦除使得它不可能for 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));