如何在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));