행복한디벨로퍼

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

그 과정에서 키워드-관심사가 매핑되어 있는 데이터를 브로드캐스트 변수를 사용해 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를 통해 데이터를 원하는대로 손쉽게 저장해서 사용하시길!!!

[ Spark ] JavaRDD로 saveAsTextFile했는데 데이터가 정상적으로 나오지 않는다???

Spark(스파크) 작업을 하고나서 결과물을 hadoop(하둡)에 쓰고 싶을 경우 보통 saveAsTextFile 메소드를 사용하게 된다.

그런데 적재를 하고 하둡 명령어로 해당 파일을 열어봤는데 파일의 내용이 정상적으로 안써지고 dataframe을 RDD로 변경할 때 사용한

모델 패키지명으로 노출되는 경우가 있다. 다음과 같이...

처음 스파크로 작업을 했을 때 적잖이 당황했던 기억이나서 포스팅한다.

코드를 보게되면

다음과 같은 식으로 integratedDF라는 변수명의 dataframe을 JavaRDD로 변환하는데 매핑 모델로 AdidBidPairingModel을 쓰고 있다.

해당 모델을 보면 다음과 같다.

스파크에서 특정 클래스파일로 데이터를 쓸 때 해당 클래스의 toString메소드를 참고해 데이터를 쓰게되는데?(내추측) 

해당 모델 클래스에는 toString메소드가 없다...

보통 일반적인 스프링프로젝트에서는 lombok을 써서 알아서 생성해주지만 일반 자바 프로젝트에서는 toString메소드를 작성해주어야 한다.

다음과 같이 AdidBidPairingModel 끝부분에 toString메소드를 정의해주고 다시 spark-submit을 해보면 ~

정상적인형태로 데이터가 나오는 것을 확인할 수 있다!!!!

처음 스파크프로젝트로 작업을 하시는 분들은 꽤나 헤맬 수 있다...나또한 나중에 또 헤맬 수 있으므로 오랜만에 spark작업하는 김에 기록해본다!!!

스파크와 맵리듀스  성능차이 그리고 엘라스틱서치

최근 업무하면서 경험했던 이슈들에 대해서 정리해볼까 한다.

Episode1. 스파크(Spark)와 하둡 스트리밍 MR작업의 성능 이슈

함께 업무를 하던 과장님께서 하둡 맵리듀스로 된 작업을 스파크로 변경하셨었다.

그런데 스파크로 코드를 작성하고 테스를 하시더니 맵리듀스 보다 1분 30초나 느리다는 것이다.

이 무슨 말이란 말인가??? 스파크가 하둡MR 보다 성능이안나온다니!

그래서 함께 코드를 봤더니 코드는 뭐 딱히 문제가 될 만한 부분도 없을만큼 단순한 코드였다.

단순히 파일을 rdd로 읽어서 parsing해서 결과파일을 쓰는....

그래서 다시 과장님이 스파크 어플리케이션을 돌리실동안 클라우데라에 들어가 잡이 돌아가는 상황을

Application Master로 보았더니 굉장히 많은 shuffle이 일어나고 있었고 순간 spark-submit할 때

어떤 옵션들이 들어있는지에 대한 의문이 들었다.

확인을 해보니….하둡 스트리밍 MR작업을 돌릴 때는

해당 옵션이 있었고 새로만든 스파크 작업에는 해당 옵션을 주지 않고 spark-submit이 이루어졌던 것이다.

결과적으로 하둡 스트리밍MR은 클러스터의 사용가능한 충분한 executor를 사용하여 작업이 이루어진 반면에

스파크작업은 10개 이내의 executor들을 사용해 작업이 진행되었던 것이다.

같은 옵션을 주고 다시 테스트를 해보니 당연히 스파크의 승리!!! 약 20초정도 더 빨랐던 것 같다. 

데이터가 커지고 로직이 더 복작해지면 질수록 이 성능은 더 차이가 많이 나지 않을까 생각한다.

Episode2. 엘라스틱서치(elastic search)와 키바나(cabana)

최근 데이터 유입쪽과 카프카-camus를 통해 hdfs 적재되는 데이터량을 쉽게 확인할 수 있는 시스템?을 개발하였다.

기존에는 데이터유입부분과 실제 hdfs에 데이터가 적재되는 양을 비교할 수 없어 데이터가 정상적으로 유입부터 적재까지

이루어지고 있는지 확인할 수 있는 방법이 없었다. 가 아니라 실제 데이터들을 다 까서 로그들의 개수를 읽어서 매칭시켜보는 방법이 있었다....(노가다...)

따라서 이부분에 대한 모니터링 작업이 필요한 상황이였다.

그래서 파이썬스크립트로 데이터유입서버와 hdfs에 적재되는 커맨드 서버에서 로그파일의 row수를 세서 시간별로 데이터 row count를 

엘라스틱서치(elastic search)에 적재하도록 하였다. 이렇게 쌓인 데이터는 키바나(kibana)를 통해 쉽게 시각화할 수 있게 함으로써 

편하게 확인할 수 있도록 작업을 진행하였다.

작업을 진행하면서 느낀것은 역시 써보지 않은 tool을 사용해 작업을 하는 것은 쉽지 않고 간단한 작업이라도 꽤나 시간을 많이 잡아 먹는다는 것이다.

하지만 새로운 도구를 경험하고 사용하면서 경험적인 측면에서 한 단계 성장해나가는 것 같은 기분이들어 좋았다.

생각보다 elasticsearch-kibana 를 사용하는데 많이 힘들진 않았지만 elastic search 인덱스를 만드는 부분과 kibana에서

index=true로 되어있지 않은 데이터들에 대해서는 그래프의 지표로 쓸수 없다는 부분을 깨닫는데까지 많은 삽질을 했던게 가장 기억에 남는다.

앞으로도 사용하고 있는 tool, framework에만 의존하기 보다는 다양한 도전과 시도를 통해 여러 문제에 대해 적합한 솔루션을 제공할 수 있도록

항상 여러 기술에 관심을 가지고 사용해볼 수 있도록 노력해야겠다는 생각이 들었다.

스파크(Spark)에서 OutOfMemoryError:Java heap space 다음과 같은 에러 메세지가 발생했다면

스파크 어플리케이션 내부에서 다음과 같은 메서드를 사용하지 않았는지 확인해볼 필요가 있다.

사실 이전에 collect() 메서드 사용시 주의점을 포스팅하긴했었다...

해당 메서드들을 rdd에 사용하게 되면 각각의 executor 노드들에서 돌고있는 rdd element들을 driver로 copy하려고 한다.

따라서 RDD의 사이즈가 크다면 driver에서 해당 RDD들을 저장할 메모리가 부족하게 되고 이에 OOME(OutOfMemory)가 발생할 수 있다.

이번에 나도 무심결에 RDD를 collectAsMap()해버려 위와 같은 에러가 발생하였다...

단순히 RDD를 map형태로 만들어 놓고 다른 RDD연산을 위해 사용하려고 했던 생각해서 저지른 실수였다...

물론 충분한 driver memory를 spark-submit시 할당해 주면 되겠지만 아무래도 클러스터 내에 해당 어플리케이션만 동작하는게 아니기 때문에 

좋지 않은 방법이다.

따라서 사이즈가 큰 데이터를 map형태로 이용하고 싶다면 Storage(RDBMS, NOSQL)을 사용하기를 권한다.