행복한디벨로퍼

이번 포스팅에서는 spark-submit 실행시 스크립트상에서 설정할 수 있는 방법에 대해 정리하도록 하겠습니다.

해당 내용은 '빅데이터 분석을 위한 스파크2 프로그래밍' 책의 내용을 기반으로 정리하였습니다.

[ 애플리케이션 관련 설정 ]

• spark.app.name : 애플리케이션 이름. SparkConf의 appName으로 설정하는 것과 같은 속성
• spark.driver.cores : 드라이버가 사용할 코어 수. 클러스터 모드에서만 사용 가능하며 기본값은 1입니다.
• spark.driver.maxResultSize : collect() 메서드 등의 호출 결과로 생성된 결과 값의 최대 크기입니다. 최소 1M 이상으로 지정해야 하며, 이 값을 초과할 경우 전체 잡은 실패로 종료됩니다. 기본값은 1g입니다.
• spark.driver.memory : 드라이버가 사용할 메모리 크기이며, 클라이언트 모드에서 사용할 경우 반드시 SparkConf가 아닌 --driver-memory 실행 옵션이나 프로퍼티 파일을 사용해서 지정해야 합니다. 기본값은 1g입니다.
• spark.executor.memory : 익스큐터 하나의 메모리 크기를 지정합니다. 기본값은 1g입니다.
• spark.local.dir : RDD 데이터를 디스크에 저장하거나 셔플 시 매퍼의 결과를 저장하는 디렉터리를 지정합니다. 콤마(,)를 이용해 여러 위치를 지정할 수 있으며, 성능에 큰 영향을 주므로 반드시 빠른 로컬 디스크를 사용해야 합니다. 기본값은 /tmp 입니다.
• spark.master : 클러스터 매니저 정보를 지정합니다.
• spark.submit.deployMode : 디플로이 모드를 지정합니다. client 또는 cluster 모드를 사용할 수 있습니다.

• spark.driver.extraClassPath : 드라이버 클래스패스에 추가할 항목을 지정합니다. 이 속성은 SparkConf가 아닌 --driver-memory 실행 옵션이나 프로퍼티 파일을 사용해서 지정해야 합니다. 유사한 속성으로 spark.driver.extraJavaOptions, spark.driver.extraLibraryPath가 있으며 각각 드라이버 실행 시 필요한 자바 옵션과 라이브러리 정보를 지정하는 용도로 사용됩니다.
• spark.executor.extraClassPath : 익스큐터의 클래스패스에 추가할 항목을 지정합니다. 유사한 속성으로 spark.executor.extraJavaOptions와 spark.executor.extraLibraryPath가 있습니다.
• spark.files, spark.jars : 각 익스큐터의 실행 디렉터리에 위치할 파일들 또는 jar 파일들을 지정하며, 콤마(,)를 이용해 여러 파일을 지정할 수 있습니다.
• spark.submit.pyFiles : PYTHONPATH에 추가될 .zip, .egg, .py 파일을 지정하며, 콤마(,)를 이용해 여러 파일을 지정할 수 있습니다.
• spark.jars.packages : 익스큐터와 드라이버의 클래스패스에 추가될 의존성 jar정보를 메이븐 코디네이트 형식으로 지정 할 수 있습니다.

• spark.reducer.maxSizeInFlight : 셔플 수행 시 각 리듀서가 매퍼의 실행 결과를 읽어갈 때 사용할 버퍼의 크기를 지정합니다. 기본값은 48m입니다.
• spark.reducer.maxReqslnFlight : 리듀서에서 매퍼의 결과를 가져갈 때 동시에 수행 가능한 최대 요청 수를 지정합니다. 기본값은 int.MaxValue입니다.
• spark.shuffle.compress : 맵의 결과를 압축할 것인지에 대한 설정입니다. true로 설정할 경우 spark.io.compress.codec에 지정한 압축 코덱을 사용해 압축합니다.
• spark.shuffle.service.enabled : 외부 셔플 서비스를 사용할 것인지 여부를 지정합니다. 이와 관련된 내용은 이후의 동적 자원 할당 부분에서 다시 확인해 보겠습니다. 기본값은 false이며 true로 설정할 경우 외부 셔플 서비스를 사용하게 됩니다.

• spark.eventLog.enabled : 스파크 이벤트 관련 로깅을 수행할 것인지를 설정합니다. 기본 값은 false이며 true로 설정할 경우 spark.eventLog.dir에 로깅을 수행할 경로를 지정해야 합니다. 이벤트 로깅을 활성화할 경우 종료된 애플리케이션에 대한 상세 실행 히스토리 정보를 스파크 UI에서 확인할 수 있습니다. 
• spark.ui.port : 스파크 UI 포트를 지정합니다. 기본값은 4040입니다.
• spark.ui.killEnabled : 스파크 UI를 통해 잡을 중지(kill)시킬 수 있도록 할 것인지 설정합니다. 기본값은 true입니다.
• spark.ui.retainedJob : 종료된 잡에 대한 정보를 몇 개까지 유지할 것인지 설정합니다. 유사한 옵션으로 spark.ui.retainedStages, spark.ui.retainedTasks, spark.ui.retainedExecutors, spark.ui.retainedDrivers, spark.ui.retainedBatches 등이 있습니다.

• spark.broadcast.compress : 브로드캐스트 변수의 값을 압축할 것인지 설정합니다. 기본값은 true입니다.
• spark.io.compression.codec : 브로드캐스트 변수나 셔플을 위한 중간 결과물 등 스파크 내부에서 사용하는 데이터를 압축할 때 사용할 압축 코덱을 지정합니다. l4z, lzf, snappy를 사용할 수 있으며 기본값은 lz4입니다.
• spark.kyro.classesToRegister : Kyro 직렬화를 위해 등록할 커스텀 클래스 정보를 지정합니다. 만약 클래스 등록 방식을 좀 더 커스텀하게 진행하고자 한다면 spark.kyro.registrator를 사용할 수 있습니다.
• spark.serializer : 스파크에서 사용할 객체 직렬화 방식을 설정합니다. org.apache.spark.Serializer의 하위 클래스를 지정할 수 있으며, 현재 스파크에서는 JavaSerializer와 KyroSerializer라는 두 클래스를 제공하고 있습니다. 

다음 메모리 관련 설정, 익스큐터 관련 설정, 네트워크 관련 설정, 보안 관련 설정, 암호화 관련 설정은 다음 포스팅에서 하도록 하겠습니다.

도움이 되셨다면 광고도 한 번 클릭해주시는 센스^_^

해당 내용은 '빅데이터 분석을 위한 스파크2 프로그래밍' 책의 내용을 정리한 것입니다.

실제로 실무에서 스파크로 작업된 결과를 hdfs에 남기기전에 coalesce명령어를 써서 저장되는 파일의 개수를 지정해주곤 했다.

업무에서 사용하긴 했지만 실제 repartition연산과 어떤점이 다른지 모르고 사용했었는데 책을 보며 알게되어 기록.

핵심은 셔플을 하느냐 안하느냐!!!

coalesce와 repartition

오우.....이런 중요한 차이점이 있었다니....그렇다면 coalesce를 사용하면 셔플을 발생시키지 않기때문에 파티션마다 데이터의 사이즈가 다를꺼고 hdfs write했을때 repartition으로 개수를 조정한것과는 다르게 사이즈가 뒤죽박죽이겠네?!!! (나중에 시간되면 테스트해보자)

[ 업데이트 내용 ] 

댓글에서 관련내용에 대해 적어주신분이 있어 확인할겸 관련 내용 업데이트 합니다.

실제 repartition내부는 coalesce메소드를 호출하는 형태로 되어있습니다.

coalesce내부 소스코드도 올려봅니다.

소스코드의 주석을 보면 'This results in a narrow dependency' 좁은 의존성을 초래한다고 적혀 있는데 관련해서는 따로 포스팅하도록 하겠습니다.

그리고 위에서는 coalesce는 파티션 수를 줄이는 것만 가능하다고 적어놨지만 'true'옵션을 주면 늘리는 것 또한 가능하네요.

하지만 기존 처리하던 partitions의 개수보다 많은 파티션수로 처리할 경우에는 반드시 shuffle옵션을 true로 주셔야합니다(매개변수로 넘겨주면됨)

이상으로 포스팅을 마치도록 하겠습니다.

도움이 되셨다면 광고도 한 번 클릭해주시는 센스^_^

'빅데이터 분석을 위한 스파크2 프로그래밍'책의 내용을 정리한 포스팅입니다.

RDD란?

스파크가 사용하는 핵심 데이터 모델로서 다수의 서버에 걸쳐 분산 방식으로 저장된 데이터 요소들의 집합을 의미하며, 병렬처리가 가능하고 장애가 발생할 경우에도 스스로 복구될 수 있는 내성을 가지고 있다. 즉, RDD란 스파크에서 정의한 분산 데이터 모델인데 내부에는 단위 데이터를 포함하고 있고 저장할 때는 여러 서버에 나누어 저장되며, 처리할 때는 각 서버에 저장된 데이터를 동시에 병렬로 처리할 수 있는 모델이다. 

RDD장점

데이터를 여러 서버에 나누어 저장하고, 처리하는 과정에서 일부 서버 혹은 데이터에 문제가 발생하더라도 스스로 에러를 복구할 수 있는 능력을 가지고 있는 데이터 모델이다.

RDD처리 방식

RDD에 속한 요소들은 파티션이라고 하는 더 작은 단위로 나눠질 수 있는데, 스파크는 작업을 수행할 때 바로 이 파티션 단위로 나눠서 병렬로 처리를 수행한다. 이렇게 만들어진 파티션은 작업이 진행되는 과정에서 재구성되거나 네트워크를 통해 다른 서버로 이동하는, 이른바 셔플링이 발생할 수 있다. 

이런 셔플링은 전체 작업 성능에 큰 영향을 주기 때문에 주의해서 다뤄야 하며, 스파크에서는 셔플링이 발생할 수 있는 주요 연산마다 파티션의 개수를 직접 지정할 수 있는 옵션을 제공한다.  (파티션의 수는 곧 데이터 처리에 참여하는 병렬 프로세스의 수이다. 즉, 하나의 데이터를 잘게 쪼개어 여러 개의 파티션을 만들면 여러 프로세스에서 동시에 작업을 처리해서 처리 속도가 증가할 수 있지만 이 정도가 지나치면 오히려 전체 성능을 떨어뜨리는 요인이 된다.

스파크의 장애시 RDD복구

하나의 RDD가 여러 파티션으로 나눠져 다수의 서버에서 처리되다 보니 작업 도중 일부 파티션에 장애가 발생해서 데이터가 유실될 수 있는데, 스파크는 손상된 RDD를 원래 상태로 다시 복원하기 위해 RDD의 생성 과정을 기록해 뒀다가 다시 복구해 주는 기능을 가지고 있다. RDD의  resilient라는 단어가 복구 능력을 의미하는데, 좀 더 정확하게 말하면 RDD에 포함된 데이터를 저장해 두는 것이 아니고 RDD를 생성하는 데 사용했던 작업 내용을 기억하고 있는 것이다. 그래서 문제가 발생하면 전체 작업을 처음부터 다시 실행하는 대신 문제가 발생한 RDD를 생성했던 작업만 다시 수행해서 복구를 수행한다.

정리하면, 스파크는 RDD가 생성되어 변경되는 모든 과정을 일일이 기억하는 대신 RDD를 한번 생성하면 변경되지 않는 읽기 전용 모델로 만든 후 RDD 생성과 관련된 내용만 기억하고 있다가 장애가 발생하면 이전에 RDD를 만들 때 수행했던 작업을 똑같이 실행해 데이터를 복구하는 방식을 사용한다. 이처럼 스파크에서 RDD 생성 작업을 기록해 두는 것을 리니지(linege)라고 한다.

도움이 되셨다면 광고도 한 번 클릭해주시는 센스^_^

하둡 데이터노드들에 색인시스템을 설치해 로그에서 관심사 뽑는 작업이 있었는데 전사 공통하둡클러스터로 넘어가게 되면 별도로 설치하기가 힘든 이유로 의존성을 제거해야하는 작업을 진행하게 되었다.

그 과정에서 키워드-관심사가 매핑되어 있는 데이터를 브로드캐스트 변수를 사용해 executor들에게 넘겨사용하는 과정에서 삽질했던 경험을 공유한다. 회사 업무로 진행해서 자세한 내용까지 포스팅하기는 힘들고 브로드캐스트에 대해 새로 알게된 내용들에 대해서만 기록해본다.

먼저 다음과 같이 브로드캐스트변수를 메인 클래스에 선언하고 해당 Broadcast.value를 executor들에게 넘겨서 사용하도록 하였다.

Driver에 Broadcast선언

문제는 값들이 정상적으로 넘어가 처리가 되었지만 실시간 배치(마이크로배치 : 1분)으로 처리되는 과정에 용납하기 힘든 delay가 발생하였다.

기존 스트리밍 처리 속도 (delay - no)

브로드캐스트 변수 적용 후

이정도 딜레이면 사용할 수 없다고 볼 수 있겠다....

원인을 찾으려고 삽질을 굉장히 많이 했었다.

원인은

스파크는 드라이버에서 각각의 executor들에게 작업을 분담시키고 executor들은 내부적으로 task를 만들어 수행하는데 broadcast를 executor에 넘기는 과정에서 모든 task마다 broadcast의 객체를 재생성하고 있었다. (실제 SuffixTree(매핑에 사용하는 객체)의 hashCode값을 찍어 보았다.

마이크로 배치잡으로 1분 간격으로 실행되는데 그때마다 15개의 executor들에서 250개정도의 task가 생기는데 각 task마다 객체를 다시 만들어 내고 있었던 것이다.....

문제해결

해결은 모든 task에서 객체를 생성하지 않고 각 executor들마다 한번만 생성하도록 싱글톤방식으로 수정하여 해결하였다.

주의 할 것은 브로드캐스트 변수의 값은 지역적이어야 하며 직렬화 가능해야 한다는 것이다!

다음과 같이 선언한 드라이버 내부에서 지역적으로 사용해야만 기대했던 방식대로 브로드캐스트를 사용할 수 있다.

이런식으로 action이 연산이 이루어지는 곳에 값을 넘겨 사용하게 되면 각 executor의 task마다 다시 생성한다는 사실....주의하자.

스스로의 기억을 위한 포스팅으로 내용이 자세하지 못한점 죄송합니당(꾸벅...)

Spark 스파크 지연 평과와 장애 내구성

스파크는 장애에 강하다라는 말을 쓰는데 이는 하드웨어나 네트워크 장애에도 작업이 완전히 실패하지 않고 데이터 유실이 일어나거나 잘못된 결과를 반환하지 않는다는 의미다.

스파크의 우수한 장애 내구성(fault-tolerance) 구현 방식은 각 파티션이 자신을 재계산하는 데 필요한 종속성 정보 등의 데이터를 갖고 있기 때문에 가능하다.

일반적인 분산 컴퓨터 패러다임은 데이터 변경을 일일이 로깅해 놓거나 노드들에 데이터를 복제해 놓는 방식으로 장애에 대비하는 반면에

스파크는 각 파티션이 복제에 필요한 모든 정보를 갖고 있으므로 각 RDD에서 데이터 변경 내역 로그를 유지하거나 실제 중간 단계들을 로깅할 필요가 없다.

만약 파티션이 유실되면 RDD는 재계산에 필요한 종속성 그래프에 대한 충분한 정보를 갖고 있으므로 더 빠른 복구를 위해 병렬 연산을 수행할 수도 있다.

메모리 영속화와 메모리 관리

맵리듀스와 비교해 스파크의 성능상 이점은 반복 연산이 들어 있는 사례에서 상당한 우위를 보인다. (다시 말해 모든 케이스에서 스파크가 월등히 빠른건 아니다, 실제 단순 작업의 경우에는 하둡 맵리듀스와 작업시간이 크게 안나는걸 여러번 경험했다.)

성능 향상의 많은 부분은 스파크가 메모리 영속화(in-memory persistence)를 활용하는 덕택이다. 스파크는 데이터가 거치는 각 단계마다 디스크에 기록하는 대신 이그제큐터의 메모리에 데이터를 로드해 놓을 수도 있다. 그러므로 파티션의 데이터에 접근이 필요할 대마다 메모리에서 꺼내 올 수 있다.  (스파크는 영속화를 위한 메모리 영역을 저장 장치처럼 따로 관리한다고 생각하면 된다.)

스파크는 메모리 관리에 대해 세 가지 옵션을 제공한다.

1. 메모리에 직렬화되지 않은 자바 객체

이 방식은 직렬화 하는 시간이 필요 없으므로 가장 빠르지만, 객체 그대로 저장하기 위해서 그를 표현하는 데이터도 같이 저장해야 하므로 메모리 공간 사용이 비효율적이다.

2. 메모리에 직렬화된 데이터

직렬화되지 않은 데이터를 읽는 것에 비해 직렬화된 데이터를 읽는 데에는 CPU가 더 많이 사용되므로 이 접근 방식은 더 느릴 것이다.

하지만 메모리 공간 사용 측면에서는 직렬화하지않고 저장하는 방식보다 뛰어나다. 자바의 기본 직렬화는 원본 객체보다는 효과적이지만 크리오(Kyro) 직렬화를 쓰면 공간 측면에서도 더욱 효과적이다. (무조건 Kyro쓰는 것을 권장한다.)

3. 디스크

각 executor의 램에 담기에 파티션이 너무 큰 RDD라면 디스크에 데이터를 쓸 수 있다. 이 전략은 당연히 반복 연산에는 속도 면에서 불리하다.

그러나 오래 걸리는 트랜스포메이션들이 반복되고, 가장 장애에 안전하고 또한 막대한 양의 연산을 해야 한다면 유일하게 선택할 수 있는 옵션이다.

해당 내용은 '하이 퍼포먼스 스파크(High Performance Spark)' 내용을 학습하다가 정리한 내용이다.

현상황  : Cloudera(클라우데라) 버전(CDH 5.5.1, Parcel), Spark버전(1.5) - jdk version 1.7

필요상황 : 기존 작업을 Spark1.5(jdk1.7) - jdk 1.8로 돌리기

준비상황 :  클러스터의 각 노드들에 jdk1.8이 설치되어 있어야 함.

spark-submit스크립트에 jdk1.8 path를 명시

--conf "spark.yarn.appMasterEnv.JAVA_HOME=/home1/irteam/jdk/jdk1.8.0_141" \
--conf "spark.driverEnv.JAVA_HOME=/home1/irteam/jdk/jdk1.8.0_141" \
--conf "spark.executorEnv.JAVA_HOME=/home1/irteam/jdk/jdk1.8.0_141" \

이렇게 driver와 executor의 JAVA_HOME은 명시가 되었고

해당 스크립트가 돌아가는 client의 JAVA_HOME은 export로 변경해준다.

기존 단순 spark-submit명령어에서 앞에 다음과 같이 추가

export JAVA_HOME=/home1/irteam/jdk/jdk1.8.0_141 && spark-submit \

이렇게 설정을 해주게 되면 jdk1.8로 빌드된 코드도 기존 spark로 돌릴 수 있게 된다.

이런식의 설정으로 클라우데라 스파크 버전 또한 2점대로 높여 사용할 수 있겠다.

Spark dataframe(스파크 데이터프레임)으로 작업 중 dataframe의 null값을 특정값으로 바꾸고 싶은 경우가 있다.

이 때 주의해야할 점은 dataframe의 컬럼의 자료형 타입에 맞게끔 변환해줘야 정상적으로 replace된다.

다음과 같은 데이터프레임(dataframe)이 있을 때 "bid_i"의 값을 0으로 변경하려고 다음을 실행

val result_df_q_new = result_df_q.na.fill(0, Seq("bid_i"))

위와 같은 명령을 수행하고 확인을 해도 정상적으로 null값이 0으로 변경되지 않은 걸 확인할 수 있었다.

원인은 bid_i의 자료형 타입에 맞지않게 변경하려했기 때문이다.

위에서 보듯이 "bid_i"의 자료형 타입은 string인데 0으로 변경(na.fill메서드를 하려고 하니 정상적으로 변환되지 않았던 것이다. 

(처리 도중 딱히 에러메세지가 없었다...)

val result_df_q_new = result_df_q.na.fill("0", Seq("bid_i"))

0을 string형(큰따옴표)를 씌워서 명령어를 주니 정상적으로 변경되는 것을 확인할 수 있었다.

데이터프레임(dataframe) 값을 na.fill을 통해 변경할 때는 자료형타입을 잘 확인하도록 하자!

Dataframe count중 scala.MatchError 발생 ( show 명령어는 정상적으로 먹는데??? )

간만에 스파크 작업을 진행하다가 다음과 같은 에러문구를 만났다.

Text 파일을 sc.textFile("path") 로 읽어와 해당 RDD를 Dataframe으로 변환하고 dataframe.count시 발생하였다.

문제의 원인은  실제 텍스트파일내의 내용을 RDD로 읽어와 split으로 나눈 후 Dataframe으로 변형하는 과정에서 

컬럼개수가 일치하지 않는 데이터가 있었다.

text파일의 포맷은 다음과 같았다.

이러한 RDD를 다음과 같이 Dataframe으로 만들려고 하였다.

전혀 문제가 없을 것 같지만 이렇게 변경 후 count명령을 내려보면 MatchError가 발생하는 것이다.

알고보니 itemName에 상품명중 탭 구분자("\t")가 들어가 있었던 상품명이 있었던 것이다.

그래서 실제로는 4개의 컬럼으로 나누어져서 Dataframe이 만들어져야 하지만 상품명에 탭구분자가 있는 경우

Array length가 4이상이되면서 macth case 조건에 정상적으로 매칭되지 않았던 것이다.

이럴 경우 Dataframe을 show로 보았을 때는 전혀 문제가 없어보일 수 있지만 count로 모든 데이터의 수를 세게 될 떄

특정 데이터에 문제가 있을 경우 matchError가 발생하게 되는 것이다.

해결방법은 map처리에서 flatMap처리로 변경 후 조건에 맞지 않는 데이터는 None처리를 진행하였다.

또 다시 같은 삽질을 하지 않기 위해 포스팅 남겨본다.

스파크 데이터프레임(Dataframe) partitionBy를 사용해 원하는대로 손쉽게 저장하자!

스파크(Spark) 데이터프레임(Dataframe) 혹은 데이터셋(Dataset)을 통해 작업하게 되면 

sql기반의 명령을 통해서 데이터를 손쉽게 활용할 수 있다는 점과 더불어 특정 컬럼 기반으로 

데이터를 저장할 수가 있다.

잠깐 데이터프레임(Dataframe)과 데이터셋(Dataset)에 대해 언급하자면 데이터셋(Dataset)은 데이터프레임과 RDD의 단점들을

보완한 모델로 Spark 1.6이상 버전부터 사용할 수 있다.

이번에 데이터프레임(Dataframe)을 partitionBy를 통해 저장해보았는데 이런 기능이 있다라고만 알았지

막상 써보니 너무 편해서 정리하게 되었다.

다음과 같은 데이터프레임(Dataframe)이 있을 때

partitionBy를 이용하면 특정 컬럼을 기반으로 디렉토리를 나누어 저장할 수 있다.

예를들어 advid를 기준으로 데이터를 나누어 저장하고 싶을 떄

라고 저장해주면 다음과 같이 파일들이 advid를 기준으로 저장되게 된다.

이 얼마나 간편한가!!!!

다들 partitionBy를 통해 데이터를 원하는대로 손쉽게 저장해서 사용하시길!!!