[spark definitive guide] - 스파크 간단히 살펴보기
이제 스파크를 사용해보자
DataFrame 과 SQL 을 사용해 클러스터, 스파크 어플리케이션 ,
그리고 구조적 API 를 봐보자
스파크의 기본 아키텍처
데이터 처리는 한 대의 컴퓨터로 수행하기 힘들 수 있다.
한 대의 컴퓨터는 대규모의 정보를 연산 할 만한 자원이나 성능을 가지지 못한다.
컴퓨터 클러스터는 여러 컴퓨터의 자원을 모아 하나의 컴퓨터 처럼 사용할 수 있게 만든다.
하지만 이 클러스터를 구성하는 것만으론 부족하고 작업을 조율 할 수 있는 프레임워크가 필요하다.
스파크가 이런 역할을 하는 프레임 워크다.
스파크는 클러스터의 데이터 처리 작업을 관리하고 조율한다.
스파크가 연산헤 사용하는 클러스터는 스파크 스탠드 얼론 클러스터 매니저 ,
하둡 YARN , 메소스 같은 클러스터 매니저 에서 관리를 한다.
사용자는 클러스터 매니저에 스파크 어플리케이션을 제출하고 제출받은 클러스터 매니저는
어플리케이션 실행에 필요한 자원을 할당해 작업을 처리한다.
- 스파크 어플리케이션
스파크 어플리케이션은 드라이버 프로세스와 다수의 익스큐터 프로세스로 구성된다.
드라이버 는 클러스터 노드 중 하나에서 실행되며 main() 함수를 실행한다.
익스큐터 프로세스의 작업과 관련된 분석, 스케줄링 역할을 수행한다.
익스큐터는 드라이버 프로세스가 할당한 작업을 수행한다.
즉 드라이버가 할당한 작업을 수행하고 다시 드라이버 노드에 보고하는 역할을 한다.
스파크는 사용 가능한 자원을 파악하기 위해 클러스터 매니저를 사용하고
드라이버 프로세스는 주어진 작업을 완료하기 위해 명령을 익스큐터에서 실행할 책임을 가진다.
스파크의 다양한 언어 API
스파크는 다른 언어로 작성된 코드를 익스큐터의 JVM 에서 실행할 수 있게
코드를 변환한다.
스파크 시작하기
실제 스파크 어플리케이션을 개발하려면
사용자 명령과 데이터를 스파크 어플리케이션에 전송 하는 방법을 알아야 한다.
SparkSession 을 생성하면서 자세히 알아보자
콘솔에서 대화형 모드로 스파크를 시작하면 SparkSession 이 자동으로 생성되지만
spark-submit 으로 시작을 하려면 어플리케이션 코드에서 직접 SparkSession 객체를 생성해야 한다.
SparkSession
스파크 어플리케이션은 SparkSession 이라 불리는 드라이버 프로세스로 제어한다.
SparkSession 은 사용자가 정의한 처리 명령을 클러스터에서 실행한다.
하나의 SparkSession 은 하나의 스파크 어플리케이션에 대응된다.
스파크 콘솔에서 spark 를 치면
SparkSession 객체를 return 한다.
이제 이 SparkSession 으로 일점 범위의 숫자를 만들어보자
이 DataFrame 은 한 개 의 컬럼에 1000개의 로우로 구성된다.
0부터 999까지 값이 할당되고 이 숫자들은 분산 컬렉션을 나타낸다
클러스터 모드에서 실행하면 숫자 범위의 각 부분이 서로 다른 익스큐터에 할당된다.
DataFrame
DataFrame 은 대쵸적인 구조적 API 다 .
테이블의 데이터를 로우와 컬럼으로 단순하게 표현한다.
컬럼과 컬럼의 타입을 정의한 목록을 스키마라고 한다.
DataFrame 은 컬럼에 이름을 붙인 스프레드 시트와 비슷하다
DataFrame 은 수천 대의 컴퓨터에 분산되어 있다.
(
RDD 와의 차이점을 잠깐 보면,
RDD 는 기본적으로 직렬화(디스크에 데이터를 기록할 때)와 GC 를 사용하는데 메모리 오버헤드가 많았다
DataFrame 은 데이터를 메모리에 저장하지 않고 오프-힙 영역에 저장해서 이를 통해 오버헤드를 감소시켰다
)
- 파티션
스파크는 익스큐터가 병렬로 작업을 수행할 수 있도록 파티션이라 불리는 청크 단위로
데이터를 분할한다.
DataFrame 의 파티션은 실행 중에 데이터가 클러스터에서 물리적으로 분산되는 방식을 나타낸다.
만약 파티션이 하나라면 익스큐터가 아무리 많아도 병렬성은 1이 되고
또, 수백 개의 파티션이 있더라도 익스큐터가 1개라면 병렬성은 1이 된다.
DataFrame 은 파티션을 수동 처리할 필요가 없다.
물리적 파티션에 데이터 변환용 함수를 지정하면 스파크가 실제 처리 방법을 결정한다.
트랜스포메이션
이 코드는 결과를 출력하진 않는다.
추상적인 트랜스포메이션만 지정한 상태이기 때문에, 액션을 호출 하지 않으면
스파크는 실제 트랜스포메이션을 실행하지 않는다.
트랜스포메이션은 스파크에서 비즈니스 로직을 표현하는 핵심 개념이다.
좁은 의존성과 넓은 의존성 두가지의 유형이 있다.
좁은 의존성 트랜스포메이션은 각 입력 파티션이 하나의 출력 파티션에만 영향을 미친다.
넓은 의존성을 가진 트랜스포메이션은 하나의 입력 파티션이
여러 출력 파티션에 영향을 미친다.
스파크가 클러스터에서 파티션을 교환하는 셔플을 사용한다.
스파크는 셔플의 결과를 디스크에 저장한다.
지금은 이렇게 두 종류의 트랜스포메이션이 있다는것만 알고 가자
- 지연 연산
지연 연산은 연산 그래프를 처리하기 직전까지 기다리다가 동작하는 방식을 의미한다.
스파크는 특정 연산 명령이 내려진 즉시 데이터를 수정하지 않고
원시 데이터에 적용할 트랜스포메이션의 실행 계획을 생성한다.
스파크는 코드를 실행하는 마지막까지 대기하다가 간결한 물리적 실행 계획으로 컴파일 한다.
스파크는 이 과정을 거치며 전체 데이터 흐름을 최적화시킨다.
액션
사용자는 트랜스포메이션을 통해 논리적 실행 계획을 세울수있지만,
실제 연산을 수행하려면 액션 명령을 내려야 한다.
액션은 트랜스포메이션의 결과를 계산하도록 지시하는 명령어다.
액션을 지정하면 스파크 잡이 시작된다.
스파크 잡은 필터(좁은 트랜스포메이션)을 수행 후 파티션 별로
레코드 수를 카운트(넓은 트랜스포메이션) 한다.
종합 예제
미국 교통통계국의 항공 운항 데이터 중 일부를 스파크로 분석한다
csv 파일을 읽으면 파일은 여러 로우를 가지고 각 로우는 DataFrame 의 로우가 된다
옵션에서 DataFrame 스키마 정보를 알아내는 스키마 추론 기능을 사용하고
파일의 첫 로우를 헤더로 지정하는 옵션을 설정했다.
스파크는 스키마 정보를 얻기 위해 데이터를 조금 읽고,
타입에 맞게 분석을 하는데 , 운영환경에서는 꼭 엄격하게 스키마를 지정하는 옵션을 써야한다.
take 액션을 호출하면 결과를 볼 수 있다.
여기서 sort() 메서드를 추가해보자
sort 는 DataFrame 을 변경하지 않고 새로운 DataFrame 을 생성한다.
explain 을 통해 실행 계획을 볼 수 있는데
rangepatitioning 으로 sort 를 사용했을때 넓은 트랜스포메이션이 사용된걸 볼 수 있다.
sort 와 take 를 하면 이미지와 같은 방식으로 실행이 된다.
스파크는 셔플 수행 시 기본적으로 200개의 셔플 파티션을 생성하는데,
이 값을 줄여서 셔플의 출력 파티션 수를 줄일 수 도 있다.
사용자는 물리적 데이터를 직접 다루지 않고
이렇게 파티션 파라미터 같은 옵션 설정으로 실행 특성을 제어할수있다.
값을 변경하면 런타임이 크게 달라질 수 있다.
스파크 UI 에서 잡의 실행 상태와 스파크 잡의 물리적, 논리적 실행 특성을 확인 할 수 있다.