Cloudera CDH/CDP 및 Hadoop EcoSystem, Semantic IoT등의 개발/운영 기술을 정리합니다. gooper@gooper.com로 문의 주세요.
* 출처 : http://kysepark.blogspot.kr/2016/03/spark-streaming.html
Spark Streaming을 작성하여 동작 시, 옵션으로 코어 개수, 익스큐터의 총 개수(yarn 모드로 동작 시), 메모리 사이즈, 각종 데이터에 쓰이는 로컬 디스크 개수 등의 하드웨어 자원등의 할당으로 성능 개선을 할 수 있다.
그러나 코드레벨 단에서 병렬화 처리가 제대로 되지 않아 주어진 자원을 충분히 활용하지 못하게 되는 문제가 발생하게 된다.
위와 같은 문제로 현재까지 팀내에서 경험을 통해 성능을 개선했던 내용에 대해 아래에 정리하였다.
- Spark Streaming으로 데이터를 받을 경우에 리시버의 수를 적절히 증가시켜주어야 한다.
- Kafka를 사용했을 때를 예로 들어보면 검색등을 통해 쉽게 찾아보면 다음과 같이 DStream을 생성하여 사용할 수 있다.val conf = new SparkConf().setAppName("Test").setMaster("local[8]")
val sc = new SparkContext(conf)
val ssc = new StreamingContext(sc, Seconds(5))val topics = List(("TestTopic", 10)).toMap
val kafkaDStreams = KafkaUtils.createStream(ssc, "test1.zookeeper.com,test2.zookeeper.com,test3.zookeeper.com", "ConsumerGroupId", topics) - 이 경우에는 DStream을 하나만 사용하게 된다. 그러므로 초당 처리해야할 데이터 유입이 더 많다면 다음과 같이 리시버의 개수를 늘려주어야 한다.
...val topics = List(("TestTopic", 10)).toMap
val numInputDStreams = 5
val kafkaDStreams = (1 to numInputDStreams).map(_ => KafkaUtils.createStream(ssc, "test1.zookeeper.com,test2.zookeeper.com,test3.zookeeper.com", "ConsumerGroupId", topics))
val unifiedStream = ssc.union(kafkaDStreams)
- 복수개의 DStream을 생성한 후에 union을 하여 하나의 DStream으로 사용할 수 있게 한다. 위와 같이 할 경우 초당 유입할 수 있는 데이터의 수가 크게 증가하게 된다.
- 주의할 사항은 예시한 샘플 코드에서는 로컬pc에서 8개의 thread를 사용하도록 되어 있다. 'numInputDStreams' 값이 실행 thread 개수보다 많거나 비슷하게 되어 있으면 RDD가 정상적으로 동작하지 않으므로 테스트 시에 이 점을 주의해야한다.
- 병렬화 수준을 높이기 위한 RDD의 파티션의 개수 조정
- repartition() 메소드를 사용하여 병렬화 개수가 너무 많거나 적을 경우 조정이 가능하다. repartition() 메소드 사용 시 데이터 셔플이 발생할 수 있다.
- 파티션 개수를 줄인다면 coalesce() 메소들 쓸 수 있다. 이 메소드 사용 시엔 데이터 셔플이 발생하지 않는다.
- 실제 RDD의 연산에서는 다음과 같은 코드로 간단하게 repartition()을 수행할 수 있다....
unifiedStream.repartition(10)
... - 실제 저 파티션 개수를 조정하면서 최적의 성능을 적절한 개수를 파악해나가는 게 좋다.
이 외에도 Kyro를 사용한 직렬화라든가 메모리 관리 등등 여러 방법이 있겠으나 실제 직면했던 문제에서 개선해나갔던 내용에 대해서만 남긴다.
결론적으로 어떤 작업을 처리하기 위해 Spark 코드를 작성하였다면 가지고 있는 리소스의 충분한 활용을 위해 옵션을 어떻게 주고 코드레벨에서의 병렬화 처리에 대한 처리 및 튜닝을 해야 한다는 교훈을 얻었다.
그렇다면 어떠한 문제에 대해 Spark를 활용하고자 한다면 다음과 같은 순으로 작업이 진행되어야 한다고 생각한다.
문제 파악 및 해결방안 마련 -> 설계 -> Spark를 활용하여 구현 -> 테스트 -> 성능 튜닝 -> 릴리즈
