설계방안
Scala vs Java vs Python vs R
Spark의 구조적 API는 속도와 안정성 측면에서 여러 언어를 일관성 있게 다룰 수 있다. 때문에 개발자에게 익숙한 언어나 상황에 따라 가장 적합한 언어를 사용하면 된다.
하지만 구조적 API로 처리할 수 없고 UDF나 RDD를 사용해야 하는 경우 R이나 Python은 사용하지 않는 것이 좋다. 여러 언어를 넘나드는 UDF를 사용하면 데이터 타입과 처리 과정을 엄격하게 보장하기 어렵기 때문이다.
DataFrame vs SQL vs Dataset vs RDD
모든 언어에서 DataFrame, Dataset, SQL의 속도는 동일하다. 하지만 Python과 R에서 UDF를 정의하면 성능 저하가 발생할 수 있으므로 Java와 Scala를 이용하는 것이 좋다. 성능개선을 위해서는 DataFrame, SQL, Dataset을 이용해야 스파크의 최적화 엔진이 더 나은 RDD 코드를 만들어내기 때문에 UDF나 RDD 사용을 자제해야 한다.
RDD를 사용하려면 Scala나 Java를 사용해야 한다. 만약 Scala나 Java를 사용할 수 없다면 애플리케이션에서 RDD 사용을 최소한으로 제한해야 한다. Python으로 RDD 코드를 실행하면 Python프로세스를 오가는 많은 데이터를 직렬화해야 하기 때문에 매우 큰 데이터를 직렬화하면 많은 비용이 발생해서 안정성이 떨어질 수 있다.
RDD 객체 직렬화
Kryo는 자바 직렬화보다 훨씬 간결하고 효율적이다. Spark는 기본적으로 자바 직렬화를 사용하기 때문에 Kryo 직렬화를 사용하려면 spark.serializer 속성값을 org.apache.spark.serializer.KryoSerializer로 설정해야 한다.
보관용 데이터
빅데이터를 효율적으로 읽으려면 적절한 저장소 시스템과 데이터 포맷을 사용해야 한다.
파일 기반 장기 데이터 저장소
CSV와 같은 파일은 파싱 속도가 느리고 예외 상황이 자주 발생한다. 가장 효율적으로 사용할 수 있는 파일 포맷은 parquet가 있다. parquet는 데이터를 바이너리 파일에 컬럼 지향 방식으로 저장하고 쿼리에서 사용하지 않는 데이터를 빠르게 건너뛸 수 있도록 통계를 함께 저장하기 때문에 Spark에서 데이터를 빠르게 읽어드릴 수 있다.
분할 가능한 파일 포맷과 압축
분할 가능한 포맷을 사용하면 여러 태스크가 파일의 서로 다른 부분을 동시에 읽을 수 있다. 이로 인해 파일을 읽을 때 모든 코어를 활용할 수 있다. 반면 JSON 파일처럼 분할 불가능한 포맷을 사용하면 단일 머신에서 전체 파일을 읽어야 하므로 병렬성이 떨어진다.
압축 포맷 또한 분할 가능 여부를 결정하는 주요 요소이다. ZIP파일이나 TAR 파일은 분할할 수 없기 때문에 병렬로 읽어드릴 수 없다. 반면 gzip, bzip2와 같은 압축 파일은 분할이 가능하다.
테이블 파티셔닝
테이블 파티셔닝은 데이터의 날짜 필드와 같은 키를 기준으로 개별 디렉터리에 파일을 저장하는 것을 의미한다. 이를 이용해 Spark는 키를 기준으로 파일을 읽지 않고 건너뛸 수 있다.
예를 들어 쿼리에서 date 컬럼을 기준으로 필터링을 한다면 date 컬럼을 기준으로 파티션을 생성 하면 파티셔닝된 디렉터리 기준으로 필요 없는 파일을 읽지 않을 수 있다.
단, 파티셔닝을 할 때 너무 작은 단위로 분할하면 작은 크기의 파일이 대량으로 생성될 수 있는데, 이 경우 저장소 시스템에서 전체 파일의 목록을 읽을 때 오버헤드가 발생하므로 주의해야 한다.
버켓팅
데이터를 버켓팅하면 데이터를 한두 개 파티션에 치우치지 않고 전체 파티션에 균등하게 분산시킬 수 있기 때문에 성능이 향상된다.
데이터 지역성
데이터 지역성은 기본적으로 네트워크를 통해 데이터를 교환하지 않고 특정 데이터를 가진 노드에서 동작할 수 있도록 지정하는 것을 의미한다. 저장소 시스템이 스파크와 동일한 노드에 있고 해당 시스템이 데이터 지역성 정보를 제공한다면 Spark는 입력 데이터 블록과 최대한 가까운 노드에 태스크를 할당한다.
예를 들어 HDFS는 데이터 지역성 힌트를 제공하는 저장소이기 때문에 Spark에서 로컬 저장소로부터 데이터를 읽을 수 있는 상황을 알게 되면 기본적으로 데이터 지역성을 활용하여 효율적으로 처리한다.
병렬화
Spark의 병렬성을 높이기 위해 spark.default.parallelism과 spark.sql.shuffle.partitions 옵션을 클러스터 코어 수에 따라 설정한다. 스테이지에서 처리해야 할 데이터양이 매우 많다면 클러스터의 CPU 코어당 최소 2~3개이 태스크를 할당한다.
파티션 재분배와 병합
Spark에서는 데이터를 워커 노드에 알맞게 분배해주는 것이 중요하다. 데이터 분배가 제대로 되지 않은 경우에는 특정 워커 노드에게만 일이 몰리는 현상(skew)이 발생할 수 있다. 이를 방지하기 위해 개발자는 coalesce나 repartition 메서드를 통해 partition 개수를 조절할 수 있다.
파티션 재분배 과정은 shuffle을 실행하지만 클러스터 전체에 데이터가 균등하게 분배되므로 잡의 전체 실행 단계를 최적화할 수 있다. 일반적으로 가능한 한 적은 양의 데이터를 shuffle하는 것이 좋다.
coalesce vs repartition
두 메서드는 모두 파티션의 수를 조절한다는 공통점이 있지만 repartition 메서드는 파티션 수를 늘리거나 줄이는 것을 모두 할 수 있는 반면, coalesce 메서드는 파티션 수를 줄이는 것만 가능하다.
coalesce 메서드는 전체 데이터를 shuffle 하지 않고 파티션을 병합하려는 경우에 사용하고, repartition 메서드는 shuffle을 통해 파티션 수를 증가 시키거나 축소시킬 때 사용한다.
캐싱
애플리케이션에서 같은 데이터셋을 계속해서 재사용한다면 캐싱을 사용해 최적화할 수 있다. 캐싱은 클러스터 익스큐터 전반에 걸쳐 만들어진 임시 저장소(메모리나 디스크)에 DataFrame, 테이블 또는 RDD를 보관해 빠르게 접근할 수 있게 한다. 하지만 데이터를 캐싱할 때 직렬화, 역직렬화, 저장소 자원을 소모하기 때문에 항상 유용하지는 않다.
조인
groupByKey vs reduceByKey
reduceByKey로 해결할 수 있는 문제 상황에서는 무조건 reduceByKey를 사용해야 한다. groupByKey를 쓰게 되면 shuffle이 모든 노드에서 발생하지만, reduceByKey는 shuffle을 하기 전에 먼저 reduce 연산을 수행해서 shuffle 데이터를 줄여준다.
참고자료
- 스파크 완벽 가이드
