Spark streaming 02 - asloud/dev_note01 GitHub Wiki

출력 연산

• 출력 연산은 스트림에서 최종 트랜스포메이션의 결과에서 무엇이 필요한가를 정의하는 것이다.
• DStream의 출력 연산이 없다면 실제로 연산이 일어나지 않는다. (transformation이 동작하지도 않음), StreamingContext에 출력 연산이 없다면 시작도 하지 않는다.
• print() : 가장 일반적인 디버깅 출력 연산
  • DStream의 각 배치마다 첫 10개의 데이터를 출력.
• save() 연산들도 있음. 파일을 저장할 디렉토리와 추가적인 접미어를 인자로 받아들인다.
• saveAsHadoopFiles() : 하둡의 OutputFormat 객체를 받아들인다.
• foreachRDD() :
  • DStream이 RDD들에 임의의 연산을 수행하게 해 주는 범용적인 출력 연산.
  • foreachRDD() 안에서는 스파크에서 갖고 있는 모든 액션을 재사용할 수 있다.

입력소스

• 일부 핵심 소스들은 spark streaming 아티팩트 안에 있지만, 다른 것들은 추가적인 아티팩트를 maven pom.xml에 추가하여 라이브러리를 추가하여야 한다.
• 핵심소스
  • 핵심 소스들로부터 DStream을 생성하는 메소드들은 모두 StreamingContext에서 사용할 수 있다.
  • 파일, 소켓, Akka actor
  • 파일
    • 스파크는 모든 하둡 호환 파일 시스템에서 읽기를 지원하므로 spark streaming은 하둡 호환 파일 시스템의 파일로부터 스트림을 만드는 것을 지원.
    • 파일은 일관성을 유지해야 한다 (포맷 포함)
• 그외 입력 소스
  • 현재 여러 리시버를 지원
  • 트위터(Twitter), 아파치 카프카(Apache kafka), 아마존 키네시스(Amazon Kinesis), 아파치 플럼(Apache Flume), 제로MQ(ZeroMQ) 등을 지원한다.
  • 추가적인 리비서들은 메이븐 아티팩트에 추가하여 포함할 수 있다.
• Apache Kafka
  • 아파치 카프카는 속도와 유연함 때문에 인기있는 입력소스.
  • 카프카에 내장 지원 모듈을 사용하면 쉽게 많은 토픽에 대한 메시지를 처리할 수 있다.
  • 카프카 스트림, 카프카 다이렉트 스트림 제공
  • KafkaUtils 객체
    • StreamingContext, JavaStreamingContext에서 동작하며 카프카 메지의 DStream을 생성한다.
    • KafkaUtils의 createStream 메소드 사용
  • 카프카 스트림
    • 카프카에서 메시지를 수신하는 방식 (말이 수신, 일종의 polling)
    • (Java)StreamingContext, 주키퍼 호스트 목록, 메시지 consumer 그룹 이름, 토픽 목록, 리시버 스레드 수 등으로 구성
  • 카프카 다이렉트 스트림 (3DVS에서 이 방식 사용)
    • 카프카에서 직접(Direct) 메시지를 읽는 방식
    • 데이터를 직접 읽으므로 데이터를 읽은 다음 복제할 필요 없고, 실패하면 그만
    • 스트림을 연결할 필요 없이 여러개의 리더를 실행할 수 있게 한다.
    • 주키퍼 호스트 목록을 직접 제공하지 않고 카프카 브로커 목록을 직접 넣어준다.
    • 사용할 스레드 개수를 지정할 필요는 없다

여러 입력소스 사용, 클러스터 사이징

• union() 같은 함수를 사용하여 DStream을 연결하는 것이 가능하다. 이런 연산으로 입력 DStream에서 가져온 데이터를 연계할 수 있다.
• 가끔 여러개의 리시버를 사용하는 경우 처리량을 늘리기 위해서 필요하다. / 서버 다른 소스들을 대상으로 서로 다른 데이터를 받아들이기 위해 데이터를 합치기도 한다.
• 각 리시버는 스파크의 익스큐터 내에서 지속적으로 실행 중인 태스크가 되어 돌아간다. 그러므로 리시버는 어플리케이션에 할당된 CPU 코어를 차지하게 된다. 리시버가 사용하는 것 외에 데이터를 처리할 코어도 필요하다.
  • 여러개의 리시버를 윤영하기 위해 최소 리시버 개수만큼의 코어가 필요하며, 추가로 연산을 위한 코어가 필요하다.

24/7 운영 (무중단 운영)

• Spark streaming의 주된 장점은 강력한 장애 내구성을 보장하는 능력이다.
• 입력 데이터가 안정적으로 저장되어 있는 한 Spark streaming은 언제나 올바른 결과를 계산할 수 있다.
• 심지어 드라이버나 작업 노드에 장애가 생겨도 "딱 한번 만에" 콘셉트를 제공한다.
• 24/7로 실행하려면 몇 가지 특수한 설정 필요
  • 안정적인 저장 시스템에 checkpointing 설정, 드러이버 프로그램의 장애 대응에 대해서도 고려, 안정적이지 않은 입력 소스에 대한 고려도 필요
• Checkpointing
  • 체크포인팅은 스파크 스트리밍에서 장애 대응을 위해 구축할 필요가 있는 핵심 매커니즘.
  • 스파크 스트리밍이 복구를 위해 안정적인 파일 시스템에 주기적으로 애플리케이션의 데이터를 저장하게 해준다
  • 두 가지 목적으로 제공
    • 장애 시 재연산이 필요한 상황을 제한한다. 체크포인팅은 상태를 재연산하는 것을 위해 얼마나 많이 되돌릴지를 제어한다.
    • 드라이버에 대해 장애 대응 기능을 제공한다. 프로그램이 죽었을 경우 이를 재실행시키고 복구하도록 요청할 수 있으며, 이 경우 이전에 실행된 체크포인트부터 데이터를 처리하여 복구할 것이다.

• 로컬 파일 시스템의 경로를 사용할 수도 있다.

• 드라이버 장애대응
  • 드라이버 노드의 장애에 대응하는 것은 StreamingContext 생성 시 체크포인팅 디렉토리를 받아들이도록 하는 방법 필요
  • new StreamingContext() 호출 대신, StreamingContext.getOrCreate() 함수 사용
    • 체크포인트 디렉토리가 존재하지 않는다면 StreamingContext 생성, 드라이버가 장애가 나서 재시작 후 이 코드가 실행되면 체크포인트 디렉토리로부터 StreamingContext를 다시 초기화하고 처리를 계속 진행
  • 드라이버 재실행
    • 드라이버 프로그램이 죽으면 자동으로 재실행하는것이 필요
    • 단독 클러스터 매니저의 경우, 드라이버를 제출할 때 --supervise 플래그를 사용하여 스파크가 드라이버를 재시작하도록 할 수 있다.
    • --deploy-mode cluster 지정하여 드라이버가 로컬 머신이 아닌 클러스터 내에서 실행되도록 해줄 필요도 있다.
  • 재실행된 드라이버는 멈춘 지점부터 동작하도록 새로운 익스큐터를 시작할 것이다.
• 작업 노드 장애 내구성
  • 스파크 스트리밍의 작업 노드 장애 대응은 스파크와 같은 방식으로 당작.
  • 외부 소스로부터 받은 모든 데이터는 스파크 자업 노드 간에 복제된다. 이 복제된 입력 데이터는 트랜스포메이션으로 생성된 모든 RDD 가계도를 통하여 입력 데이터의 살아있는 복제본으로부터 유실된 데이터를 시스템이 재연산할 수 있으므로 장애에 대해 내성이 높다.
• 리시버 장애 내구성
  • 리시버가 동작 중인 작업 노드의 장애 내구성은 중요한 이슈
  • 장애에서 스파크 스트리밍은 클러스터의 다른 노드에서 실패한 리시버를 재시작한다.
  • 데이터 유실 가능성이 있는지는 입력 소스의 동작 방식과 리시버의 구현에 따라 다르다.
  • 리시버들에서 보통 보장하는 것들
  • 안정적인 파일 시스템(예, HDFS)에서 읽어 온 모든 데이터는 해당 파일 시스템이 복제 기반이므로 안정적이다. 스파크 스트리밍은 체크포인트로 어느 데이터까지 처리도었는지 기억하고 애플리케이션이 중르면 끝난 이후부터 처리.
  • 리시버 기반의 가크카나 푸시 기반의 플럼 등은 안정적이니 않은 소스에 대해서는 스파크에서 입력 데이터를 복제하지만, 리시버 태스크가 일시 다운되면 데이터 유실 위험이 있다.
  • 스파크 1.2부터 HDFS와 같은 안정적인 파일 시스템에 로깅이 가능해져 드라이버 재시작의 경우에도 데이터 유실 위험은 없어졌다.
  • 모든 데이터가 처리되었다는 것을 확인하기 위한 가장 좋은 방법은 안정적인 입력 소스를 사용하는 것이다.
  • 안정적이지 않은 입력 소스를 갖고 있고, 데이터 유실을 피하고 싶다면
    • spark.streaming.receiver.writeAheadLog.enable을 true로 설정하여 WAS(Write-Ahead Logging)을 활성화해야 한다.
• 작업 처리에 대한 보장
  • 스파크 스트리밍의 작업 노드 장애 내구성에 대한 보장 덕택에 몯ㄴ 트랜스포메이션에 대해 "딱 한번 처리(exactly once)" 컨셉 제공이 가능하다.
  • 출력 연산을 통하여 외부 시스템에 보내질 때는 태스크의 오류로 여러번 실행될 수 있다.
    • 트랜잭션, 멱등연산 등의 설계가 필요하다 -> 그러니까 잘 만들어라...

성능 고려사항

• 스파크 스트리밍 애플리케이션은 몇 가지 특화된 성능 최적화 옵션을 갖고 있다.
• 배치/윈도 크기
  • 어느 정도의 최소 배치가 스파크 스트리밍에 적당한가??? ==> 500ms가 최적의 최소 크기로 판명
  • 가장 좋은 방법은 큰 배치 크기로 시작하여 작은 사이즈로 낮춰 가면서 테스트 진행하여 확인
  • 윈도 연산도 유사한 방식으로 계산하여 간격이 성능에 큰 영향을 미칠 수 있다.
• 병렬화 수준
  • 배치의 처리 시간을 줄이는 일반적인 방법은 병렬화 수준을 올리는 것. 병렬화 수준을 올리는 데는 3가지 방법이 있다.
  • 리시버 개수 늘리기
    • 리시버는 하나의 머신으로 너무 많은데이터를 읽고 분배한다면 병목현상이 나타날 수 있다.
    • 여러 개의 입력 DStream을 만들어 리시버를 추가하고, union을 적용하여 합쳐서 단일 스트림으로 보내주면 된다.
  • 받은 데이터를 명식적으로 재파티셔닝
    • 리시버를 늘리기 힘들다면 DStream.repartition을 사용하여 입력 스트림을 재파티셔닝하여 받은 데이터를 재배치한다.
  • 집합 연산 시 병렬화 개수 늘리기
    • reduceByKey() 같은 함수를 사요할 때 병렬화 정도 지정.
• 가비지 컬렌션과 메모리 사용량
  • 예상치 못한 이슈/문제를 일으키는 부분은 Java의 GC 부분.
  • 자바의 병렬 마크 스윕(Concurrent-Mark-Sweep) 가비지 컬렉터를 사용하여 예측되지 않은 GC에 의한 일시 정지를 최소화할 수 있다.
    • 병렬 마크 스윕 가비지 컬렉터는 전체적으로 자원을 좀 더 사용하지만 일시정지 상태를 줄여준다.
  • GC는 spark.executor.extraJavaOptions의 설정 속에 -XX:+UseConcMarkSweepGC를 추가하여 제어할 수 있다.
    • spark-submit --conf spark.executor.extraJavaOptions=-XX:+UseConcMarkSweepGC App.jar
  • GC의 부담을 줄여 큰 차이를 만들수도 있다.
    • 직렬화 형태로 RDD를 캐싱하는 것 또한 GC 부탐을 줄여준다. => 이는 기본적으로 스파크 스트리밍에 의해 생성된 RDD는 직렬화 성태로 저장되기 때문.
    • 카이로(Kyro) 직렬화를 사용하는 ㄴ것이 캐시된 데이터의 메모리 저장에 사용하는 메모리를 더욱 줄여준다.
  • 스파크는 캐시 또는 영속화된 RDD가 캐시로부터 삭제될지를 사용자가 제어할수 있도록 한다.
    • 기본적으로 스파크는 LRU 캐시를 사용한다. 만약 spark.cleaner.ttl을 설정한다면 스파크는 특정 시간이 자난 RDD를 모두 삭제할 수도 있다.
    • 캐시에서 사용되지 않을 것 같은 RDD를 미리 삭제함으로써 GC의 부담을 줄일 수 있다.