Spark 프로그래밍

RDD 

- 복구 가능한 분산 데이터 모델

- 메서드 인자에 데이터 처리를 위한 함수를 직접 작성해서 전달

 

DataFrame

- 내장함수와 표현식 등을 통한 다소 간접적인 처리방식을 사용

- DataFrame의 split() : org.apache.spark.sql.catalyst.expressions.Substring 이라는 클래스의 인스턴스를 생성해서 리턴하는 역할을 수행

- 데이터 프레임의 연산

> 사전엔 정의된 표현식과 내부 함수를 이용해 코드 작성

> 논리적 트리로 변환 후,

> 전반적인 최적화 과정을 수행 후,

> 최종동작을 수행 ( collect() )

- 스파크에서 정의한 Row라는 타입의 객체로만 구성된 데이터셋의 특별한 형태

- 최적화 과정을 거친 후, 최종적을는 RDD API를 호출해서 동작한다.

 

map() , flatMap() 을 쓸때 RDD와 DataFrame의 차이

- RDD

> 전체 데이터 요소 하나하나에 일일이 적용

 

- DataFrame

> 스칼라 언어가 제공하는 런타임 바이트 코드 생성기능을 이용해 cpu 부하를 최소화하는 최적의 바이트코드로 변환해서 적용

> 즉, 최적화를 수행한다. --> 기본적으로 RDD 보다 권장됨.

 

 

스파크 개요

 

1. 스파크 클러스터

- 데이터가 여러 서버에 나눠져 병렬로 처리된다.

 

2. 분산데이터로서의 RDD

- 작업 수행도중 서버나 네트워크, 자원할당등에 일시적/부분적 문제가 발생했을때 RDD의 작업 히스토리를 이용한 재시도를 수행함 ( 작업히스토리는 리니지하게 되어있음 )

 

3. RDD의 불변성

- 데이터가 유실되면 RDD를 이용해 다시 생성하는 방식으로 복구한다.

 

4. 파티션

- RDD 데이터는 클러스터를 구성하는 여러 서버에 나누어 저장된다.

- 이렇게 분할된 데이터를 파티션이라는 단위로 관리한다.

> 하나의 HDFS 블록에 하나의 파티션 구성

 

5. HDFS

- 스파크는 하둡의 파일 입출력 API에 의존성을 갖고 있다.

- 스파크의 데이터 입출력은 하둡의 InputFormat 과 OutputFormat API를 쓴다.

- 하둡 HDFS 블록단위가 보통 파티션 단위가 된다. 조정가능

 

6. Job과 Executor

- 하나의 잡은 클러스터에서 병렬로 처리된다.

- 서버마다 익스큐터라는 프로세스가 생성됨

- 이를 통해 각자 할당된 파티션을 처리한다. 

 

7. 드라이버 프로그램

- 스파크에서 잡을 실행하는 프로그램

- 메인함수를 갖고 있는 프로그램

- 자신을 실행한 서버에서 동작하면서 스파크 컨텍스트를 생성해 클러스터의 각 워커 노드들에게 작업을 지시하고 결과를 취합

 

8. 트랜스포메이션과 액션

- 리턴타입이 RDD이면 트랜스포메이션

- 리턴타입이 RDD가 아니면 액션

 

9. 지연동작과 최적화

- 액션이 실행되야 진짜 실행되는 방식

- 최적화 가능 (== 셔플 파일수를 줄일수 있다)

 

 

RDD 액션

 

1) collect

- RDD의 모든 원소를 모아서 배열로 반환

JavaRDD<Integer> rdd= sc.parallelize(Arrays.asList(1, 2, 3, 4, 5);

List<Integer> result = rdd.collect();

for ( Integer i : result ) System.out.println(i);

 

2) count

- RDD를 구성하는 전체 요소의 개수를 반환

JavaRDD<Integer> rdd= sc.parallelize(Arrays.asList(1, 2, 3, 4, 5);

long result  = rdd.count();

 

 

RDD 트랜스포메이션

- 맵연산 , 그룹화 연산, 집합연산, 파티션 연산, 필터와 정렬 연산

 

* 집합연산?

- RDD에 포함된 요소를 하나의 집합으로 간주할때 서로다른 RDD간의 합집합, 교집합등을 계산

 

* 파티션 연산?

- RDD의 파티션 개수를 조정

 

 

1) map

- 하나의 입력을 받아 하나의 값을 돌려주는 함수를 인자로 받는다.

- RDD의 모든 요소에 이 함수를 적용한 뒤, 그 결과로 구성된 새로운 RDD를 생성해 반환한다.

JavaRDD<Integer> rdd2 = rdd1.map((Integer v1) -> v1+1);

 

2) flatMap

- 이차원 배열을 일차원으로 만든다.

-> 배열속에 포함된 요소를 모두 배열 밖으로 나오게하는 작업을 한다.

JavaRDD<Integer> rdd2 = rdd1.flatMap((String s) -> Arrays.asList(t.split(",")).iterator());

rdd2.collect();

rdd1 = {{"apple,orange"}{"grape,apple,mango"}}

 

- "apple,orange"를 ","로 구분하여 자른후에 하나씩 값을 증가해서 flatMap을 적용해서 이중배열안에 있는 것을 완전 밖으로 나오게 한다.

- 하나씩 값을 증가한다 == apple, orange, grape 순으로 순차적으로 진행한다.

- flatMap을 적용해서 이중 배열 안에 있는것을 가져올 수 있다. = {{"apple"}} --> "apple"

 

3) mapPartitions

- 파티션 단위로 요소를 처리한다

- 전달받은 함수를 파티션 단위로 적용, 그 결과로 구성된 새로운 RDD를 생성한다

- numbers는 각 파티션에 담긴 요소를 가리킨다. 파티션내의 이터레이터

JavaRDD<Integer> rdd1 = sc.parallelize(Arrays.asList(1,2,3,4,5,6,7),  3);

JavaRDD<Integer> rdd2 = rdd1.mapPartitions((Iterator<Integer> numbers) -> {
	System.out.println("db연결");
    List<Integer> result = new ArrayList<>();
    numbers.forEachRemaining(i-> result.add(i+1));
    return result.iterator();
});

System.out.println(rdd2.collect());

 

 

4) mapPartitionsWithIndex

- 인자로 전달받은 함수를 파티션 단위로 적용하고 그 결과값으로 구성된 새로운 RDD를 생성

-  인자로 전달되는 함수를 호출할때 파티션에 속한 요소의 정보뿐 아니라 해당 파티션의 인덱스 정보도 함께 제공한다.

- 아래는 첫번째 파티션에서만 결과를 추출하는 예제다

JavaRDD<Integer> rdd2 = rdd1.mapPartitionsWithIndex((Integer idx, Iterator<Integer> numbers) -> {
    System.out.println("db연결");
    List<Integer> result = new ArrayList<>();
    if (idx == 1) 
        numbers.forEachRemaining( i-> result.add(i+1));
    return result.iterator();
}, true);

rdd2.collect();

 

5) mapValues

- key-value 쌍일때 value에만 map연산을 수행해서 새로운 rdd를 생성 & 반환

JavaRDD<String> rdd1 = sc.parallelize(Arrays.asList("a", "b", "c"));
JavaPairRDD<String, Integer> rdd4 = rdd1
    .mapToPair((String t) -> new Tuple2<String, Integer>(t, 1))        // 단일 RDD -> 페어 RDD
    .mapValues((Integer v1) -> v1+1);                                        // 페어 RDD의 값에만 연산적용
rdd4.collect();
// 결과 : {{"a", 2} , {"b", 2} , {"c", 2}}

 

6) flatMapValues

- key-value 쌍일 때 value에만 flatMap연산을 수행

JavaRDD<String> rdd1 = sc.parallelize( Arrays.asList(new Tuple2(1,"a,b"), new Tuple2(2, "a,c")) );
JavaPairRDD<Integer, String> rdd4 = rdd1.flatMapValues((String v1) -> Arrays.asList(v1.split(",")));
rdd4.collect();

// 결과 : ((1,a) (1,b) (2,a) (2,c))

 

그룹연산

 

7) zip

- 서로 다른 RDD를 각 요소의 인덱스에 따라 하나의 (키, 값) 쌍으로 묶어준다.

- 서로 크기가다르면 사용 불가능

rdd1= (a,b,c)    // 약식
rdd2= (1,2,3)    // 약식
JavaPairRDD<String, Integer> result = rdd1.zip(rdd2);
System.out.println(result.collect());

// 결과 : (a,1) , (b,2) , (c,3)

 

8) groupBy

JavaRDD<Integer> rdd1 = sc.parallelize(Arrays.asList(1,2,3,4,5,6,7,8,9,10));
JavaPairRDD<String, Iterable<Integer>> rdd2
	= rdd1.groupBy((Integer v1) -> (v1 % 2 == 0) ? "even" : "odd");
rdd2.collect()

// even, [2,4,6,8,10]
// odd, [1,3,5,7,9]

 

 

9) groupByKey

- key-value 인 경우 사용

- 키를 기준으로 같은 키를 가진 요소들로 그룹을 만든다.

(a,1),(b,1),(c,1),(b,1),(c,2) => (a, 1) (b, (1,1)) (c, (1,2))

 

10) cogroup

- key-value 인 경우 사용

- 같은 키를 갖는 값 요소를 찾아서 키와 그 키에 속하는 요소의 시퀀스로 구성된 튜플을 생성

rdd1 = (k1, v1) (k2, v2) (k1, v3)
rdd2 = (k1, v4)
result = rdd1.cogroup(rdd2)

//  k2 (v2), ()     // 키, rdd1 요소집합, rdd2 요소집합  (만약 rdd2.cogroup(rdd1) 으로 하면 반대로 나옴)
//  k1 (v1,v3), (v4)

 

집합연산

 

11) distinct

- RDD의 원소에서 중복을 제거

 

12) cartesian

- 두 RDD 요소의 칼테시안 곲을 구하고 그 결과를 요소로한다.

- 모든 조합을 다 생성

rdd1 = (1,2,3)
rdd2 = (a,b,c)
result = rdd1.cartesian(rdd2)

// (1,a) , (1,b) , (1,c) , (2,a) ,(2,b) ....

 

13) subtract

rdd1.subtract(rdd2)

- rdd1에서 rdd2를 뺀다.

 

14) union

rdd1.union(rdd2)

- 두요소를 하나로 합친다. (합집합)

 

15) intersection

rdd1.intersection(rdd2)

- rdd1에도 있고 rdd2에도 있는 것 (교집합을 구하고 중복을 제거한다)

 

16) join

- 두 rdd에서 서로 같은 키를 가지고 있는 요소를 모아서 그룹을 형성하고 이 결과로 구성된 새로운 rdd를 생성한다.

- 데이터 베이스 조인과 유사하다.

- 리턴타입이 튜플이다.

 

17) leftOurterJoin, rightOuterJoin

 

18) subtractByKey

- key-value인 경우 사용

- 같은 키가 있으면 그 튜플자체를 제거한다.

 

 

집계

 

19) reduceByKey

- key-value인 경우 사용

- 같은 키를 가진 값들을 하나로 병합해 key-value를 생성

rdd.reduceByKey( (v1, v2) -> v1+v2);

// rdd1 = (a,1) (b,1) (b,1) --> (a,1) (b,2)

 

20) foldByKey

- reduceByKey 와동일

- 병합연산의 초기값을 메서드의 인자로 전달해서 병합시 사용가능

rdd.foldByKey(0, (v1, v2) -> v1+v2));

 

 

21) combineByKey

- reduceByKey와 동일하지만 value타입이 바뀔 수 있다

JavaPairRDD<String, Long) rdd = sc.parallelizePairs(data);
rdd.combineByKey( (Long v) -> new Record(v),   // Record는 내가만든 클래스, createCombiner
   (Record record, Long v) -> record.add(v)           // mergeValue
   (Record r1, Record r2) -> r1.add(v2)                 // mergeCombiners
);

 

22) aggregateByKey

- key-value 쌍인 경우 사용가능

- comebineByKey() 의 특수한 경우로서 병합을 시작할 초기값을 생성하는 부분이 추가된 케이스다.

- combineByKey에서는 createCombiner()로 초기값을 지정해주지만 이것은 0이라는 값으로 초기값을 지정해준다.

 

 

파티션 연산

 

23) pipe

- 데이터를 처리하는 과정에서 외부 프로세스를 활용할 수 있다.

- 세개의 숫자로 구성된 문자열을 리눅스의 cut 유틸리티를 이용해 분리한 뒤 남은것을 반환

 

24) coalesce , repartition

- rdd를 생성한 뒤 filter() 연산을 비롯한 다양한 트랜스포메이션 연산을 수행하다보면 최초에 설정한 파티션 개수가 적절하지 않은 경우 발생할 수 있음. 그때 파티션 개수 조정을 위해 쓰는 함수들 

- coalesce : 파티션 줄이기만 가능 - 셔플 이용 안함

- repartition : 줄이기 늘리기 둘다 가능 - 셔플 이용함

 

 

// 쓸만한거

subtractByKey - 키로제거

intersection    - 교집합인거 출력(중복인 부분출력...? 굳이?)

union  - 원본문서 합칠때?

distinct    - 배열에서 중복제거... 흠... 안쓸듯도

groupByKey - 여러개의 body를 모은다? 

contents:body { " adfa", "Adfa", "adfadf" ...}

contents:title    { " dfkdf", "adfa", "Afdaf" ... }