메뉴 건너뛰기

Cloudera, BigData, Semantic IoT, Hadoop, NoSQL

Cloudera CDH/CDP 및 Hadoop EcoSystem, Semantic IoT등의 개발/운영 기술을 정리합니다. gooper@gooper.com로 문의 주세요.


* 출처 : http://kysepark.blogspot.kr/2016/03/spark-streaming.html



Spark Streaming을 작성하여 동작 시, 옵션으로 코어 개수, 익스큐터의 총 개수(yarn 모드로 동작 시), 메모리 사이즈, 각종 데이터에 쓰이는 로컬 디스크 개수 등의 하드웨어 자원등의 할당으로 성능 개선을 할 수 있다.
그러나 코드레벨 단에서 병렬화 처리가 제대로 되지 않아 주어진 자원을 충분히 활용하지 못하게 되는 문제가 발생하게 된다.
위와 같은 문제로 현재까지 팀내에서 경험을 통해 성능을 개선했던 내용에 대해 아래에 정리하였다.

  1. Spark Streaming으로 데이터를 받을 경우에 리시버의 수를 적절히 증가시켜주어야 한다.
    1. 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)
    2. 이 경우에는 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)
    3. 복수개의 DStream을 생성한 후에 union을 하여 하나의 DStream으로 사용할 수 있게 한다. 위와 같이 할 경우 초당 유입할 수 있는 데이터의 수가 크게 증가하게 된다.
    4. 주의할 사항은 예시한 샘플 코드에서는 로컬pc에서 8개의 thread를 사용하도록 되어 있다. 'numInputDStreams' 값이 실행 thread 개수보다 많거나 비슷하게 되어 있으면 RDD가 정상적으로 동작하지 않으므로 테스트 시에 이 점을 주의해야한다.
  2. 병렬화 수준을 높이기 위한 RDD의 파티션의 개수 조정
    1. repartition() 메소드를 사용하여 병렬화 개수가 너무 많거나 적을 경우 조정이 가능하다. repartition() 메소드 사용 시 데이터 셔플이 발생할 수 있다.
    2. 파티션 개수를 줄인다면 coalesce() 메소들 쓸 수 있다. 이 메소드 사용 시엔 데이터 셔플이 발생하지 않는다.
    3. 실제 RDD의 연산에서는 다음과 같은 코드로 간단하게 repartition()을 수행할 수 있다.
      ...
      unifiedStream.repartition(10)
      ...
    4. 실제 저 파티션 개수를 조정하면서 최적의 성능을 적절한 개수를 파악해나가는 게 좋다.


이 외에도 Kyro를 사용한 직렬화라든가 메모리 관리 등등 여러 방법이 있겠으나 실제 직면했던 문제에서 개선해나갔던 내용에 대해서만 남긴다.
결론적으로 어떤 작업을 처리하기 위해 Spark 코드를 작성하였다면 가지고 있는 리소스의 충분한 활용을 위해 옵션을 어떻게 주고 코드레벨에서의 병렬화 처리에 대한 처리 및 튜닝을 해야 한다는 교훈을 얻었다.
그렇다면 어떠한 문제에 대해 Spark를 활용하고자 한다면 다음과 같은 순으로 작업이 진행되어야 한다고 생각한다.
문제 파악 및 해결방안 마련 -> 설계 -> Spark를 활용하여 구현 -> 테스트 -> 성능 튜닝 -> 릴리즈
번호 제목 날짜 조회 수
322 원격의 origin/master를 기준으로 dev branch를 만들어 작업후 원격의 origin/dev에 push하는 방법 file 2016.11.22 867
321 .gitignore파일에 지정되지 않은 파일이 ignore되는 경우 확인방법 2016.11.22 872
320 github에 있는 프로젝트와 로컬에서 작업한 프로젝트 합치기 2016.11.22 878
319 특정 커밋 시점(commit id를 기준으로)으로 돌리기(reset) 2016.11.21 608
318 Github를 이용하는 전체 흐름 이해하기 2016.11.18 269
317 특정 단계의 commit상태로 만들기(이렇게 하면 중간에 반영된 모든 commit를 history가 삭제된다) 2016.11.17 881
316 git 초기화(Windows에서 Git Bash사용) 2016.11.17 1105
315 spark notebook 0.7.0설치및 설정 2016.11.14 753
314 참고할만한 spark예제를 설명하는 사이트 2016.11.11 576
313 Kafka Offset Monitor로 kafka 상태 모니터링 하기 file 2016.11.08 1480
312 Eclipse실행시 Java was started but returned exit code=1이라는 오류가 발생할때 조치방법 2016.11.07 862
311 [SparkR]SparkR 설치 사용기 1 - Installation Guide On Yarn Cluster & Mesos Cluster & Stand Alone Cluster file 2016.11.04 552
310 데이타 분석및 머신러닝에 도움이 도움이 되는 사이트 2016.11.04 772
309 java스레드 덤프 분석하기 file 2016.11.03 248
308 centos 6에서 mariadb 5.1 to 10.0 으로 upgrade 2016.11.01 265
» Spark Streaming 코드레벨단에서의 성능개선 2016.10.31 817
306 Flume과 Kafka를 사용한 초당 100만개 로그 수집 테스트 file 2016.10.31 1199
305 Flume을 이용한 데이타 수집시 HBase write 성능 튜닝 file 2016.10.31 806
304 How-to: Build a Complex Event Processing App on Apache Spark and Drools file 2016.10.31 546
303 How-to: Tune Your Apache Spark Jobs (Part 2) file 2016.10.31 457
위로