임시

RDD 개념

- RDD(Resilient Distributed Dataset)

- Resilient: Memory 내 데이터 손실 시, 다시 생성할 수 있음. 즉, 유실된  파티션을 재연산해 복구 (탄력적)

- Distributed: Cluster를 통해 메모리에 분산되어 저장 (분산)

- Dataset: 파일을 통해 가져올 수 있음

- RDD는 스파크에서 기본적인 데이터 단위라고 볼 수 있다.  스파크에서의 모든 작업은 새로운 RDD를 만들거나, 존재하는 RDD를 변형하거나, 결과 계산을 위해 RDD에서 연산을 호출하는 것 중의 하나로 표현됨

- RDD는 Lazy Execution!

- Action 함수의 실행 전까지는, 실제로 이벤트가 발생하지 않음

쉽게 말해서, RDD가 새로운 RDD를 만들었다고 가정했을 때(Transformation), 값들은 없는 빈 껍데기만 생성이 된다. 사슬처럼 RDD가 생성이 된다고 해도, 값은 없다.

Action 함수인 count를 했을 때, 비로소 RDD에 값이 들어가는 느긋한 실행?(ㅠㅠ) 방식으로 되어 있다.

그런데 이는 메모리에 값을 쓰고 있지 않기 때문에 성능면에서 상당히 좋은 방법이다.

만약 모든 라인을 로드해서 저장하면 얼마나 낭비겠는가!

RDD Function 종류

- Transformations: 존재하는 RDD에서 새로운 RDD 생성

- RDD의 데이터는 바꿀 수 없음

- map(function): RDD의 각 행별로 기능 수행(ex: 각 행의 ‘abc’가 포함되는 단어 Count하기)

- flatmap(function): RDD의 각 라인별 요소 단위로 매핑

(ex: 각 행들을 하나의 리스트로 통합 된다고 보면 쉬움(1,2,3행 -> 1행으로 통합))

- filter(function): 조건에 맞는 RDD의 행을 새로운 RDD로 생성(ex: 각 행의 길이가 x이상일 때 생성

- distinct(function): 중복 제거(ex:  행에 중복 문자열이 있을 경우 제거)

- {1,2,3,3} 의 값을 가진 RDD Transformation

- {1,2,3},{3,4,5} 두 RDD를 가진 Transformation

- Actions: 실제 수행하는 함수로, 값을 받아옴. 또는 외부 스토리지에 저장

(ex: reduce(RDD 값들을 병렬로 병합 연산), count, take(n) (n개의 리스트 반환) , collect(모든 데이터 반환)등)

- {1,2,3,3} RDD Actions

RDD 예제 (Local 에서 사용)

- RDD 예제 (Line 수 세기)

- 실제 104 라인으로 구성되어 있는 것을 확인할 수 있다. 

(spark 실행을 bin파일 내에서 실행하였기 때문에 상위 경로의 path 사용)

>>>lines = sc.textFile(“../README.md”)

lines이라는 RDD를 생성하면 연산이 즉시 수행되는 것이 아닌, 메타데이터에 요청이 들어왔다는 것만 기록해둠

그런데 때때로 동일한 RDD를 여러 번 사용하고 싶을 때도 있지 않을까? 

-> persist(), cache()있음. [차이점은 cache()는 persist()에서 저장 옵션을 MEMORY_ONLY로 한 옵션과 동일)

(저장 옵션: https://spark.apache.org/docs/latest/programming-guide.html#rd-persistence  참고)

>>>lines.count()

count() 라는 Action을 수행하면 비로소 아래와 같이 각 라인에 데이터가 채워지고 (Lazy Execution), 라인 수를

카운트 하게 된다.

- 앞서 설명한, take(5)의 경우 5라인까지만 출력되는 것을 확인할 수 있다.  

- RDD 예제 (필터링)

- RDD 예제 (RDD 제곱)

-RDD 예제(Word Count)