대형 데이터 세트를위한 맵리 듀스 코드를 실행하는 동안 Java 힙 공간 오류가

나는 맵리 듀스 프로그래밍 초보자 오전 1 네임 노드, 3 DatanNodes을 포함하는 하둡 클러스터에서 실행을 위해 다음과 같은 자바 프로그램을 코딩했다 :

package trial;

import java.io.IOException;
import java.util.*;
import java.lang.Iterable;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapred.*;
import org.apache.hadoop.util.*;


public class Trial 
{

public static class MapA extends MapReduceBase implements Mapper<LongWritable, Text, Text, Text> 
{

    public void map(LongWritable key, Text value, OutputCollector<Text,Text> output, Reporter reporter) throws IOException
    {
       String[] rows = value.toString().split("\r?\n");          
       for(int i=0;i<rows.length;i++)
       {
           String[] cols = rows[i].toString().split(",");

           String v=cols[0];
           for(int j=1;j<cols.length;j++)
           {
               String k =j+","+cols[j];
               output.collect(new Text(k),new Text(v));
           }
       }


   }
}


public static class ReduceA extends MapReduceBase implements Reducer<Text, Text, Text, Text>
{

        public void reduce(Text key, Iterator<Text> values, OutputCollector<Text, Text>output, Reporter reporter) throws IOException 
        {
            int count =0;                
            String[] attr = key.toString().split(",");      
            List<String> list = new ArrayList<String>();

           while(values.hasNext())               
            {
                list.add((values.next()).toString());
                count++;

            }

           String v=Integer.toString(count);
           for(String s:list)
           { 
               output.collect(new Text(s),new Text(v));
           }

        }   

}




public static void main(String[] args) throws IOException
{
    JobConf conf1 = new JobConf(Trial.class);
    conf1.setJobName("Trial");

    conf1.setOutputKeyClass(Text.class);
    conf1.setOutputValueClass(Text.class);

    conf1.setMapperClass(MapA.class);
    //conf.setCombinerClass(Combine.class);
    conf1.setReducerClass(ReduceA.class);

    conf1.setInputFormat(TextInputFormat.class);
    conf1.setOutputFormat(TextOutputFormat.class);

    FileInputFormat.setInputPaths(conf1, new Path(args[0]));
    FileOutputFormat.setOutputPath(conf1, new Path(args[1]));

    JobClient.runJob(conf1);

    JobConf conf2 = new JobConf(Final.class);
    conf2.setJobName("Final");

    conf2.setOutputKeyClass(Text.class);
    conf2.setOutputValueClass(Text.class);

    conf2.setMapperClass(Final.MapB.class);
    //conf.setCombinerClass(Combine.class);
    conf2.setReducerClass(Final.ReduceB.class);

    conf2.setInputFormat(TextInputFormat.class);
    conf2.setOutputFormat(TextOutputFormat.class);

    FileInputFormat.setInputPaths(conf2, new Path(args[1]));
    FileOutputFormat.setOutputPath(conf2, new Path(args[2]));

    JobClient.runJob(conf2);


  }


  }  

class Final
{

public static class MapB extends MapReduceBase implements Mapper<LongWritable, Text, Text, Text> 
{

    public void map(LongWritable key, Text value, OutputCollector<Text,Text> output, Reporter reporter) throws IOException
    {
       String[] r = value.toString().split("\r?\n");
       String[] p1= new String[5];

       for(int i=0;i<r.length;i++)
       {
           p1 = r[i].split("\t");               
           output.collect(new Text(p1[0]),new Text(p1[1]));
       }

   }
}

 public static class ReduceB extends MapReduceBase implements Reducer<Text, Text, Text, Text>
{

        @Override
        public void reduce(Text key, Iterator<Text> values, OutputCollector<Text, Text>output, Reporter reporter) throws IOException 
        {
           int sum=0;
           while(values.hasNext())
           {
               String s = (values.next()).toString();
               int c=Integer.parseInt(s);
               sum+=c;
           }
           float avf =(float)sum/3;
           String count=Float.toString(avf);
           output.collect(key,new Text(count));
        }   

}

}

이 프로그램은이 같은 데이터 세트에서 실행 :

ID1,1,2,3 
ID1,1,3,2
ID3,2,2,3

각 행은 3 콤마로 분리 특성 다음에 ID를 갖는다. 내 문제는 각 ID의 (아닌 데이터 세트는 2-D 어레이로 간주되면 행에서 열을 따라) 각각의 특성 값의 빈도를 발견 한 다음 ID 각 특성의 주파수를 요약하고 발견하는 데이터 세트 위의 average.Thus :

ID1 : 2+2+2/3=2
ID2 : 2+1+1/3=1.33
ID3 : 1+2+2/3=1.67

위의 코드는 2백~5백메가바이트 같은 작은 데이터 세트와 함께 잘 작동합니다. 그러나 1기가바이트 위의 데이터 세트에 대한이 같은 오류를 얻고있다 :

 map 100% reduce 50%
       14/04/12 12:33:06 INFO mapred.JobClient: Task Id :  attempt_201404121146_0002_r_000001_0, Status : FAILED
      Error: Java heap space
      attempt_201404121146_0002_r_000001_0: Exception in thread  "LeaseRenewer:hdfs@NameNode:8020" java.lang.OutOfMemoryError: Java heap space
      attempt_201404121146_0002_r_000001_0:     at org.apache.hadoop.hdfs.LeaseRenewer.renew(LeaseRenewer.java:397)
     attempt_201404121146_0002_r_000001_0:     at org.apache.hadoop.hdfs.LeaseRenewer.run(LeaseRenewer.java:436)
      attempt_201404121146_0002_r_000001_0:     at org.apache.hadoop.hdfs.LeaseRenewer.access$700(LeaseRenewer.java:70)
     attempt_201404121146_0002_r_000001_0:     at org.apache.hadoop.hdfs.LeaseRenewer$1.run(LeaseRenewer.java:297)
     attempt_201404121146_0002_r_000001_0:     at java.lang.Thread.run(Thread.java:662)
     attempt_201404121146_0002_r_000001_0: Exception in thread "Thread for syncLogs" java.lang.OutOfMemoryError: Java heap space
     attempt_201404121146_0002_r_000001_0:     at java.util.AbstractList.iterator(AbstractList.java:273)
     attempt_201404121146_0002_r_000001_0:     at org.apache.hadoop.mapred.TaskLog.syncLogs(TaskLog.java:363)
     attempt_201404121146_0002_r_000001_0:     at org.apache.hadoop.mapred.Child$3.run(Child.java:158)
     14/04/12 12:33:10 INFO mapred.JobClient:  map 100% reduce 33%
     14/04/12 12:33:12 INFO mapred.JobClient: Task Id :    attempt_201404121146_0002_r_000003_0, Status : FAILED
     Error: Java heap space
      attempt_201404121146_0002_r_000003_0: log4j:WARN No appenders could be found for logger (org.apache.hadoop.mapred.Task).
     attempt_201404121146_0002_r_000003_0: log4j:WARN Please initialize the log4j system properly.
      attempt_201404121146_0002_r_000003_0: log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
     14/04/12 12:33:15 INFO mapred.JobClient:  map 100% reduce 16%
     14/04/12 12:33:16 INFO mapred.JobClient:  map 100% reduce 18%
     14/04/12 12:33:16 INFO mapred.JobClient: Task Id : attempt_201404121146_0002_r_000000_0, Status : FAILED
     Error: Java heap space
      attempt_201404121146_0002_r_000000_0: Exception in thread "LeaseRenewer:hdfs@NameNode:8020" java.lang.OutOfMemoryError: Java heap space
     attempt_201404121146_0002_r_000000_0:     at java.lang.StringCoding.set(StringCoding.java:53)
     attempt_201404121146_0002_r_000000_0:     at java.lang.StringCoding.decode(StringCoding.java:171)
     attempt_201404121146_0002_r_000000_0:     at java.lang.String.<init>(String.java:443)
     attempt_201404121146_0002_r_000000_0:     at java.util.jar.Attributes.read(Attributes.java:401)
      attempt_201404121146_0002_r_000000_0:     at java.util.jar.Manifest.read(Manifest.java:182)
      attempt_201404121146_0002_r_000000_0:     at java.util.jar.Manifest.<init>(Manifest.java:52)
       attempt_201404121146_0002_r_000000_0:     at java.util.jar.JarFile.getManifestFromReference(JarFile.java:167)
       attempt_201404121146_0002_r_000000_0:     at java.util.jar.JarFile.getManifest(JarFile.java:148)
       attempt_201404121146_0002_r_000000_0:     at sun.misc.URLClassPath$JarLoader$2.getManifest(URLClassPath.java:696)
       attempt_201404121146_0002_r_000000_0:     at java.net.URLClassLoader.defineClass(URLClassLoader.java:228)
        attempt_201404121146_0002_r_000000_0:     at java.net.URLClassLoader.access$000(URLClassLoader.java:58)
        attempt_201404121146_0002_r_000000_0:     at java.net.URLClassLoader$1.run(URLClassLoader.java:197)
       attempt_201404121146_0002_r_000000_0:     at      java.security.AccessController.doPrivileged(Native Method)
       attempt_201404121146_0002_r_000000_0:     at java.net.URLClassLoader.findClass(URLClassLoader.java:190)
     attempt_201404121146_0002_r_000000_0:     at java.lang.ClassLoader.loadClass(ClassLoader.java:306)
     attempt_201404121146_0002_r_000000_0:     at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:301)
   attempt_201404121146_0002_r_000000_0:     at java.lang.ClassLoader.loadClass(ClassLoader.java:247)
   attempt_201404121146_0002_r_000000_0:     at org.apache.hadoop.hdfs.LeaseRenewer.renew(LeaseRenewer.java:400)
   attempt_201404121146_0002_r_000000_0:     at org.apache.hadoop.hdfs.LeaseRenewer.run(LeaseRenewer.java:436)
  attempt_201404121146_0002_r_000000_0:     at org.apache.hadoop.hdfs.LeaseRenewer.access$700(LeaseRenewer.java:70)
  attempt_201404121146_0002_r_000000_0:     at org.apache.hadoop.hdfs.LeaseRenewer$1.run(LeaseRenewer.java:297)
  attempt_201404121146_0002_r_000000_0:     at java.lang.Thread.run(Thread.java:662)
 14/04/12 12:33:21 INFO mapred.JobClient:  map 100% reduce 20%

내 프로그램이 너무 많은 메모리를 소비하고 생각하고 최적화 할 필요가있다. 난 1천24메가바이트 내 자바 힙 공간을 증가하지만 여전히 내가 같은 오류가 점점 오전에 의해이 문제를 해결하기 위해 노력했다. 내가 사용했던 데이터 세트 (9 개) 속성이 행 ID를 exclusing로 5cr 행을했다 1.4GB했다. 내 문제는 빅 데이터이기 때문에, 작은 데이터와 코드를 테스트하는 것은 해결책이 아니다. Plz은 당신이 메모리 문제가 해결 될 수 있도록 난 내 코드를 최적화 어떻게 저를 제안 할 수 있습니다. 미리 감사드립니다.

해결법

==============================

1.반복자를 통과 할 수있는 옵션이 두 번 불가능하며 힙이 목록에 저장된 값의 많은 양을 처리 할 수 없기 때문에, 나는 당신이 당신의 작업에 대한 세 가지 맵리 듀스 단계의 총을주고, 중간 맵리 듀스 단계를 추가하는 것이 좋습니다. 다음과 같이 나의 제안은 다음과 같습니다

반복자를 통과 할 수있는 옵션이 두 번 불가능하며 힙이 목록에 저장된 값의 많은 양을 처리 할 수 없기 때문에, 나는 당신이 당신의 작업에 대한 세 가지 맵리 듀스 단계의 총을주고, 중간 맵리 듀스 단계를 추가하는 것이 좋습니다. 다음과 같이 나의 제안은 다음과 같습니다

그것은 당신의 힙이 당신의 HashMap를 처리 할 수 있다고 가정으로이 솔루션은 매우 낙관적이다. 그것을 시도하고 어떻게되는지.

다음은 샘플 코드는 다음과 같습니다

public class Trial {

public static class MapA extends MapReduceBase implements Mapper<LongWritable, Text, Text, Text> 
{

public void map(LongWritable key, Text value, OutputCollector<Text,Text> output, Reporter reporter) throws IOException
{
        String[] rows = value.toString().split("\r?\n");
        for (int i = 0; i < rows.length; i++) {
            String[] cols = rows[i].toString().split(",");

            String v = cols[0];
            for (int j = 1; j < cols.length; j++) {
                String k = j + "," + cols[j];
                output.collect(new Text(k), new Text(v));
            }
        }
}
}


public static class ReduceA extends MapReduceBase implements Reducer<Text, Text, Text, Text>
{

    public void reduce(Text key, Iterator<Text> values,
            OutputCollector<Text, Text> output, Reporter reporter)
            throws IOException {

        int count = 0;

        while (values.hasNext()) {
            output.collect(key, values.next());
            count++;
        }
        output.collect(new Text("." + key),
                new Text(count));
    }  

}


public static class MapB extends MapReduceBase implements Mapper<Text, Text, Text, Text> 
{

public void map(Text key, Text value, OutputCollector<Text, Text> output, Reporter reporter) throws IOException
{
    output.collect(key, value);
}
}


public static class ReduceB extends MapReduceBase implements Reducer<Text, Text, Text, Text>
{

private Map<String, Integer> total_count = new HashMap<String, Integer>();
private Set<String> attributes = new HashSet<String>(); // count the distinct number of attributes

    public void reduce(Text key, Iterator<Text> values,
            OutputCollector<Text, IntWritable> output, Reporter reporter)
            throws IOException {

        String rKey = key.toString();
        if(rKey.startsWith(".")){
            while (values.hasNext()) {
                total_count.put(rKey.substring(1), Integer.valueOf(values.next().toString()));
                attributes.add(rKey.substring(1).split(",")[0]);
                return;
            }
        }
        while (values.hasNext()) {
            Text value = values.next();
            output.collect(value, new Text(Integer.toString(total_count.get(rKey))));
            output.collect(value, new Text("." + attributes.size())); // send the total number of attributes
        }
    }  
}


public static class MapC extends MapReduceBase implements Mapper<Text, Text, Text, Text> 
{

public void map(Text key, Text value, OutputCollector<Text, Text> output, Reporter reporter) throws IOException
{
        output.collect(key, value);
    }
}

public static class ReduceC extends MapReduceBase implements Reducer<Text, Text, Text, DoubleWritable>
{

    @Override
    public void reduce(Text key, Iterator<Text> values, OutputCollector<Text, DoubleWritable>output, Reporter reporter) throws IOException 
    {
       long sum = 0;
       int nbAttributes = 0;
       while(values.hasNext()){
           String value = values.next();
           if(value.startsWith(".")){ // check if line corresponds to the total number of attributes
               nbAttributes = Integer.parseInt(value.substring(1)); 
           } else{
               sum += Integer.parseInt(value);   
           }
       }
       output.collect(key, new DoubleWritable(sum / nbAttributes));
    }   
}

}

from https://stackoverflow.com/questions/23042829/getting-java-heap-space-error-while-running-a-mapreduce-code-for-large-dataset by cc-by-sa and MIT license

'HADOOP' 카테고리의 다른 글

[HADOOP] 돼지 스크립트를 통해 매개 변수를 전달하면 오류 점점 (0)	2019.10.15
[HADOOP] 어떻게 하둡 맵에서 브로 출력이 감소 쓰기? (0)	2019.10.15
[HADOOP] 코드 블록에 대한 magic.wand 기능 사용 (0)	2019.10.15
[HADOOP] 만에 의해 종료 LINES 지금은 '\ n'은 줄 바꿈을 지원합니다 (0)	2019.10.15
[HADOOP] oozie의 sharelib 설치 (0)	2019.10.15

복붙노트

[HADOOP] 대형 데이터 세트를위한 맵리 듀스 코드를 실행하는 동안 Java 힙 공간 오류가

대형 데이터 세트를위한 맵리 듀스 코드를 실행하는 동안 Java 힙 공간 오류가

해결법

'HADOOP' 카테고리의 다른 글

티스토리툴바