[HBase]text 파일 hbase에 input하기 1. 맵퍼 클래스

   

package insertCsv;

import java.io.BufferedReader;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

//import org.apache.hadoop.fs.FileSystem;
//import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Mapper;

public class AirMapper extends Mapper<LongWritable, Text, ImmutableBytesWritable, Put> {
    // longWritable은 텍스트 라인번호, text는 한줄 string 타입
    // 출력은 (불변)바이트코드 타입 -->로우키, put은 ??
   
    private SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
    private BufferedReader br = null;
    private Date date = null;
   
    private HTable table =null; //테이블 인스턴스, HTable 선언은 리소스를 많이 쓰므로 각 쓰레드당 한번씩만 선언하는게 좋다.

    private final byte[] COLUMN_FAMILY = "AIRFLIGHT_DATA".getBytes(); //컬럼 패밀리선언 바이트형식으로 변환시켜 놓는다.
   
    @Override
    public void setup(Context context) throws IOException,InterruptedException{
        table = new HTable(context.getConfiguration(),"AIRFLIGHT_TABLE"); //테이블 선언
        table.setAutoFlush(false, true); //메모리데이터 --> 디스크로 설정
    }
    @Override
    public void cleanup(Context context) throws IOException, InterruptedException{
        br.close(); //스트림닫기
        table.flushCommits(); //테이블 내용을 플러쉬후
        table.close(); //종료
    }
   
    @Override
    public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException{
       
            parseCsv parser = new parseCsv(value); //한줄 씩 읽어서 분리
           
            String rowKey = new String(parser.getYear()+","+parser.getMonth()+","+
            parser.getDay_of_month()+","+parser.getUnique_carrier()+","+
            parser.getFlightnum()); //key는 년도 월 일 항공사번호 항공기 번호로 구성
                //key를 정하는 전략은 매우 중요하다,!! 서로 관련된것이 인접하게 정렬되게끔 하는것 이 좋다.
               
            List<Put> puts = new ArrayList<Put>(); //리스트 puts 객체를 선언해 rpc로 인한 부하를 줄여준다.
               
            Put put = new Put(Bytes.toBytes(rowKey)); //row키로 put객체 선언
           
            put.add(COLUMN_FAMILY, Bytes.toBytes("ad"), Bytes.toBytes(parser.getArrdelay()));
            puts.add(put);
            put.add(COLUMN_FAMILY, Bytes.toBytes("dd"), Bytes.toBytes(parser.getDepdelay()));
            puts.add(put);
            put.add(COLUMN_FAMILY, Bytes.toBytes("di"), Bytes.toBytes(parser.getDistance()));
            puts.add(put);
           
            //put.add(COLUMN_FAMILY, Bytes.toBytes("data"), ts, Bytes.toBytes(valueColumns[1]));
            //형식을 잘보자 (컬럼패밀리, 컬럼명(quliafier), 타임스탬프, 데이터값)
            //모든 값은 바이트로 전환하여 put객체에 추가 한다.
            //생성자는 여러가지가 존재한다. ts는 시스템시간상으로 설정 할수도 있다..4
                                           
            table.put(puts);
    }

}

오늘 책을 응용하여 만든 맵퍼 클래스...

책에 있는 데이터 예제는 메타데이터 text파일에서 참조를 해 디렉토리경로를 찾아가 삽입하는 예제다.

내가 쓰는 예제는 하나의 csv파일에 모든 data가 들어가 있는 비교적 다루기 쉬운 예제.

데이터 분해 잘되고 로우키도 잘 들어간다.! 타임스탬프는 시스템 시간으로 잘 들어가고 문제는 컬럼 패밀리의 데이터값

데이터 형태를 더 볼 필요가 있다. ( type이 맞지 않는 값이 들어가 있나보다 ... )

**그나저나 컴파일해 작동 시키면 노트북이 바로 항공기 소리를 낸다 ..,, 웅웅우우우웅