Инкрементное добавление файла, созданного с помощью RDD

Question

В Java У меня есть двумерный массив, представляющий пиксели в изображении. Из-за размера массива (9744 × 9744) Я не могу сохранить весь массив в одном RDD. Я решил обработать половину одной строки изображения за раз, а затем вывести ее в файл с помощью saveAsTextFile(). Когда я это делаю, я получаю Exception in thread "main" org.apache.hadoop.mapred.FileAlreadyExistsException после обработки первой строки.

Есть ли способ постепенного добавления в файл, созданный первой половиной RDD первой строки? Ниже приведен пример того, что я пытаюсь сделать.

int pixs[][] = new int[4872][2]; 
    int count = 0; 
    int rowI = 0; 
    int colJ = 0; 
    int colJCurrent = 0; 
    JavaRDD<int[]> firstHalf = null; 
    sparkPixels = new double[ 1 ][ 9744 ][3]; 

    for(; rowI < 1; rowI++) 
    {  
     for(int colCount=0;colCount < 2;colCount++) 
     { 
      for(colJ=colJCurrent; colJ < (colJCurrent+(rawWidth)); colJ++) 
      { 
       pixs[count][0]=rowI; 
       pixs[count][1]=colJ; 
       count++; 
      } 
      colJCurrent=colJ; 
      count=0; 

       firstHalf = ctx.parallelize(Arrays.asList(pixs)); 
       JavaRDD<SimpleMatrix> firstResults = firstHalf.map(new Function<int[], SimpleMatrix>() { 
        private static final long serialVersionUID = 1L; 

        public SimpleMatrix call(int pix[]) 
        { 
         return PixelInfoFunction(pix[0], pix[1]); 
        } 
       }); 

       JavaRDD<String> stringOutput = firstResults.map(new Function<SimpleMatrix, String>() { 
       private static final long serialVersionUID = 1L; 

       public String call(SimpleMatrix i) 
       { 
        return PixelInfoInStringFormatt; 
       } 
      }); 

      stringOutput.saveAsTextFile("/home/bielasjj/Projection_Output/test"); 
     } 
     colJ=0; 
     count = 0; 
     colJCurrent=0; 

    } 

    ctx.stop(); 
    ctx.close(); 

Я изменил код для обработки одной строки за один раз, и добавить, что в массив для последующего вывода, но во время выполнения я получаю Exception in thread "main" java.lang.OutOfMemoryError: GC overhead limit exceeded

for(int rowI = 0 ; rowI < 9744; rowI++) 
    { 
     for(int colJ = 0 ; colJ < 9744; colJ++) 
     { 
      pixs[colJ][0]=rowI; 
      pixs[colJ][1]=colJ; 
     } 

     JavaRDD<int[]> firstHalf = ctx.parallelize(Arrays.asList(pixs)); 
     JavaRDD<SimpleMatrix> rowResults = firstHalf.map(new Function<int[], SimpleMatrix>() { 
      private static final long serialVersionUID = 1L; 

      public SimpleMatrix call(int pix[]) 
      { 
       return PixelInfoFunction(pix[0], pix[1]); 
      } 
     }); 

     rowResults.foreach(new VoidFunction<SimpleMatrix>(){ 
      private static final long serialVersionUID = 1L; 

      public void call(SimpleMatrix i) 
      { 
       sparkPixels[(int)i.get(3,0)][(int)i.get(4,0)][0] = i.get(0, 0); 
       sparkPixels[(int)i.get(3,0)][(int)i.get(4,0)][1] = i.get(1, 0); 
       sparkPixels[(int)i.get(3,0)][(int)i.get(4,0)][2] = i.get(2, 0); 
      } 
     }); 

    } 
    PrintWriter csv = new PrintWriter("/home/CSV.csv"); 
    for(int i=0; i < (2 * rawWidth)-1; i++) 
    { 
     for(int j=0; j < (3 * rawHeight)-1; j++) 
     { 
      String line = i+ "," + j + ", " + (int)sparkPixels[i][j][0] +", "+(int)sparkPixels[i][j][1]+", "+(int)sparkPixels[i][j][2]; 
      csv.println(line); 
     } 
    } 
    csv.close(); 

Answer 1

Добавить номер строки в имя выходного файла:

stringOutput.saveAsTextFile("/home/bielasjj/Projection_Output/test" + rowI);