Relational Processing3

---

• 코딩테스트 준비해라.
• 그래프는 과감히 버리도록..
• 백준 코딩테스트 준비
• ios, 안드로이드와 순열,재귀는 아무 상관도 없지만. 생각하는 능력?을 기르기 위함
• 또 다른 능력을 보기 위함
• 좌절할 필요는 없지만 연습을 해라

✅ Relational Data

• 조인을 마무리 짓기
• 스파크에서 조인을 어떻게 할지
• 코드를 완성해야지만 기말고사를 잘 볼 수 있음

✅ map

• 조인키
• ??
• 조인키를 맵의 아웃풋키로
• 나머지를 밸류로 묶음
• 테이블 A에서 왔는지 B에서 왔는지 확인하기 위해 prefix넣음
• 밸류들을 묶음

✅ reduce

• 버퍼 : 루프가 돌 때 밸류 하나하나에 대해서 도는데 물건종류에 대한 것이 아직 입력이 안 돼있으면 안되므로 버퍼에 임시저장
• 앞에 3개는 버퍼에 임시저장. 루프가 끝나면 다시 버퍼를 읽음.
• 만약 description이 맨 앞에 있으면 좋지만. 정렬되어 있는 것이 아니므로 버퍼가 필요한 것.
• 강제적으로 한번 만들어보자 -> secondary sort

✅ preference

• secondary key를 가지고 sort를 하자.
• composite key(복합키) : 키를 여러 개 가지고 하나의 키처럼 사용하자

✅ composite key

• 복수개의 필드를 맵리듀스의 키로 사용
• 조인키와 테이블구별자를 섞어서 사용
• 앞 프라이머리키만 같으면 같은 엔트리로 묶자

✅ secondary sort

• 조인키만 이용하여 셔플
• b부터 들어가도록

✅ solution

• 조인키만 같으면 같은 리듀서로 들어감.
• 파티셔너
• 그룹컴퍼너터??
• 프라이머리키만 같으면 같은 리듀스에 집어넣기
• 라이터블 컴퍼러터??
• 시퀀스는 이미 있으므로 룰만 구현해주기
• 파티셔너 -> 룰을 정하는 것

✅ 추가

• doublestring, firstpartitioner, firstgroupingcomparator, compositekey어쩌구

✅ doublestring

• 2개를 비교하는 것?
• O(nlogn) ??
• comparable -> sorting을 위함, 순서를 정해서 버킷에 집어넣음
• readfields, write : 같은 순서로 읽고 쓰는 것
• datainput : javaio 패키지 안에 있는 것. 추상화시킨 것. 데이터를 읽어 와라. 뭐임..네트워크 또는 hdfs의 인터페이스라고 생각해라
• readFields : 데이터 인풋에서 문자열하나 읽어와.
• compareTo : 크고 작음을 판별해달라. 조인키와 테이블네임이 같으면 0,

✅ partitioner

• extends partitioner
• doublestring : map output key
• text : map output value
• numpartition : job을 수행할 때 필요한 리듀스의 개수
• joinkey가 같으면 같은 partition으로
• return은 reduce의 아이디
• 조인키가 같으면 같은 리듀스 아이디를 갖도록
• 해시코드 : 해시값
• 문자열이 같으면 문자열의 해시값은 항상 같아요
• 해시코드는 long
• 어떤 리듀스로 할당할 건지 룰을 작성한 것

✅ groupcomparator

• compare
• 입력 파라미터 : 두 개의 객체를 가지고 순서를 정하면 됨
• w1, w2는 doubleString의 타입
• 조인키가 같으면 0을 return
• 같은 키를 가지면 같은 리듀스에 들어가도록 같으면 0

✅ compositecomponator

• ??
• -1 : 역순으로 정렬

✅ 다른 점

• mapper : output key DoubleString
• reducer : input key DoubleString
• 메인에 클래스 지정코드 추가

🚨🚨🚨이건 반드시 준비하기. 무조건 시험에 나옴.!!

• 지금까지 한 것이 reduceSideJoin

🌟🌟 이제 맵에서 조인해보기 mapSideJoin

✅ key idea

• 하나가 작으면 합쳐서 올리기

✅ solution

• relation_b를 map을 수행하기 전 미리 읽고 (setup)

• mapper에서는 relation_a만 처리

• 붙여서 처리

• 셔플이랑 리듀스 없음

• 파일을 매퍼에서 읽는 법 :task tracker의 working directory를 복사해주는 매커니즘 -> distributed cache

• main에서 relation_b를 distributed cache에 추가해줌

• path는 hdfs의 path임 -> 모든 매퍼가 공유를 함

• 셋업 : 캐시된 파일의 목록을 가져올 수 있음 array로

• 한 개 밖에 없으므로 1번째가 우리가 원하는 것

• bufferedReader로 읽으면 한줄한줄 읽을 수 있음

• hash table의 key:A,B,F,K value: Adult, Baby, Food, Kitchin

public class DoubleString implements WritableComparable
{
	String joinKey = new String(); 
	String tableName = new String();
	public DoubleString() {}
	public DoubleString( String _joinKey, String _tableName )
	{
		joinKey = _joinKey;
		tableName = _tableName; 
	}
    
	public void readFields(DataInput in) throws IOException
	{
		joinKey = in.readUTF();
		tableName = in.readUTF(); 
	}

	public void write(DataOutput out) throws IOException { 
    	out.writeUTF(joinKey);
		out.writeUTF(tableName); 
    }
    
	public int compareTo(Object o1)
	{
		DoubleString o = (DoubleString) o1;
		int ret = joinKey.compareTo( o.joinKey );
		if (ret!=0) return ret;

		return -1*tableName.compareTo( o.tableName);
	}

	public String toString() { 
    	return joinKey + " " + tableName; 
    } 
}

public static class CompositeKeyComparator extends WritableComparator { 
	protected CompositeKeyComparator() {
    	super(DoubleString.class, true);
	} 

	public int compare(WritableComparable w1, WritableComparable w2) { 
    	DoubleString k1 = (DoubleString)w1;
		DoubleString k2 = (DoubleString)w2;
		int result = k1.joinKey.compareTo(k2.joinKey);

		if(0 == result) {
        	result = -1* k1.tableName.compareTo(k2.tableName);
		} 
		
        return result; 
    }
}

public static class FirstPartitioner extends Partitioner<DoubleString, Text>
{
	public int getPartition(DoubleString key, Text value, int numPartition)
	{
		return key.joinKey.hashCode()%numPartition; 
    }
}

public static class FirstGroupingComparator extends WritableComparator {
	protected FirstGroupingComparator()
	{
		super(DoubleString.class, true);
	}

 	public int compare(WritableComparable w1, WritableComparable w2) { 
    	DoubleString k1 = (DoubleString)w1;
		DoubleString k2 = (DoubleString)w2;

		return k1.joinKey.compareTo(k2.joinKey); }
	}
}