spark streaming에서 중요한 옵션 정리

최근 스파크 스트리밍을 이용해서 실시간 로그 전처리 작업을 진행중이다.

작업을 하면서 문제가 발생했던 부분을 옵션을 수정하면서 해결하였다.

1. spark.streaming.kafka.maxRatePerPartition

-> 초당 파티션의 처리량 제한

메시지가 한번에 대량으로 들어오거나 어떠한 이유로 스트리밍 어플리케이션이 다운되어서 카프카에 대량의 메시지가 쌓여있을 경우 어플리케이션을 재시작하면 대량의 메시지가 한번에 받아들여져서 설정된 executor-memory 이상의 데이터가 몰리면 OOM에러가 발생하게 된다. 만약에 상황에 대비해서 자원할당을 무작정 크게 할 수는 없기 때문에 옵션을 찾아보았는데 maxRatePerPartition 옵션을 발견하였다. 이 옵션은 초당 파티션에 들어오는 레코드의 양을 제한하는 것이다. 적절한 executor-core / memory와 maxRatePerPartiton 옵션을 이용하면 효율적으로 자원을 활용할 수 있다.

예) 아래는 nginx로그가 쌓여있는 상황에서 logstash를 수시간 중단시켰다가 재시작하여 한번에 대량의 메시지를 카프카에 전송한 경우이다. 이 때 maxrate를 설정하지 않을 경우 스파크 스트리밍 어플리케이션에서 OOM이 발생할 수 있다.

2. spark.streaming.backpressure.enabled

-> 클러스터의 작업 처리시간에 따른 동적으로 입력 레코드 수의 조정

다양한 이유(하둡 클러스터 네트웤, 서버 부하등)로 클러스터에서 레코드 처리가 굉장히 느려져서 배치 간격보다 처리 시간이 길어져 레코드가 무한정 쌓이는 경우가 발생할 수 있다. backpressure 옵션을 true로 설정하면 클러스터의 배치간격과 처리시간에 따라 받아오는 레코드드의 수를 동적으로 조절되어 받아들인다.  

예) 아래는 하둡 클러스터의 네트워크 문제로 인하여 스트리밍 어플리케이션이 영향을 받아 지연이 발생한 상태이다. 지연이 발생되자 input size가 줄었다가 지연이 해소되자 다시 늘어나는 것을 볼 수 있다.