java - hadoop inputFile作为BufferedImage

Question

我对Hadoop开发有疑问。

我必须在使用Hadoop的简单图像处理项目上训练自己。
我要做的就是用Hadoop旋转图像(当然，我不希望Hadoop使用整个图像)。
我的inputFormat有问题。
我该怎么办？

我发现的解决方案是:

将SequenceFile用作inputFormat(但是，我必须先做一个map / reduce作业，然后再做一个旋转)？

用我开发的类ImageWritable扩展了FileInputFormat和RecordReader。它包含一个BufferedImage(原始图像)，一个bufferedImage [](它是我们想要分割的任意数量的图像)，以及一个int [] pixelArray，它是图像的“序列化”。

我也不太熟悉Java中的泛型，因此我扩展了此类:

public class ImageInputFormat extends FileInputFormat< IntWritable, ImageWritable>
public class ImageRecordReader extends RecordReader< IntWritable, ImageWritable> 

这是正确的吗？

实际上，我迷失在那个丛林中，甚至不知道接下来该做什么/使用。我阅读了很多有关Hadoop的文章，但是我仍然不太了解map / reduce如何通过输入文件来工作。
请问有人在这里帮助我吗？

Answer 1

我一直在寻求帮助两个星期，但是今天下午，我发现了一个有趣的页面:
http://wiki.apache.org/hadoop/HadoopMapReduce?highlight=(inputf)

因此，我探讨了第二点。现在，我可以毫无问题地到达recordReader了。
我看了课Text代码源。真的帮了我
这是ImageRecordReader的一部分

final class ImageRecordReader extends RecordReader<IntWritable, ImageWritable>{

    ImageWritable iwri;
    IntWritable key;


    public ImageRecordReader(TaskAttemptContext context, FileSplit fileSplit) throws IOException, InterruptedException {
        System.out.println(fileSplit.getPath().toString());
        key = new IntWritable(0);
        iwri = new ImageWritable(4);
    }

    @Override
    public void initialize(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException {
        FileSplit fsplit = (FileSplit) split;
        iwri.CreateImageWritableFromSerial(fsplit.getPath());
        iwri.affiche();
    }

这是使BufferedImage变为可写的主要代码:

public final class ImageWritable implements Writable{

    private BufferedImage bi;
    private int[] pixelsArray;
    private BufferedImage[] biPart;
    private int nbPart;

    public int[] getPixelsArray(){return pixelsArray;}

    private void setPixelArray(){
        int width = bi.getWidth();
        int height = bi.getHeight();
        pixelsArray = new int[width * height +2];
        Image im = bi;
        PixelGrabber pg = new PixelGrabber(im, 0, 0, width, height, pixelsArray, 2, width);
        try{ pg.grabPixels(); }
        catch(InterruptedException ex){ ex.printStackTrace(); }
        pixelsArray[0] = width;
        pixelsArray[1] = height;
    }
    public int[] getPixelArray(Image im){
        int width = im.getWidth(null);
        int height = im.getHeight(null);

        pixelsArray = new int[width * height +2];

        PixelGrabber pg = new PixelGrabber(im, 0, 0, width, height, pixelsArray, 2, width);
        try{ pg.grabPixels(); }
        catch(InterruptedException ex) { ex.printStackTrace(); }

        pixelsArray[0] = width;
        pixelsArray[1] = height;
        System.out.println("Width = "+ width);
        System.out.println("Heitgh = "+ height);
        return pixelsArray;
    }

    private void createPartfromPixelArray(int[] pixArr){
        MemoryImageSource mis;
        Toolkit tk = Toolkit.getDefaultToolkit();
        int wPart = pixArr[0]/(nbPart/2);
        int hPart = pixArr[1]/2;
        int lgLi = pixArr[0];
        for(int i = 1; i <= nbPart; ++i){
           if(i<=nbPart/2){
               mis = new MemoryImageSource(wPart, hPart,  pixArr, 2+ i*wPart, lgLi);
            } else {
               mis = new MemoryImageSource(wPart, hPart, pixArr, (pixArr.length+2)/2 + (i%(nbPart/2))*wPart,  lgLi);
            }

           biPart[i-1] = RotationToolKit.getBuffered(tk.createImage(mis));
           affiche(biPart[i-1], Integer.toString(i));
        }

    }

    public ImageWritable(int nbPart){
        this.nbPart = nbPart;
        bi = new BufferedImage(1, 1, BufferedImage.TYPE_INT_ARGB);
        biPart = new BufferedImage[this.nbPart];
    }
    public boolean CreateImageWritableFromSerial(Path path){
        try {
            System.out.println(path.toString());
            DataInput dataInput = new FileImageInputStream(new File(path.toString().substring(5)));
            readFields(dataInput);
            return true;
        } catch (FileNotFoundException ex) {
            Logger.getLogger(ImageWritable.class.getName()).log(Level.SEVERE, null, ex);
        } catch (IOException ex) {
            Logger.getLogger(ImageWritable.class.getName()).log(Level.SEVERE, null, ex);
        }
            return false;
    }


    public void write(DataOutput d) throws IOException {
        System.out.println("Width du tableau = "+ pixelsArray[0]);
        System.out.println("Heitgh du tableau = "+ pixelsArray[1]);
        System.out.println("length du tableau = "+ pixelsArray.length);
        for(int o : pixelsArray){
            d.writeInt(o);
        }
        System.out.println();
    }

    public void readFields(DataInput di) throws IOException {
        int w = di.readInt();
        int h = di.readInt();
        int length = w * h;
        System.out.println("Width lue du tableau = "+ w);
        System.out.println("Heitgh lue du tableau = "+ h);
        System.out.println("length calculée du tableau = "+ length);
        pixelsArray = new int[length+2];
        pixelsArray[0] = w;
        pixelsArray[1] = h;
        for(int i = 2; i<pixelsArray.length; i++)
            pixelsArray[i] = di.readInt();

        System.out.println();

        createPartfromPixelArray(pixelsArray);
        bi = createfromPixelsArray(pixelsArray);
    }

    public ImageWritable getPart(int i){ return new ImageWritable(biPart[i]); }
    public BufferedImage getBi() { return bi; }
    public void setWithoutCreateNewBi(BufferedImage bi){ this.bi = bi; }
    public void setBi(BufferedImage bi) {
        Graphics2D g = bi.createGraphics();
        g.setComposite(AlphaComposite.DstIn);
        g.drawImage(this.bi, null, null);
        g.dispose();
    }

    public int getNbPart() {return nbPart; }
    public void setNbPart(int part) {  nbPart = part; }

}

剩下的问题是:
*我真的不知道它是否仍然正确
*如何充分分割图像以使hdfs块大小小的图像？

感谢那些可以帮助我的人。