.

Data_Engineering_TIL(20210820)

[학습자료]

“스파크 완벽 가이드” 책을 읽고 정리한 내용입니다.

** 한빛미디어, 빌 체임버스 & 마테이 자하리아 지음, 우성환 & 이영호 & 강재원 옮김

[학습내용]

Spark 이란

1) 클러스터 환경에서 데이터를 병렬로 처리하는 라이브러리를 제공하는 통합 컴퓨팅 엔진

2) 자바, 스칼라, 파이썬 등 여러 프로그래밍 언어를 지원하고, SQL 뿐만 아니라 스트리밍, 머신러닝 등 넓은 범위의 라이브러리도 제공함

• spark 컴포넌트

Spark 의 철학

1) 통합

빅데이터 어플리케이션 개발에 필요한 통합 플랫폼을 제공하자

2) 컴퓨팅 엔진

통합이라는 철학을 중요하게 생각하지만 이는 컴퓨팅 파워에만 제한됨. 저장소에 저장된 데이터를 연산하는 역할만 수행할 뿐 데이터 저장소 역할을 수행하지 않음. 대신에 s3나 HDFS 등 외부 저장소와의 데이터 송수신을 위한 API를 제공함

3) 라이브러리

데이터 분석 작업에 필요한 통합 API를 제공하는 통합 엔진 기반의 자체 라이브러리. 스파크 엔진에서 제공하는 표준 라이브러리와 오픈소스 커뮤니티에서 서드파티 형태로 제공하는 다양한 외부 라이브러리를 지원함.

Spark의 등장배경

하드웨어의 발전으로 데이터 저장 및 수집 비용은 매우 저렴해졌지만, 데이터는 클러스터에서 처리해야할 만큼 거대해졌음. 지난 50년간 개발된 소프트웨어는 더는 자동으로 성능이 향상되지 않고 데이터 처리 어플리케이션에 적용한 전통적인 프로그래밍 모델도 더이상 발전을 못하고 있음. 따라서 새로운 프로그래밍 모델이 필요해졌으며 이런 필요성으로 Spark이 등장하게 되었음

Spark의 개발현황

1) 1.0 버전이전의 초기버전

함수형 연산 관점에서 API를 개발

** 함수형 연산 : 자바 객체로 이루어진 컬랙션에 맵이나 리듀스 같은 병렬 연산을 수행하는 방식

2) 1.0 버전이후

정형데이터를 기반으로 동작하는 신규 API인 spark SQL 라이브러리가 추가됨

이후 Dataframe, 스트리밍 등이 추가됨

Spark 실행하기

spark는 스칼라로 구현되어 자바 버츄얼머신 기반으로 동작함. 따라서 스파크를 실행하려면 자바를 설치해야함. 파이썬 API를 사용하려면 파이썬 2.7 버전 이상을 설치해야함. 마찬가지로 R을 이용해서 spark를 사용하고 싶으면 R을 설치해야함.

spark는 HDFS이 없는 로컬환경에서도 실행이 가능함. 하지만 하둡클러스터 위에서 spark를 실행하고 싶으면 spark 다운로드 페이지에서 하둡버전에 맞는 패키지 유형을 선택해 spark를 다운로드 받으면 됨.

Spark 기본 아키텍처

1) 클러스터 매니저

컴퓨터 클러스터 형태로 spark는 컴퓨터 클러스터에서 데이터 작업을 관리하고, 조율하는 프레임 워크임

spark에서 데이터 작업을 관리하고 조율하는 역할은 spark standalone 클러스터 매니저, 하둡 Yarn, 메소스 같은 클러스터 매니저가 있음

2) 스파크 어플리케이션

driver process와 executor process로 구성됨.

driver process : 클러스터 노드중 하나에서 실행되며 main() 함수를 실행함. 이는 스파크 어플리케이션 정보의 유지&관리, 사용자 프로그램이나 입력에 대한 응답, 전반적인 익슈큐터 프로세스의 작업과 관련된 분석, 배포, 스케쥴링 역할을 수행함

executor process : 드라이버 프로세스가 할당한 작업을 수행함. 즉, 드라이버가 할당한 코드를 실행하고 진행상황을 다시 드라이버 노드에 보고하는 두가지 역할을 수행함

• spark application 아키텍처

spark는 사용가능한 자원을 파악하기 위해 클러스터 매니저를 사용함

드라이버 프로세스는 주어진 작업을 완료하기 위해 드라이버 프로그램의 명령을 익스큐터에서 실행할 책임이 있음

SparkSession

스파크 어플리케이션은 SparkSession이라 불리는 드라이버 프로세스로 제어함

SparkSession 인스턴스는 사용자가 정의한 처리 명령을 클러스터에서 실행함.

하나의 SparkSession은 하나의 스파크 어플리케이션에 대응함

SparkSession과 spark 언어 API 간의 관계

사용자는 스파크 코드를 실행하기 위해 SparkSession 객체를 진입점으로 사용함.

python이나 R로 스파크를 사용할때는 JVM 코드를 명시적으로 작성하지 않음.

스파크는 사용자를 대신해 python이나 R로 작성한 코드를 익스큐터의 JVM에서 실행할 수 있는 코드로 변환함

예시코드

myRange=spark.range(1000).toDF("number")

한개의 컬럼과 1000개의 로우로 구성되며 각 로우에 0부터 999까지 값이 할당됨.

이 숫자들은 분산 컬렉션임

클러스터 모드에서 위에 코드를 실행하면 숫자 범위의 각 부분이 서로 다른 익스큐터에 할당됨.

DataFrame

가장 대표적인 구조적 API

DataFrame은 로우와 컬럼으로 단순하게 표현됨

컴럼과 컬럼의 타입을 정의한 목록을 스키마라고 부름

DataFrame은 컬럼에 이름을 붙인 스프레드시트와 비슷함

• DataFrame과 스프레드 시트의 차이점

스프테드시트는 한대의 컴퓨터에 저장되어 있음.

반면에 DataFrame은 수천대의 컴퓨터에 분산저장됨.

여러 컴퓨터에 데이터를 분산저장하는 이유는 단순하다. 단일 컴퓨터에 저장하기에는 데이터가 너무 크거나 계산이 너무 오래걸리기 때문이다.

Partition

스파크는 모든 익스큐터가 병렬로 작업을 수행할 수 있도록 파티션이라고 불리는 청크단위로 데이터를 분할함.

파티션은 클러스터의 물리적 머신에 존재하는 로우의 집합을 의미함

DataFrame의 파티션은 실행중에 데이터가 컴퓨터 클러스터에서 물리적으로 분산되는 방식을 나타냄. 만약에 파티션이 하나라면 스파크에 수천개의 익스큐터가 있더라도 병렬성은 1이 되버림. 또한 수백개의 파티션이 있더라도 익스큐터가 하나밖에 없다면 병렬성은 1이 되버림.

DataFrame을 사용하면 파티션을 수동 또는 개별적으로 처리할 필요가 없음. 물리적 파티션에 데이터 변환용 함수를 지정하면 스파크가 실제 처리방법을 결정함.

Transformation

스파크의 핵심 데이터 구조는 불변성이다. 즉, 한번 생성하면 변경할 수 없다.

DataFrame을 ‘변경’하려면 원하는 변경 방법을 스파크게 알려줘야 한다. 이때 사용하는 명령을 Transformation이라고 부른다.

DataFrame에서 짝수를 찾는 Transformation 예시코드

divisBy2=myRange.where("number % 2 = 0")

위의 코드를 실행하도 결과는 출력되지 않는다. 추상적으로 데이터 프레임을 변경하는 방법만 정의한 상태이기 때문에 Action이라는 연산을 호출하지 않으면 스파크는 실제 Transformation 연산을 수행하지 않는다.

Transformation은 스파크에서 비지니스 로직을 구현하는 핵심개념임

Transformation은 두가지 유형이 있음.

1) Narrow dependency

각 입력 파티션이 하나의 출력 파티션에만 영향을 미침 위에 코드예시에서 where 구문은 narrow dependency를 가짐. 따라서 아래 그림과 같이 하나의 파티션이 하나의 출력 파티션에만 영향을 미침

2) Wide dependency

하나의 입력 파티션이 여러 출력파티션에 영향을 미침.

Narror dependency를 사용하면 스파크에서 pipelining을 자동으로 수행함. 이말은 데이터프레임에 여러 필터를 지정하는 경우 모든 작업이 메모리에서 일어남.

반면에 wide dependency의 경우에는 스파크가 클러스터에서 파티션을 교환하는 shuffle 이 동작하는데 이는 셔플의 결과를 디스크에 저장한다는 점을 유의해야함.

Lazy evaluation

스파크가 연산 그래프를 처리하기 직전까지 기다리는 동작 방식을 의미

스파크는 특정 연산 명령이 내려진 즉시 데이터를 수정하지 않고 원시데이터에 적용할 트랜스포메이션의 실행계획을 생성함. 스파크는 코드를 실행하는 마지막 순간까지 대기하다가 원형 DataFrame 트랜스포메이션을 간결한 물리적 실행계획으로 컴파일함.

스파크는 이 과정을 거치며 전체 데이터 흐름을 최적화하는 강점을 갖고 있음.

DataFrame의 predicate pushdown이 예시가 될수 있다. 아주 복잡한 스파크 잡이 원시데이터에서 하나의 로우만 가져오는 필터를 가지고 있다면 필요한 레코드 하나만 읽는 것이 효율적이다. 스파크는 이 필터를 데이터 소스로 위임하는 최적화 작업을 자동으로 수행함

Action

사용자는 트랜스포메이션을 사용해 논리적 실행계획을 세울 수 있다. 하지만 실제 연산을 수행하려면 action 명령을 내려야 한다.

action은 일련의 트랜스포메이션으로부터 결과를 계산하도록 지시하는 명령임

가장 단순한 action인 count 메서드는 Dataframe의 전체 레코드 수를 반환함

divisBy2.count()

위에 코드를 실행하면 500이 출력될 것이다.

count 외에도 다음 세가지 유형의 action이 있음

1) 콘솔에서 데이터를 보는 액션

2) 각 언어로 된 네이티브 객체에 데이터를 모으는 액션

3) 출력 데이터소스에 저장하는 액션

액션을 지정하면 스파크 잡이 시작됨. 스파크 잡은 필터(narrow dependency)를 수행한 후 파티션별로 레코드 수를 count(wide transformation)함. 그리고 각 언어에 적합한 네이티브 객체에 결과를 모음. 이때 스파크가 제공하는 스파크 UI로 클러스터에서 실행중인 스파크 잡을 모니터링 할 수 있음