java hadoop inputFile作为缓冲区映像

Question

我有一个关于Hadoop开发的问题

我必须训练自己使用Hadoop进行一个简单的图像处理项目。 我只想用Hadoop旋转一个图像（当然我不希望Hadoop使用整个图像）。 我的输入格式有问题。 我该怎么办

我找到的解决办法是：

• 将一个SequenceFile作为inputFormat（但是，我必须先做一个map/reduce作业，然后再做一个来进行旋转？）
• 使用I开发的类ImageWriteable扩展FileInputFormat和RecordReader。它包含一个BuffereImage（原始图像）、一个BuffereImage[]和一个int[]像素数组，前者是图像按我们想要的任何数字分割的图像，后者是图像的“序列化”

我也不太熟悉java中的泛型，因此我扩展了如下类：

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

对吗

事实上，我在丛林中迷失了方向，甚至不知道下一步该做什么/用什么。我读了很多关于Hadoop的文章，但我仍然不太明白map/reduce是如何使用inputfile的。 请问这里有人帮我吗

Answer 1

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

因此，我探讨了第二点。现在，我可以无误地访问recordReader。 我查看了类文本源代码。这对我真的很有帮助。 这是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();
    }

下面是使BuffereImage变得可写的主要代码：

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块大小的小图像

谢谢能帮助我的人