[ Spark ] 스파크 브로드캐스트 변수 사용하기

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

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

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

Driver에 Broadcast선언

Broadcast<SuffixTree<Multiset<String>>> keywordCategoryBrdTree = jssc.sparkContext().broadcast((LexiconMain.getSuffixTree()));

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

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

브로드캐스트 변수 적용 후

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

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

원인은

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

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

문제해결

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

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

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

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

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