메뉴 건너뛰기

검색창 닫기

Cloudera, BigData, Semantic IoT, Hadoop, NoSQL

Cloudera CDH/CDP 및 Hadoop EcoSystem, Semantic IoT등의 개발/운영 기술을 정리합니다. gooper@gooper.com로 문의 주세요.


mapreduce 2개 data를 join하고 마지막으로 code정보를 join하여 결과를 얻는 mr 프로그램

총관리자 2014.06.30 20:42 조회 수 : 466

1. data준비

(각 File1, File2는 hdfs상에 존재해야한다,

 즉, hadoop fs -put UserDetails.txt  DeliveryDetails.txt /data1/hadoop/mr/in을 실행한다)

File 1 – UserDetails.txt(mobile#, 사용자이름)
123 456, Jim
456 123, Tom
789 123, Harry
789 456, Richa
 
File 2 – DeliveryDetails.txt(mobile#, 상태코드)
123 456, 001
456 123, 002
789 123, 003
789 456, 004
 
File 3 – DeliveryStatusCodes.txt(상태코드, 상태코드명)
001, Delivered
002, Pending
003, Failed
004, Resend
 
* File3은 os 파일시스템에 존재해야한다.
(예, /home/hadoop/hadoop/working/DeliveryStatusCodes.txt)
2. 예상되는 최종 결과 포맷
Jim, Delivered
Tom, Pending
Harry, Failed
Richa, Resend
 
3. UserFileMapper.java
package com.gooper.join;
import java.io.IOException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
public class UserFileMapper extends MapReduceBase implements Mapper<LongWritable, Text, Text, Text> {
 private String cellNumber, customerName, fileTag="CD~";
 public void map(LongWritable key, Text value, OutputCollector<Text, Text> output, Reporter reporter) throws IOException {
  String line = value.toString();
  
  String splitarray[] = line.split(",");
  cellNumber = splitarray[0].trim();
  customerName = splitarray[1].trim();
  // reducer에서의 구분을 위해서 value값 앞에 "CD~"를 붙여준다.
  output.collect(new Text(cellNumber),  new Text(fileTag+customerName));
  }
 }
 
4. DeliverFileMapper.java
package com.gooper.join;
import java.io.IOException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
public class DeliverFileMapper extends MapReduceBase implements Mapper<LongWritable, Text, Text, Text> {
private String cellNumber, deliveryCode, fileTag="DR~";
public void map(LongWritable key, Text value, OutputCollector<Text, Text> output, Reporter reporter) throws IOException {
 String line = value.toString();

 String splitarray[] = line.split(",");
 cellNumber = splitarray[0].trim();
 deliveryCode = splitarray[1].trim();
 // reducer에서의 구분을 위해서 value값 앞에 "DR~"를 붙여준다.
 output.collect(new Text(cellNumber),  new Text(fileTag+deliveryCode));
 }
}
 
5. SmsDriver.java
package com.gooper.join;
import java.io.BufferedReader;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.StringTokenizer;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
public class SmsReducer extends MapReduceBase implements Reducer<Text, Text, Text, Text> {
 private String customerName, deliveryReport;
 private static Map<String,String> DeliveryCodesMap = new HashMap<String,String>();
 
 public void configure(JobConf job) {
  loadDeliveryStatusCodes();
 }
 
 public void reduce(Text key, Iterator<Text> values, OutputCollector<Text,Text> output, Reporter reporter) throws IOException {
  while(values.hasNext()) {
   String currValue = values.next().toString();
   String valueSplitted[] = currValue.split("~");
   String keyValue = key.toString();
   
   System.out.println("키값 , 라인값 : "+keyValue+","+currValue);
   //System.out.println("reporter값 : "+reporter.toString());
   
   // CD로 시작되면 고객명을 그대로 사용하고..
   if(valueSplitted[0].equals("CD")) {
    customerName = valueSplitted[1].trim();
   // DR로 시작되면 code값이므로 code에 대한 값을 찾아서 그 명칭을 출력한다.
   } else if(valueSplitted[0].equals("DR")) {
    deliveryReport = DeliveryCodesMap.get(valueSplitted[1].trim());
   }
  }
  if(customerName != null && deliveryReport != null) {
   output.collect(new Text(customerName+"("+key+")"), new Text(deliveryReport));
  } else if(deliveryReport == null) { // codeㅇ mapping되는 값이 없으면 "deliveryReport"라는 문자열 자체를 출력한다.   
   output.collect(new Text(customerName+"("+key+")"),  new Text("deliveryReport"));
  }
 }
 
 private void loadDeliveryStatusCodes() {
  String strRead;
  try {
                       
    //BufferedReader reader = new BufferedReader(new FileReader("/data1/hadoop/mr/in/DeliveryStatusCodes.txt"));
       // 여기는 로컬 os파일 위치를 지정해야 함..
    BufferedReader reader = new BufferedReader(new FileReader("/home/hadoop/hadoop/working/DeliveryStatusCodes.txt"));
    
    while((strRead = reader.readLine() ) != null) {
     String splitarray[] = strRead.split(",");
     DeliveryCodesMap.put(splitarray[0].trim(), splitarray[1].trim()) ;
     
    }
  } catch (FileNotFoundException e) {
   e.printStackTrace();
  } catch(IOException e) {
   e.printStackTrace();
  }
 }
}

6. 설명
가. SmsDriver.java에서
MultipleInputs.addInputPath(conf, new Path(args[0]), TextInputFormat.class, UserFileMapper.class);
MultipleInputs.addInputPath(conf, new Path(args[1]), TextInputFormat.class, DeliverFileMapper.class);
을 이용하여 각각의 파일을 읽어 들인다.
이때 읽어들이는 처리는 UserFileMapper와 DeliverFileMapper가 각각 맡는다
이때 join key는 mobile phone no가 된다.
 
나. 두개의 input파일을 reduce에게 보내면 reducer에서는 key를 기준으로 value가 각각의 값을 가지고 있으므로
code값인 경우는 HashMap에 저장된 code정보를 mapping하여 code에 대한 명칭을 output하게 만든다.
최종적으로는 이름과 상태값을 가지는 결과 파일이 만들어진다.
 
다. Reducer에슨 각각의 구분이 필요하므로 Mapper에서 각각의 값에 "CD~" 혹은 "DR~"을 붙여 구분할 수 있도록 선처리를 한다
 
라. 파일을 build하고 jar로 만들고 다음과 같이 실행한다.
hadoop@bigdata-host:~/hadoop/working$ hadoop jar gooper-hadoopexamples.jar com.gooper.join.SmsDriver /data1/hadoop/mr/in/UserDetails.txt /data1/hadoop/mr/in/DeliveryDetails.txt /data1/hadoop/mr/out/join/d
마. 실행 결과 확인
hadoop@bigdata-host:~/hadoop/working$ hadoop fs -cat /data1/hadoop/mr/out/join/d/part-00000
Jim(123 456) Delivered
Tom(456 123) Pending
Harry(789 123) Failed
Richa(789 456) Resend
 
바. reducer에 인입되는 key와 value를 확인하면 아래와 같다.
키값 , 라인값 : 123 456,CD~Jim
키값 , 라인값 : 123 456,DR~001
키값 , 라인값 : 456 123,DR~002
키값 , 라인값 : 456 123,CD~Tom
키값 , 라인값 : 789 123,CD~Harry
키값 , 라인값 : 789 123,DR~003
키값 , 라인값 : 789 456,DR~004
키값 , 라인값 : 789 456,CD~Richa

이 게시물을
이 글의 추천인 목록 목록
번호 제목 날짜 조회 수
81 [Solr in Cloudera]Solr Data Directory변경 방법/절차 2023.04.21 169
80 파일끝에 붙는 ^M 일괄 지우기(linux, unix(AIX)) 혹은 파일내에 있는 ^M지우기 2016.09.24 168
79 다중 모듈 프로젝트 설정에 대한 설명 2016.09.21 168
78 [hive] hive.tbls테이블의 owner컬럼값은 hadoop.security.auth_to_local에 의해서 filtering된다. 2022.04.14 167
77 collection생성시 -shards와 -replicationFactor값을 잘못지정하면 write.lock for client xxx.xxx.xxx.xxx already exists오류가 발생한다. 2016.04.28 167
76 org.apache.hadoop.hbase.ClockOutOfSyncException: org.apache.hadoop.hbase.ClockOutOfSyncException 오류시 조치사항 2016.07.14 165
75 spark 온라인 책자링크 (제목 : mastering-apache-spark) 2016.05.25 165
74 [Impala jdbc]CDP7.1.7환경에서 java프로그램을 이용하여 kerberized impala cluster에 접근하여 SQL을 수행하는 방법 2023.08.22 164
73 oozie webui접근시 id/pw를 물어보는 Windows보안 팝업창이 뜰때 확인/조치방법 2022.05.02 164
72 test333444 2017.05.01 162
71 시맨틱 관련 논문 모음 사이트 2017.06.13 161
70 hadoop에서 yarn jar ..를 이용하여 appliction을 실행하여 정상적(?)으로 수행되었으나 yarn UI의 어플리케이션 목록에 나타나지 않는 문제 2017.05.02 161
69 you are accessing a non-optimized hue please switch to one of the available addresses 2021.10.06 160
68 [KUDU] kudu tablet server여러가지 원인에 의해서 corrupted상태가 된 경우 복구방법 2023.03.28 159
67 [SBT] project.sbt에 libraryDependencies에 필요한 jar를 지정했으나 sbt compile할때 클래스를 못찾는 오류가 발생했을때 조치사항 2016.08.03 159
66 전체 컨택스트 내용 file 2017.12.19 158
65 hadoop에서 yarn jar ..를 이용하여 appliction을 실행하여 정상적으로 수행되었으나 yarn UI의 어플리케이션 목록에 나타나지 않는 문제 2017.05.02 157
64 [Hue admin]Add/Sync LDAP user, Sync LDAP users/groups 버튼 기능 설명 2023.08.09 155
63 [oozie]oozie ssh action으로 패스워드 없이 다른 서버에 ssh로그인 하여 shellscript호출하는 설정하는 방법 2022.11.10 154
62 kerberos연동된 CDH 6.3.4에서 default realm값이 잘못된 상태에서 서비스 기동시 오류 2022.10.14 154
쓰기 태그
위로