[HADOOP] 날짜 별 Java MapReduce 계산

날짜 별 Java MapReduce 계산

저는 Hadoop을 처음 사용하고 있습니다. 그리고 MapReduce 프로그램을 수행하려고합니다. 날짜별로 처음 두 개의 출현자를 계산합니다 (월별로 그룹화). 그래서 제 입력은 다음과 같습니다.

2017-06-01 , A, B, A, C, B, E, F 
2017-06-02 , Q, B, Q, F, K, E, F
2017-06-03 , A, B, A, R, T, E, E 
2017-07-01 , A, B, A, C, B, E, F
2017-07-05 , A, B, A, G, B, G, G

그래서,이 MapReduce 프로그램의 결과로 기대하고 있습니다.

2017-06,  A:4, E:4
2017-07,  A:4, B:4

공용 클래스 ArrayGiulioTest {     공공 정적 로거 로거 = Logger.getLogger (ArrayGiulioTest.class);     public static class CustomMap는 Mapper를 확장합니다. {         개인 텍스트 단어 = 새 텍스트 ();         public void map (LongWritable key, 텍스트 값, 컨텍스트 컨텍스트)에 의해 IOException, InterruptedException {             TextWritable 배열 = 새 TextWritable ();             String line = value.toString ();             StringTokenizer tokenizer = new StringTokenizer (line, ",");             문자열 dataAttuale = tokenizer.nextToken (). 하위 문자열 (0,                     line.lastIndexOf ( "-"));             텍스트 tmp = null;             텍스트 [] tmpArray = 새 텍스트 [tokenizer.countTokens ()];             int i = 0;             while (tokenizer.hasMoreTokens ()) {                 String prod = tokenizer.nextToken ( ",");                 word.set (dataAttuale);                 tmp = 새로운 텍스트 (prod);                 tmpArray [i] = tmp;                 나는 ++;             }             array.set (tmpArray);             context.write (word, array);         }     }     public static class CustomReduce는 Reducer 를 확장합니다. {         public void reduce (텍스트 키, Iterator 값,                 컨텍스트 컨텍스트)가 IOException을 Throw하면 InterruptedException {             MapWritable map = 새 MapWritable ();             텍스트 txt = 새 텍스트 ();             while (values.hasNext ()) {                 TextWritable array = values.next ();                 텍스트 [] tmpArray = (텍스트 []) array.toArray ();                 for (텍스트 t : tmpArray) {                     if (map.get (t)! = null) {                         IntWritable val = (IntWritable) map.get (t);                         map.put (t, new IntWritable (val.get () + 1));                     } else {                         map.put (t, new IntWritable (1));                     }                 }             }             Set set = map.keySet ();             StringBuffer str = new StringBuffer ();             for (쓰기 가능 k : 집합) {                 str.append ( "key :"+ k.toString () + "value :"+ map.get (k) + "**");             }             txt.set (str.toString ());             context.write (key, txt);         }     }     public static void main (String [] args) 예외를 throw합니다.         long inizio = System.currentTimeMillis ();         구성 conf = 새 구성 ();         작업 작업 = Job.getInstance (conf, "countProduct");         job.setJarByClass (ArrayGiulioTest.class);         job.setMapperClass (CustomMap.class);         //job.setCombinerClass(CustomReduce.class);         job.setReducerClass (CustomReduce.class);         job.setMapOutputKeyClass (Text.class);         job.setMapOutputValueClass (TextWritable.class);         job.setOutputKeyClass (Text.class);         job.setOutputValueClass (Text.class);         FileInputFormat.addInputPath (job, newPath (args [0]));         FileOutputFormat.setOutputPath (job, 새 경로 (args [1]));         job.waitForCompletion (true);         long fine = System.currentTimeMillis ();         logger.info ( "*************************** End"+ (End-Start ));         System.exit (1);     } } 그리고 난 내 사용자 지정 TextWritable 이런 식으로 구현했습니다 : 공용 클래스 TextWritable extends ArrayWritable {     public TextWritable () {         슈퍼 (Text.class);     } } .. 내 MapReduce 프로그램을 실행할 때 이런 종류의 결과를 얻습니다. 2017-6 wordcount.TextWritable@3e960865 2017-6 wordcount.TextWritable@3e960865 제 감속가가 효과가 없다는 것은 명백합니다. 내 맵퍼의 출력 같아. 어떤 생각? 솔루션에 대한 올바른 길은 누군가가 말할 수 있습니다. 여기 콘솔 로그 (단지 정보를 위해, 나의 입력 파일은 5 대신에 6 개의 행을 가지고있다) * Eclipse에서 MapReduce 문제 (mono JVM)를 시작하거나 Hadoop과 Hadoop을 사용하여 동일한 결과를 얻는다. 파일 시스템 카운터     FILE : 읽은 바이트 수 = 1216     FILE : 기록 된 바이트 수 = 431465     FILE : 읽기 연산 수 = 0     FILE : 큰 읽기 연산의 수 = 0     FILE : 쓰기 조작 수 = 0 지도 축소 프레임 워크     지도 입력 레코드 = 6     지도 출력 레코드 = 6     지도 출력 바이트 = 214     맵 출력 materialized bytes = 232     입력 분할 바이트 = 97     입력 레코드 결합 = 0     출력 레코드 결합 = 0     입력 그룹 줄이기 = 3     셔플 바이트 감소 = 232     입력 레코드 줄이기 = 6     출력 레코드 감소 = 6     유출 기록 = 12     섞인지도 = 1     실패한 셔플 = 0     병합 된지도 출력 = 1     GC 경과 시간 (ms) = 0     총 커밋 된 힙 사용 (바이트) = 394264576 셔플 오류     BAD_ID = 0     연결 = 0     IO_ERROR = 0     WRONG_LENGTH = 0     WRONG_MAP = 0     WRONG_REDUCE = 0 파일 입력 형식 카운터     읽은 바이트 수 = 208 파일 출력 형식 카운터     기록 된 바이트 = 1813

해결법

1. ==============================

   1.나는 당신이 매퍼에서 너무 많은 일을하려한다고 생각합니다. 날짜를 그룹화하기 만하면됩니다 (예상 출력에 따라 서식을 제대로 지정하지 않은 것 같습니다).

   나는 당신이 매퍼에서 너무 많은 일을하려한다고 생각합니다. 날짜를 그룹화하기 만하면됩니다 (예상 출력에 따라 서식을 제대로 지정하지 않은 것 같습니다).

   다음과 같은 접근 방식은이 선을 돌릴 것입니다. 예를 들어

   2017-07-01 , A, B, A, C, B, E, F
   2017-07-05 , A, B, A, G, B, G, G
   

   이 감속기 쌍으로

   2017-07 , ("A,B,A,C,B,E,F", "A,B,A,G,B,G,G")
   

   즉, ArrayWritable을 사용하여 텍스트로 유지하면 실제 이익을 얻을 수 없습니다.

   그래서 매퍼는 다음과 같이 보일 것입니다.

   class CustomMap extends Mapper<LongWritable, Text, Text, Text> {
   
       private final Text key = new Text();
       private final Text output = new Text();
   
       @Override
       protected void map(LongWritable offset, Text value, Context context) throws IOException, InterruptedException {
   
           int separatorIndex = value.find(",");
   
           final String valueStr = value.toString();
           if (separatorIndex < 0) {
               System.err.printf("mapper: not enough records for %s", valueStr);
               return;
           }
           String dateKey = valueStr.substring(0, separatorIndex).trim();
           String tokens = valueStr.substring(1 + separatorIndex).trim().replaceAll("\\p{Space}", "");
   
           SimpleDateFormat fmtFrom = new SimpleDateFormat("yyyy-MM-dd");
           SimpleDateFormat fmtTo = new SimpleDateFormat("yyyy-MM");
   
           try {
               dateKey = fmtTo.format(fmtFrom.parse(dateKey));
               key.set(dateKey);
           } catch (ParseException ex) {
               System.err.printf("mapper: invalid key format %s", dateKey);
               return;
           }
   
           output.set(tokens);
           context.write(key, output);
       }
   }
   

   감속기는 값 문자열에서 값을 수집하고 계산하는 Map을 작성할 수 있습니다. 다시 말하지만, 텍스트 만 쓰십시오.

   class CustomReduce extends Reducer<Text, Text, Text, Text> {
   
       private final Text output = new Text();
   
       @Override
       protected void reduce(Text date, Iterable<Text> values, Context context) throws IOException, InterruptedException {
   
           Map<String, Integer> keyMap = new TreeMap<>();
           for (Text v : values) {
               String[] keys = v.toString().trim().split(",");
   
               for (String key : keys) {
                   if (!keyMap.containsKey(key)) {
                       keyMap.put(key, 0);
                   }
                   keyMap.put(key, 1 + keyMap.get(key));
               }
           }
   
           output.set(mapToString(keyMap));
           context.write(date, output);
       }
   
       private String mapToString(Map<String, Integer> map) {
           StringBuilder sb = new StringBuilder();
           String delimiter = ", ";
           for (Map.Entry<String, Integer> entry : map.entrySet()) {
               sb.append(
                       String.format("%s:%d", entry.getKey(), entry.getValue())
               ).append(delimiter);
           }
           sb.setLength(sb.length()-delimiter.length());
           return sb.toString();
       }
   }
   

   당신의 의견을 감안할 때, 저는 이것을 얻었습니다.

   2017-06 A:4, B:4, C:1, E:4, F:3, K:1, Q:2, R:1, T:1
   2017-07 A:4, B:4, C:1, E:1, F:1, G:3
   

2. ==============================

   2.주된 문제는 reduce 메소드의 부호에 관한 것입니다.

   주된 문제는 reduce 메소드의 부호에 관한 것입니다.

   나는 쓰고 있었다 : public void reduce (텍스트 키, Iterator 값,                 컨텍스트 컨텍스트)

   대신에

       public void reduce(Text key, Iterable<ArrayTextWritable> values,
   

   이것이 Reduce 출력 대신 Map 출력을 얻는 이유입니다

from https://stackoverflow.com/questions/44399163/java-mapreduce-counting-by-date by cc-by-sa and MIT license