[Spark The Definitive Guide] Part 2-1. 구조적 API: DataFrame, SQL, Dataset

데이터 엔지니어링 심화트랙 1주차 정리 내용

 

Chapter 4. 구조적 API 개요

• 구조적 API는 비정형 로그 파일부터 반정형 CSV 파일, 매우 정형적인 파케이(Parquet) 파일까지 다양한 유형의 데이터를 처리할 수 있다.
• 구조적 API에는 Dataset, DataFRame, SQL테이블과 뷰의 세 가지 분산 컬렉션 API가 있다.
• batch, streaming 처리에서 구조적 API를 사용할 수 있다. 구조적 API를 활용하면 배치 작업을 스트리밍 작업으로 손쉽게 변환 가능하다.

 

 이 장에서 다음 세 가지 기본 개념을 이해하고 넘어가야 한다.

1. typed/untyped API의 개념과 차이점
2. 핵심 용어
3. 스파크가 구조적 API의 데이터 흐름을 해석하고 클러스터에서 실행하는 방식

 

4.1. DataFrame과 Dataset

• 스파크는 DataFrame과 Dataset이라는 두 가지 구조화된 컬렉션 개념을 가지고 있다.
• DataFrame과 Dataset은 잘 정의된 로우와 컬럼을 가지는 분산 테이블 형태의 컬렉션이다.
• 결과를 생성하기 위해 어떤 데이터에 어떤 연산을 적용해야 하는지 정의하는 지연연산의 실행계획이며, 불변성을 가진다.
• DataFrame에 액션을 호출하면 스파크는 트랜스포메이션을 실제로 실행하고 결과를 반환한다.

4.2. 스키마

• 스키마는 DataFrame의 컬럼명과 데이터 타입을 정의

 

4.3. 스파크의 구조적 데이터 타입 개요

• 스파크는 실행 계획 수립과 처리에 사용하는 자체 데이터 타입 정보를 가지고 있는 카탈리스트(Catalyst) 엔진을 사용한다.
• 카탈리스트 엔진은 다양한 실행 최적화 기능을 제공한다.
• 다음 코드 예제는 스파크 데이터 타입의 스파크 덧셈 연산이다.

 

4.3.1. DataFrame과 Dataset 비교

• 구조적 API에는 untyped인 DataFrame과 typed인 Dataset이 있다. 
• DataFrame을 사용하면 스파크의 최적화된 내부 포맷을 사용할 수 있다.

4.3.2. 컬럼

• 컬럼은 정수형이나 문자열 같은 단순 데이터 타입, 배열이나 맵 같은 복합 데이터 타입 그리고 null 값을 표현한다.

4.3.3. 로우

• 로우는 데이터 레코드
• range 메서드를 이용해 dataframe을 생성하는 예제

 

4.3.4. 스파크 데이터 타입

-> 스파크 데이터 타입을 파이썬에서 사용하려면 위와 같은 코드 사용

 

4.4. 구조적 API의 실행 과정

1. DataFrame/Dataset/SQL을 이용해 코드를 작성
2. 정상적인 코드라면 스파크가 논리적 실행 계획으로 변환
3. 스파크는 논리적 실행 계획을 물리적 실행 계획으로 변환하며, 그 과정에서 추가적 최적화가 가능한지 확인
4. 스파크는 클러스터에서 물리적 실행 계획(RDD 처리)을 실행

작성한 코드를 콘솔이나 spark-submit 셸 스크립트로 실행 -> 카탈리스트 옵티마이저는 코드를 넘겨받고 실제 실행 계획을 생성 -> 스파크는 코드를 실행한 후 결과를 반환

 

4.4.1. 논리적 실행 계획

• 논리적 실행 계획 단계에서는 추상적 트랜스포메이션만 표현
• 스파크 분석기는 컬럼과 테이블을 검증하기 위해 카탈로그, 모든 테이블의 저장소 그리고 DataFrame 정보를 활용
• 검증 결과는 옵티마이저로 전달된다.

 

4.4.2. 물리적 실행 계획

• 스파크 실행 계획이라고도 불리는 물리적 실행 계획은 논리적 실행 계획을 클러스터 환경에서 실행하는 방법을 정의한다.

 

4.4.3. 실행

• 스파크는 물리적 실행 계획을 선정한 다음, 저수준 프로그래밍 인터페이스인 RDD를 대상으로 모든 코드를 실행한다.
• 스파크는 전체 task나 stage를 제거할 수 있는 자바 바이트 코드를 생성해 추가적인 최적화를 수행합니다.
• 마지막으로 스파크는 처리 결과를 사용자에게 반환한다.

 

Chapter 5. 구조적 API 기본 연산

• DataFrame은 Row 타입의 레코드(테이블의 로우 같은)와 각 레코드에 수행할 연산 표현식을 나타내는 여러 column(스프레드시트의 컬럼 같은)으로 구성된다. 
• DataFrame의 파티셔닝은 DataFrame이나 Dataset이 클러스터에서 물리적으로 배치되는 형태를 정의한다.
• 파티셔닝 스키마는 파티션을 배치하는 방법을 정의한다.

DataFrame 생성 후, 스키마 확인

 

5.1. 스키마

 

 

5.2. 컬럼과 표현식

• 스파크의 컬럼은 스프레드시트, R의 dataframe, Pandas의 DataFrame 컬럼과 유사하다.
• 사용자는 표현식으로 DataFrame의 컬럼을 선택, 조작, 제거할 수 있다.
• 컬럼 내요을 수정하려면 반드시 DataFrame의 스파크 트랜스포메이션을 사용해야 한다.

5.2.1. 컬럼

 

5.2.2. 표현식

• 표현식: DataFrame 레코드의 여러 값에 대한 트랜스포메이션 집합을 의미
• 컬럼은 단지 표현식일 뿐이다.
• 컬럼과 컬럼의 트랜스포메이션은 파싱된 표현식과 동일한 논리적 실행 계획으로 컴파일된다.

 

DataFrame 컬럼에 접근하기

 

5.3. 레코드와 로우

 

5.4. DataFrame의 트랜스포메이션

DataFrame을 다루는 방법

 

 

5.4.2. select와 selectExpr

select와 selectExpr을 사용하면 데이터 테이블에 SQL을 실행하는 것처럼 DataFrame에서도 SQL을 사용할 수 있다.

 

 

 

 

-> 컬럼명을 destinationd으로 변경

 

 

 

 

5.4.3. 스파크 데이터 타입으로 변환하기

 

 

5.4.4. Column 추가하기

dataframe에 신규 컬럼 추가 -> withColumn 메서드 사용

-> withcolumn는 두 개의 인수 가짐 -> 하나는 컬럼명, 다른 하나는 값을 생성할 표현식

-> withcolumn 메서드로 컬럼병도 변경할 수 있음

 

5.4.5. 컬럼명 변경하기

 

5.4.6. 예약 문자와 키워드

 

5.4.7. 대소문자 구분

기본적으로 스파크는 대소문자를 가리지 않는데, 설정을 통해 구분하게 할 수 있다.

 

5.4.8. 컬럼 제거하기

 

5.4.9. 컬럼의 데이터 타입 변경하기

 

5.4.10. 로우 필터링하기

 

-> 조건 중첩

 

5.4.11. 고유한 로우 얻기

 

5.4.12. 무작위 샘플 만들기

 

5.4.13. 임의 분할하기

 

5.4.14. 로우 합치기와 추가하기

• DataFrame은 불변성을 가진다.
• DataFrame에 레코드를 추가하려면 원본 데이터프레임을 새로운 데이터프레임과 통합해야 한다.

 

5.4.15. 로우 정렬하기

-> 기본 동작 방식은 오름차순 정렬

 

-> 정렬 기준을 명확히 지정

 

-> 트랜스포메이션을 처리하기 전에 성능을 최적화하기 위해, 파티션별 정렬을 수행

 

5.4.16. 로우 수 제한하기

-> 추출할 로우 수 제한

 

 

5.4.17. repartition과 coalesce

• 자주 필터링하는 컬럼을 기준으로 데이터를 분할 -> 파티셔닝 스키마와 파티션 수를 포함해 클러스터 전반의 물리적 데이터 구성을 제어할 수 있다.
• repartition 메서드를 호출하면 무조건 전체 데이터를 셔플한다. 향후에 사용할 파티션 수가 현재 파티션 수보다 많거나 컬럼을 기준으로 파티션을 만드는 경우에만 사용해야 한다.
• 특정 컬럼을 기준으로 자주 필터링한다면 자주 필터링되는 컬럼을 기준으로 파티션을 재분배하는 것이 좋다.

 

-> 전체 데이터를 셔플하지 않고 파티션을 병합하려는 경우에 사용하는 coalesce 메서드

-> 목적지를 기준으로 셔플을 수행해 5개의 파티션을 나누고, 전체 데이터를 셔플 없이 병합하는 예제

 

5.4.18. 드라이버로 로우 데이터 수집하기

스파크는 드라이버에서 클러스터 상태 정보를 유지한다. 로컬 환경에서 데이터를 다루려면 드라이버로 데이터를 수집해야 한다.

-> collect 메서드로 전체 데이터프레임의 모든 데이터를 수집

-> take 메서드는 상위 N개의 로우를 반환

-> show 메서드는 여러 로우를 보기 좋게 출력

 

Chapter 6. 다양한 데이터 타입 다루기

6.1. API는 어디서 찾을까

 

-> 분석에 사용할 dataframe을 생성하는 예제

 

6.2. 스파크 데이터 타입으로 변환하기

• 프로그래밍 언어의 고유 데이터 타입을 스파크 데이터 타입으로 변환 
• 데이터 타입 변환은 lit 함수를 사용

 

6.3. 불리언 데이터 타입 다루기

 

 

-> or 구문은 동일 구문에 조건 정의해야 함, and는 차례로 조건 나열하는 것이 편함

 

-> 불리언 컬럼을 사용해 DataFrame을 필터링

 

 

6.4. 수치형 데이터 타입 다루기

• count는 빅데이터 처리에서 필터링 다음으로 많이 수행하는 작업이다.
• 대부분은 수치형 데이터 타입을 사용해 연산 방식을 정의하기만 하면 된다

 

 

-> 두 컬럼 사이의 상관관계 계산

 

-> 피어슨 상관계수 계산

 

-> 컬럼의 요약 통계는 describe 메서드를 이용

 

-> 데이터의 백분위수를 계산

 

-> 교차표(cross-tabulation)나 자주 사용하는 항목 쌍을 확인하는 용도의 메서드 제공

 

-> 로우에 0부터 시작하는 고유 ID값을 추가

 

6.5. 문자열 데이터 타입 다루기

• 문자열 다루는 작업은 거의 모든 데이터 처리 과정에서 발생
• 로그 파일에 정규 표현식을 사용해 데이터 추출, 데이터 치환, 문자열 존재 여부, 대/소문자 변환 처리 등의 작업 가능

-> 모든 단어의 첫 글자르 대문자로 변경

 

-> lower, upper는 문자열 전체를 소/대문자화

 

-> 문자열 주변의 공백 제거, 추가하는 작업

 

6.5.1. 정규 표현식

• 문자열의 존재 여부를 확인하거나 일치하는 모든 문자열을 치환할 때는 보통 정규 표현식을 사용한다.

-> 'description' 컬럼의 값을 'COLOR'로 치환

 

-> 주어진 문자를 다른 문자로 치환

-> 정규 표현식이 아닌 translate 함수로 치환

 

-> 처음 나타난 색상 이름을 추출하는 것과 같은 작업

 

-> instr 함수를 사용해 값의 존재 여부 확인

 

-> 인수의 개수가 동적으로 변하는 상황

-> 문자열의 위치를 정수로 반환하는 locate 함수를 사용 -> 위치 정보를 불리언 타입으로 변환

 

6.6. 날짜와 타임스탬프 데이터 타입 다루기

• 스파크는 두 가지 종류의 시간 관령 정보만 집중적으로 관리
• 하나는 달력 형태의 날짜, 다른 하나는 날자와 시간 정보를 모두 가지는 타임 스탬프
• 스파크는 inferSchema 옵션이 활성화된 경우 날짜와 타임스탬프를 포함해 컬럼의 데이터 타입을 최대한 정확하게 식별하려 시도한다.
• 스파크는 특정 날짜 포맷을 명시하지 않아도 자체적으로 식별해 데이터를 읽을 수 있다.

-> 오늘 날짜와 현재 타임스탬프값

 

-> 위에서 만든 데이터프레임을 사용해 오늘을 기준으로 5일 전후의 날짜를 계산

 

-> 두 날짜 사이의 차이를 datediff 함수를 사용해 계산

 

-> 문자열을 날짜로 변환

 

-> 스파크는 날짜를 파싱할 수 없으므로 NULL값 반환

 

6.7. null 값 다루기

• 명시적으로 null 값을 제거
• 전역 or 컬럼 단위로 null 값을 특정 값으로 채워 넣기

6.7.1. coalesce

-> 스파크의 coalesce 함수는 인수로 지정한 여러 컬럼 중 null이 아닌 첫번째 값을 반환한다.

 

6.7.3. drop

-> 기본적으로 null 값을 가진 모든 로우를 제거

-> any를 지정하면 로우의 컬럼값 중 하나라도 null 값을 가지면 해당 로우를 제거

-> all을 지정하면 모든 컬럼의 값이 null이거나 NaN인 경우에만 해당 로우를 제거한다.

 

6.7.4. fill

-> fill 함수를 사용해 하나 이상의 컬럼을 특정 값으로 채울 수 있다.

 

6.7.5. replace

-> 조건에 따라 다른 값으로 대체

 

6.9. 복합 데이터 타입 다루기

-> 복합 데이터 타입을 가진 DataFrame 생성

-> 이를 다른 DataFrame을 조회하는 것과 동일하게 사용할 수 있다

-> 유일한 차이점은 문법에 점(.)을 사용하거나 getField 메서드를 사용하는 것

 

6.9.2. 배열

-> 배열로 변환

 

-> 배열의 길이 조회

 

-> 배열에 특정 값이 존재하는지 확인

 

 

6.9.3. 맵

 

---

참고자료

• "스파크 완벽 가이드" , 빌 체임버스 , 마테이 자하리아 저자(글) · 우성한 , 이영호 , 강재원 번역