블로그 이미지
평범하게 살고 싶은 월급쟁이 기술적인 토론 환영합니다.같이 이야기 하고 싶으시면 부담 말고 연락주세요:이메일-bwcho75골뱅이지메일 닷컴. 조대협


Archive»


 
 

클러스터 상에서 하둡 배포 아키텍쳐


조대협 (http://bcho.tistory.com)


오늘 빅데이타 관련 교육을 받다가 클라우드 상에서 하둡 클러스터 활용에 대한 영감을 받은 부분이 있어서 정리해보고자 한다. 하둡의 경우에는 On-prem 환경에 적절하게 디자인이 된 오픈 소스라서, 이걸 클라우드에서 사용할 경우에도 on-prem에서 사용하는 형태와 유사하게 사용하는 경우가 많다. 일종의 습관 또는 관성이라고 해야 하나? 인프라가 바뀌면 그 장점에 맞는 아키텍쳐를 선택해야 하는데, 이 부분을 놓치고 있지 않았나 싶다.

Job별 클러스터를 생성하는 아키텍쳐

job을 수행하는 방법을 보면, 일반적으로 On-Prem에서 사용하는 방법은 하나의 하둡 클러스터에 Job을 실행하고 Job이 끝나면 다음 Job을 수행하는 방식을 사용한다.



이러한 경험(습관)때문인지 보통 클라우드 상에 하둡 클러스터를 만들어놓고 그 클러스터에 On-prem과 마찬가지 방식으로 Job을 넣고 그 작업이 끝나면 같은 클러스터에 Job을 넣는 방식을 사용한다.

그런데 클라우드의 특성상 자원이 거의 무한대인데, 여러개의 Job을 하나의 클러스터에서 순차적으로 실행할 필요가 있을까?


Job별로 클러스터를 생성해서 연산을 하고 Job이 끝나면 클러스터를 끄는 모델을 사용하면, 동시에 여러 작업을 빠르게 처리할 수 있지 않을까?



이 경우 문제가 될 수 있는 것이 같은 데이타셋을 가지고 여러가지 Job을 실행할 경우 기존의 On-Prem의 경우에는 동시에 같은 데이타셋에 접근할 경우 오버로드가 심하기 때문에, 어렵지만 클라우드의 경우에는 바로 HDFS를 사용하는 것보다는 GCS와 같은 클라우드 스토리지를 사용하는데, 여기서 오는 장점이 이런 클라우드 스토리지에 저장된 데이타는 동시에 여러 하둡 클러스터에서 접근이 가능하다.


이러한 아키텍쳐를 생각하게된 이유는 구글 클라우드의 특성 때문인데, 구글 클라우드의 과금 모델은 분당 과금 모델을 사용한다. 그래서 Job별로 클러스터를 전개했다가 작업이 끝난 다음에 끄더라도 비용 낭비가 없다. 시간당 과금인 경우에는 클러스터를 껐다켰다 하면 비용 낭비가 발생할 수 있기 때문에 바람직하지 않을 수 있다. 예를 들어서 1시간 10분이 걸리는 작업과 20분이 걸리는 작업을 각각 다른 클러스터에서 돌린다면 2시간, 그리고 1시간 비용이 과금되기 때문에 총 3시간이 과금된다.

그러나 구글의 분당 과금을 적용하면 총 1시간30분이 과금되기 때문에 비용이 1/2로 절감된다.


또한 구글 클라우드의 경우 하둡 클러스터 배포시간이 통상 90초이내로 빠르기때문에, Job마다 클러스터를 생성하고 끈다고 해도, 실행 시간이 크게 늘어나지 않는다.


클러스터별로 인프라를 최적화

만약에 위에처럼 Job별로 클러스터를 나눈다면, Job 의 특성에 맞춰서 클러스터의 인스턴스 수나 구조(?)를 변경하는 것도 가능하다.

즉 많은 컴퓨팅 작업이 필요한 Job이라면 클러스터에 많은 인스턴스를 넣고, 그렇지 않다면 인스턴스 수를 줄일 수 있다.




조금 더 나가면, 상태 정보 같이 HDFS에 저장하는 데이타가 적을 경우 워커노드가 중간에 정지되더라도 전체 연산에 크게 문제가 없을 경우에는 Preemptible VM과 같이 가격은 싸지만 저장 기능이 없는 VM의 수를 늘려서 전체 비용을 낮출 수 있다.




여기서 한걸음 더 나가면, 구글 클라우드의 인스턴스 타입중의 하나가 Cutomizable 인스턴스로 CPU와 메모리 수를 마음대로 조정할 수 있다. 즉 Job이 메모리를 많이 사용하는 Spark과 같은 Job일 경우 클러스터를 구성하는 인스턴스에 CPU보다 메모리를 상대적으로 많이 넣고 구성하는 것들이 가능하다.


아키텍쳐 설계도 의미가 있었지만, 그동안 간과하고 있었던 점이 기술이나 인프라가 바뀌면 활용 하는 방법도 다시 고민해야 하는데, 그동안의 관성 때문인지 기존 시스템을 활용하는 방법을 그대로 의심 없이 사용하려 했다는 것에 다시한번 생각할 필요가 있었던 주제였다.






데이타 스트리밍 분석 플랫폼 Dataflow 개념 잡기 #2/2

(트리거, 이벤트 타임, 워터마크 개념)


조대협 (http://bcho.tistory.com)


앞글 http://bcho.tistory.com/1122 에 의해서 Dataflow에 대한 개념에 대해서 계속 알아보자

트리거

윈도우와 더블어서 Dataflow 프로그래밍 개념중에서 유용한 개념중의 하나가 트리거이다. 트리거는 처리중인 데이타를 언제 다음 단계로 넘길지를 결정하는 개념이다. 특히 윈도우의 개념과 같이 생각하면 좋은데, 윈도우는 일반적으로 윈도우가 종료되는 시간에 그 데이타를 다음 Transform으로 넘기게 된다.


그런데 이런 의문이 생길 수 있다. “윈도우의 크기가 클때 (예를 들어 한시간), 한시간을 기다려야 데이타를 볼 수 있는 것인가? 그렇다면 한 시간 후에 결과를 본다면 이것을 실시간 분석이라고 할 수 있는가?”

그래서 여기서 트리거의 개념이 나온다.

예를 들어 한시간 윈도우가 있더라도, 윈도우가 끝나지 않더라도 현재 계산 값을 다음 Transform으로 넘겨서결과를 볼 수 있는 개념이다. 1분 단위로 트리거를 걸면 1분 결과를 저장하고, 2분째도 결과를 저장하고, 3분째도…. 60분째에도 매번 결과를 업데이트 함으로써, 윈도우가 종료되기 전에도 실시간으로 결과를 업데이트 할 수 있게 된다.


트리거의 종류

그렇다면 이러한 트리거는 앞에서 언급한 시간 단위의 트리거만 있을까? Dataflow는 상당히 여러 종류의 트리거를 지원한다.


  • Time trigger (시간 기반 트리거) : 시간 기반 트리거는 일정 시간 주기로 트리거링을 해주는 트리거 이다. 1분 단위, 1초 단위 같이 일정 주기를 지정하거나, “윈도우 시작후 2분후 한번과 윈도우 종료후 한번"과 같이 절대적인 시간을 기준으로도 정의가 가능하다.

  • Element Count (데이타 개수 기반 트리거) : 다음은 개수 기반인데, 예를 들어 “어떤 데이타가 100번 이상 들어오면 한번 트리거링을 해라” 또는 “매번 데이타가 100개씩 들어올때 마다 트리거링을 해라" 라는 형태로 정의가 가능하다.

  • Punctuations  (이벤트 기반 트리거) : Punctuations는 엄밀하게 번역하면 “구두점" 이라는 의미인데, 구두점 처럼 특정 데이타가 들어오는 순간에, 트리거링을 하는 방법이다.

트리거 조합

이러한 트리거는 하나의 트리거 뿐 아니라, 여러개의 트리거를 동시에 조합하여 사용이 가능하다.

  • AND : AND 조건으로 두개의 트리거의 조건이 만족해야 트리거링이 된다. 예를 들어, Time Trigger가 1분이고, Element Count 트리거가 100개이면, 윈도우가 시작된 1분 후에, Element Count가 100개가 되면 트리거링이 된다.

  • OR : OR 조건으로 두개의 트리거의 조건 중 하나만 만족하면 트리거링이 된다.

  • Repeat : Repeat는 트리거를 반복적으로 수행한다. Element Count 트리거 10개를 반복으로 수행하면, 매 10개 마다 트리거링이 된다. Time 트리거를 1분 단위로 반복하면, 매 1분 마다 트리거링이 된다.

  • Sequence : Sequence 트리거는 등록된 트리거를 순차적으로 실행한다. Time 트리거 1분을 걸고 Element count 트리거 100개를 걸면, 윈도우 시작후 1분 후 트리거링인 된후, 그 후 부터 Element 가 100개 들어오면 두번째 트리거링이 발생하고 트리거링이 종료 된다.


트리거 결과의 누적

그러면 트리거링이 될때 마다 전달 되는 데이타는 어떻게 될까라는 질문이 나올 수 있는데. 무슨 이야기인가 하면 윈도우 내에서 트리거가 발생할때, 이전 데이타에 대한 처리를 어떻게 할것인가이다.


데이타가 A,B,C,D,E,F 가 들어왔다고 가정하자. 트리거가 C 다음 발생했다고 했을때, 윈도우가 끝난 F에는 어떤 값이 리턴이 될까?

첫번째 트리거링에는 당연히 A,B,C 가 전달된다.

윈도우가 끝나면 A,B,C,D,E,F 가 전달되는 것이 맞을까 아니면 트리거링 된 이후의 값인 D,E,F 만 전달되는 것이 맞을까?

맞는 건 없고, 옵션으로 지정이 가능하다.

  • Accumulating
    Accumulating은 트리거링을 할때 마다 윈도우 내에서 그때까지의 값을 모두 리턴한다.

  • Discarding
    트리거링 한 후에, 이전 값은 더이상 리턴하지 않고, 그 이후 부터 다음 트리거링 할때까지의 값만을 리턴한다.

예를 들어서 보자


다음과 같은 윈도우가 있고, 3번, 23번, 10번에서 트리거링이 된다고 했을때,

Accumulating mode의 경우

  • 첫번째 트리거 후 : [5,8,3]

  • 두번째 트리거 후 : [5,8,3,15,19,23]

  • 세번째 트리거 후 : [5,8,3,15,19,239,13,10]

와 같이 값이 반환되고

Discarding mode의 경우

  • 첫번째 트리거 후 [5,8,3]

  • 두번째 트리거 후 [15,19,23]

  • 세번째 트리거 후 [9,13,10]

이 반환된다.

데이타 지연에 대한 처리 방법

실시간 데이타 분석은 특성상 데이타의 전달 시간이 중요한데, 데이타는 모바일 클라이언트 등에서 인터넷을 통해서 데이타가 서버로 전송되는 경우가 많기 때문에, 데이타의 실제 도달 시간이 들쭉날쭉 하다. 이러다 보니 데이타의 도착 순서나 지연등이 발생하는데, 이에 대한 처리가 필요하다. 먼저 데이타 도달 시간의 개념을 이해하려면, 이벤트 타임과 프로세싱 타임의 개념을 먼저 이해해야 한다.

이벤트 타임과 프로세싱 타임

모바일 단말에서 다음과 같이 A,B,C,D의 데이타를 1시1초, 1시2초,3초,5초에 보냈다고 하자.


서버에 도착해서 Dataflow에 도착하는 시간은 물리적으로 서버와 단말간의 거리 차이가 있기 때문에 도착 시간은 단말에서 데이타가 발생한 시간보다 느리게 되며, 또한 각 단말의 위치나 단말이 연결되어 있는 네트워크 상황이 다르기 때문에 순차적으로 도착하는 것이 아니라, 늦게 보낸 데이타가 더 빨리 도착할 수 도 있다.

아래 그림을 보면 A데이타는 1시1초에 단말에서 생성되었지만 서버에 도착한 시간은 1시2초가 된다. C,D의 경우, 순서가 바뀌어서 도착하였다.



이렇게 실제로 데이타가 발생한 시간을 이벤트 타임, 그리고 서버에 데이타가 도착한 시간을 프로세싱 타임이라고 정의한다.


이 프로세싱 타임은 네트워크 상황이나 데이타에 크기에 따라 가변적으로 변하기 때문에, 이벤트 타임과 프로세싱 타임의 상관 관계를 그래프로 표현해보면 다음과 같아진다.


가장 이상적인 결과는 이벤트 타임과 프로세싱 타임이 동일한 것이겠지만 불가능하고, 위의 그림처럼 이벤트 타임보다 프로세싱 타임이 항상 늦게 되고, 이벤트 타임과 프로세싱 타임의 차이는 매 순간 다르게 된다.

워터 마크 (Water Mark)

이렇게 위의 그림과 같이 실제 데이타가 시스템에 도착하는 시간을 예측 하게 되는데, 이를 워터 마크라고 한다. 위의 그림에서 “실제 처리 그래프"로 표시되는 부분을 워터마크라고 생각하면 된다. 이 예측된 시간을 기반으로 윈도우의 시스템상의 시작 시간과 종료 시간을 예측 하게 된다.

지연 데이타 처리 방법

윈도우 처리 관련해서, 실제 발생한 시간과 도착 시간이 달라서, 처리 시간내에 못 들어오는 경우가 발생할 수 있다. 아래 그림을 보면, 실제 윈도우는 1시1초~1시6초까지의 데이타를 처리하기를 바라고 정의했을 수 있는데, 시스템에서는 이 윈도우의 값이 프로세싱 타임 기준으로 (워터 마크를 기준으로 연산함) 1시2초~1시6초에 도착하기를 기대하고 있는데, 데이타 C의 경우에는 기대했던 프로세싱 타임에 도착하지 않았기 때문에 이 데이타는 연산에서 누락될 수 있다.



비단 늦게 도착한 데이타 뿐만 아니라, 시스템이 예측한 프로세싱 타임 보다 일찍 데이타가 도착할 수 있는데, 이런 조기 도착한 데이타와 지연 도착한 데이타에 대한 처리는 어떻게 해야 할까?

Dataflow에서는 이런 조기 도착이나 지연 데이타에 대한 처리 메카니즘을 제공한다.

윈도우를 생성할때, withAllowedLateness라는 메서드를 사용하면, 늦게 도착하는 데이타에 대한 처리 기간을 정의할 수 있다.


PCollection<String> items = ...;

 PCollection<String> fixed_windowed_items = items.apply(

   Window.<String>into(FixedWindows.of(1, TimeUnit.MINUTES))

         .withAllowedLateness(Duration.standardDays(2)));

https://cloud.google.com/dataflow/model/windowing#managing-time-skew-and-late-data


위의 예제는 1분 단위의 Fixed Window를 정의하고, 최대 2일까지 지연 도착한 데이타 까지 처리할 수 있도록 정의한 예제이다.


지금까지 간단하게 dataflow를 이용한 스트리밍 데이타 처리의 개념에 대해서 알아보았다.

데이타 스트리밍 분석 플랫폼 Dataflow 개념 잡기 #1/2


조대협 (http://bcho.tistory.com)


실시간 데이타 처리에서는 들어오는 데이타를 바로 읽어서 처리 하는 스트리밍 프레임웍이 대세인데, 대표적인 프레임웍으로는 Aapche Spark등을 들 수 있다. 구글의 DataFlow는 구글 내부의 스트리밍 프레임웍을 Apache Beam이라는 형태의 오픈소스로 공개하고 이를 실행하기 위한 런타임을 구글 클라우드의 DataFlow라는 이름으로 제공하고 있는 서비스이다.


스트리밍 프레임웍 중에서 Apache Spark 보다 한 단계 앞선 개념을 가지고 있는 다음 세대의 스트리밍 프레임웍으로 생각할 수 있다. Apache Flink 역시 유사한 개념을 가지면서 Apache Spark의 다음 세대로 소개 되는데, 이번글에서는 이 DataFlow에 대한 전체적인 개념과 프로그래밍 모델등에 대해서 설명하고자 한다. 스트리밍 데이타 처리에 대한 개념은 http://bcho.tistory.com/1119 글을 참고하기 바란다.

개념 소개

dataflow에 대해서 이해하기 위해서 프로그래밍 모델을 먼저 이해해야 하는데, dataflow의 프로그래밍 모델은 얼마전에 Apache에 Beam이라는 오픈 소스 프로젝트로 기증 되었다. Apache Spark이나, Apache Flink와 유사한 스트리밍 처리 프레임웍이라고 생각하면 된다. dataflow는 이 Apache beam의 프로그래밍 모델을 실행할 수 있는 런타임 엔진이라고 생각하면 된다. 예를 들어 Apache beam으로 짠 코드를 Servlet이나 Spring 코드라고 생각하면, dataflow는 이를 실행하기 위한 Tomcat,Jetty,JBoss와 같은 런타임의 개념이다.


먼저 dataflow의 개념을 이해해보도록 하자. 아래 그림은 dataflow에 대한 컨셉이다.


데이타가 들어오면, Pipeline IO에서 데이타를 읽어드린다. 읽어드린 데이타는 PCollection이라는 데이타 형으로 생성이 되고, 이 PCollection 데이타는 여러개의 중첩된 PTransform을 통해서 변환 및 가공이 된다. 가공이 끝난 결과는 마지막으로 Pipeline IO의 Output을 통해서 데이타 저장소 (빅쿼리나 파일등)에 저장이 된다.  이 Pipeline IO에서 부터 PTransform을 걸친 일련의 프로세싱 과정을 Pipeline이라고 한다.


예를 들어 설명해보자, 문자열을 입력 받은 후에, 문자열에서 단어를 추출하여, 각 단어의 개수를 세어 주는 파이프라인이 있다고 하자.


첫번째 실행에서 “Hello my daddy”라는 문자열이 입력되었다. 첫번째 Transform인 Extract words Transform을 거치면서, “Hello my daddy” 라는 문자열은 “Hello”, “my”, “daddy” 라는 각각의 단어로 쪼게진다. 다음으로 Count Element 라는 Transform에 의해서, 각 단어의 수를 세어서 저장한다. “Hello”는 1번, “my”는 1번, “daddy”는 1번 의 값이 저장된다.


두번째 실행에서 “Hello my bro” 라는 문자열이 들어오면, Extract words 에 의해서 “Hello”, “my”, “bro”라는 각각의 단어로 쪼게지고, Count Element Transform에서 이전에 세어놓은 단어의 수와 합산하여 계산이 된 결과가 저장이 된다. “Hello”는 이전에 한번 카운트가 되었고 이번에도 들어왔기 때문에, 2가 되고, 같은 원리로 “my”라는 단어의 카운트도 2가된다. “bro” 라는 단어는 이번에 처음 들어왔기 때문에 새 값으로 1로 저장된다.




세번째 “Hello my mom” 이라는 문자열이 들어오면 앞의 두개의 문자열과 마찬가지로 간 단어로 쪼게진 다음 Count Element에 의해서 각 단어의 수가 카운트되어 기존의 값과 누적 합산된다. 모든 데이타를 다 읽어서 처리가 끝나면, 저장된 결과를 Pipeline IO를 통해서 파일에 그 결과를 쓰게 된다.

배치와 스트리밍 처리

dataflow는 위에서 설명한 파이프라인의 개념을 배치와 스트리밍 처리 두가지 개념 모두로 지원해서 처리가 가능하다. 데이타가 파일과 같이 이미 쓰여지고 더 이상 증가나 수정이 되지 않은 데이타에 대해서는 일괄로 데이타를 읽어서 결과를 내는 배치 처리가 가능하고, 계속해서 들어오고 있는 데이타 (트위터 피드, 로그 데이타)는 스트리밍으로 처리가 가능하다.

윈도우의 개념

배치 처리야, 데이타 처리가 모두 끝난 후에 결과를 내보낸다고 하지만, 그렇다면 스트리밍 데이타는 계속해서 데이타가 들어오고 있는데, 언제 결과를 내보내야 할까?

개별 데이타를 변환해서 저장하는 경우에야, 개별 데이타 처리가 끝난후에 각각 하나씩 저장한다고 하지만, 위와 같이 들어오는 데이타에서 특정데이타 들에 대한 합이나 평균과 같은 처리를 하는 경우 어느 기간 단위로 해야 할까? 스트리밍 처리에서는 이러한 개념을 다루기 위해서 윈도우라는 개념을 사용한다.


예를 들어, “1시~1시10분까지 들어온 문자열에 대해서 문자열에 들어 있는 각 단어의 수를 카운트해서 출력해주는 기능" 이나, 또는 “매 5분 단위로 현재 시간에서 10분전까지 들어온 문자열에 대해서 각 단어의 수를 카운트 해서 출력 해주는 기능" 과 같이 작은 시간 기간의 단위를 가지고 그 기간 단위로 계산 하는 방법이며, 이 시간 단위를 윈도우(Window)라고 한다.


Fixed Window (고정 크기 윈도우)

앞의 예에서 1시~1시10분, 1시10분~1시20분 과 같이 고정된 크기를 가지는 윈도우의 개념을 Fixed Window라고 한다.


Sliding Window (슬라이딩 윈도우)

앞의 예에서와 같이 윈도우가 상대적인 시간 (이전 10분까지)의 개념을 가지면서, 다른 윈도우와 중첩되는 윈도우를 슬라이딩 윈도우라고 한다.


그림과 같이 1시10분의 윈도우는 1시 10분의 10분전인 1시에서 부터, 현재 시간 까지인 1시10분까지 값을 읽어서 처리하고 윈도우가 끝나는 시점인 1:10분에 그 값을 저장한다. 윈도우의 간격은 5분 단위로, 1시 15분에는 1시 15분의 10분전인 1시05분 부터 현재 시간인 1시15분까지 들어온 데이타에 대해서 처리를 하고 그 결과 값을 1시15분에 저장한다.

Session window (세션 윈도우)

다음은 세션 윈도우라는 개념을 가지고 있는데, 이를 이해하기 위해서는 먼저 세션의 개념을 먼저 이해해야 한다.

세션이랑 사용자가 한번 시스템을 사용한 후, 사용이 끝날때 까지의 기간을 정의한다. 스트리밍 시스템에서는 사용자 로그인이나 로그 아웃을 별도의 이벤트로 잡는 것이 아니기 때문에, 데이타가 들어온 후에, 일정 시간 이후에 그 사용자에 대한 데이타가 들어오지 않으면, 세션이 종료 된것으로 판단한다.

일반 적인 웹 프로그램에서 HttpSession과 같은 원리인데, 웹 사이트에 접속한 후, Session time out 시간이 지날때 까지 사용자가 별도의 request를 보내지 않으면 세션을 끊는 것과 같은 원리이다.

아래 그림은 세션 윈도우의 개념을 설명하기 위한 윈도우인데, User A와 User B의 데이타가 들어오고 있다고 하자.


그리고 세션 타임 아웃이 10분으로 정의했다. 즉 같은 사용자에 대해서 데이타가 들어온 후, 10분 내에 추가 데이타가 들어오지 않으면 세션이 종료 된것으로 판단한다.


User A는 1:00 에 첫 데이타가 들어와서1:00~1:10 사이에 두번째 데이타가 들어왔고, 1:10~1:20 사이에 세번째 데이타가 들어온 후, 네번째 데이타는 10분이 지난 후에 들어왔다. 그래서 1:00~1:20 까지가 하나의 세션이 되고, 이것이 User A에 대한 1:00~1:20의 세션 윈도우가 된다. 네번째 데이타 부터는 새로운 윈도우로 처리가 되는데, 1:40~1:50 사이에 다섯번째 데이타가 도착한후, 그 이후로 도착하지 않았기 때문에 이게 두번째 윈도우가 되고, 1:30~1:50의 시간 간격을 가지는 User A의 두번째 윈도우가 된다.

각 윈도우의 값은 User A의 1:00~1:20 윈도우의 값은 (1+1+1)로 3이 되고, 두번째 윈도우인 1:30~1:50 윈도우는 (2.5+1)로 3.5가 된다.


User B는 1:10에 데이타가 들어오고, 10분 후인 1:20까지 데이타가 들어오지 않고 그 이후 1:30 분에 두번째 데이타가 들어왔기 때문에, 1:10~1:10 길이의 첫번째 세션 윈도우가 생성된다. 다음 으로 1:30분에 데이타가 들어왔기 때문에 두번째 세션 윈도우를 생성하고, 2:00까지 계속 데이타가 들어오다가 멈추고 2:10까지 새로운 데이타가 들어오지 않았기 때문에 1:30~2:00 까지 두번째 윈도우로 취급한다.


이 Session Window는 앞서 언급한 Fixed Window나, Sliding Window와는 다르게, User A, User B와 사용자 단위와 같이 어떤 키에 따라서 개별적으로 윈도우를 처리 한다.  즉 Session Window는 User A나 USer B처럼 특정 키에 종속된 윈도우만을 갖는다.


반대로 Fixed Window나 Sliding Window는 키단위의 윈도우가 아니라 그 시간 범위내에 들어 있는 모든 키에 대한 값을 처리한다..

Fixed Window의 경우에는 30분 사이즈를 갖는 윈도우라고 하면 아래 그림과 같이


1:00~1:30 윈도우는 User A의 값 = (1+1+1) 과 User B의 값 1을 합쳐서 총 4가 되고

1:30~2:00 윈도우는 User A값 = (2.5+1)과 User B의 값 = (2+2+2) 를 합쳐서 9.5가 된다.


Sliding Window의 경우에는 길이가 30분이고, 주기가 20분인 Sliding 윈도우라고 할때,


1:00~1:30, 1:20~1:50, 1:40~2:00 3개의 Sliding 윈도우가 생성된다.

1:00~1:30 윈도우는 User A의 값=(1+1+1)과 User B의 값 1을 합산하여 4가 되고

1:20~1:50 윈도우는 User A의 값 = (2.5+1)과 User B의 값 =(2+2)를 합산하여 7.5가 된다.

1:40~2:00 윈도우는 User A의 값 = (2.5+1)과 User B의 값 (2+2)를 합산하여 7.5가 된다.



스파크 성능이 안나오면, 우리 회사 데이타팀 팀장왈. 먼저 파이썬으로 짰는지 확인 부터 해보라길래, 파이썬과 스칼라로 만들어진 스파크 성능 차이가 얼마나 나는지 찾아봤더니 다음과 같은 수치가 나왔다.


http://emptypipes.org/2015/01/17/python-vs-scala-vs-spark/ (원본 출처)


일단 스파크를 할려면 스칼라는 필수인듯 하다. 

간단한 프로토타입핑등에는 파이썬을 사용할 수 있겠지만 결국 프로적션은 스칼라로 최적화해야 할듯.

근데. 자바대 스칼라 성능 비교는 없네


Apache Spark(스파크) - RDD Persistence (스토리지 옵션에 대해서)


조대협 (http://bcho.tistory.com)


Spark Persistence에 대해서


앞에 글에서 Spark RDD가 메모리에 상주 되는 방법에 대해서 간략하게 언급했는데, 다시 되 짚어 보면 Spark의 RDD는 filter() 등. 여러  Transformation Operation을 실행하더라도  Transformation  단계가 아니라 Action이 수행되는 단계에 로드된다고 설명하였다.


그리고, 매번 해당 RDD가 Action으로 수행될 때마다 다시금 소스에서 부터 다시 로드되서 수행된다고 했는데, 그렇다면 매번 로드 해서 계산하여 사용하는 것이 아니라, 저장해놓고 사용 하는 방법이 무엇이 있을까?


스파크에서는 RDD 를 저장해놓고 사용하는 기능으로 persist()와 cache() 라는 두 가지 오퍼레이션을 지원한다.

스파크는 RDD를 저장함에 있어서,  메모리와 디스크 두 가지 영역을 사용하며, 옵션에 따라서 RDD의 저장 영역을 지정할 수 있다.


기본 디폴트는 메모리에 저장하고, 옵션으로 디스크를 지정할 수 있다. 디스크를 지정하면, 메모리에서 모지란 부분을 마치 swapping 하듯이 디스크에 데이타를 저장한다.


아래 옵션 참고 




출처 : https://spark.apache.org/docs/latest/programming-guide.html#rdd-persistence


여기에 특이한 점은, 메모리나 디스크에 저장할때, RDD를 RAW (원본 형식)으로 저장할 것인지 자바의 Serialized 된 형태로 저장할 지를 선택할 수 있다. ( Serealized 된 형태로 저장하기 MEMORY_ONLY_SER, MEMORY_AND_DISK_SER) 이렇게 저장하면, 메모리 사용량은 더 줄일 수 있지만, 반대로 저장시 Serizalied하는 오버로드와, 읽을때 De-Seriazlie 하는 오버로드가 더 붙어서 CPU 사용량은 오히려 증가하게 된다.


아래 데이타는 http://sujee.net/2015/01/22/understanding-spark-caching/#.VWcOh1ntlBc 의 데이타 긴데,

Serialized 로 저장하는 경우, 최대 약 4배 정도의 메모리 용량을 절약할 수 있으나, 반대로, 처리 시간은 400배 이상이 더 들어간다.

 









 


다음으로, 특이한 옵션중에 하나가 OFF_HEAP 이라는 옵션인데, 스파크는 JVM 상에서 동작하기 때문에, 스파크가 저장하는 메모리란 JVM 상의 메모리를 뜻한다. JVM  특성상 Garbage collection 에 의한 성능 제약을 받을 수 있으며 또한 별도로 서로 복제가 되지 않기 때문에, (기본 옵션의 경우에만), 안정적인 서비스를 원할 경우에는 별도의 복제 옵션을 선택해야 한다.

이런 문제를 해결하기 위한 다른 옵션으로는 JVM 내에 데이타를 저장하는 것이 아니라, 별도의 JVM 외의 메모리 공간에 데이타를 저장하는 방식이 OFF_HEAP 이라는 옵션이다. 아직 안정화 되지는 않았지만, http://tachyon-project.org/ 이라는 메모리 클러스터를 이용하여, 서로 복제가 가능한 외부 메모리 클러스터에 저장하는 방식으로, JVM  상 메모리 보다는 성능이 약간 떨어지지만, 디스크보다는 빠르며, 큰 메모리 공간을 장애 대응에 대한 상관 없이 (자체 적으로 HA  기능을 제공함) 사용이 가능하다. 

 cf. Redis나  Infinispan등과 같은 메모리 기반의 데이타 그리드 솔루션의 하나인 Hazelcast 역시 JVM 밖의 네이티브 메모리 공간에 데이타를 저장하는 유사한 방식을 사용한다.


Persist vs Cache


그렇다면, persist()와 cache()의 차이점은 무엇인가? cache()는  persist() 에서 저장 옵션을 MEMORY_ONLY로 한 옵션과 동일하다.


저장된 RDD는 메모리나 디스크에서 언제 삭제 되는가?

RDD가 메모리나 디스크에 로드되었다고 항상 로드된 상태로 있는 것이 아니다. 기본적으로 LRU (Least Recently Used)  알고리즘 (가장 근래에 사용되지 않은 데이타가 삭제되는 방식)에 의해서 삭제가 되가나, 또는 RDD.unpersiste() 함수를 호출하면 명시적으로 메모리나 디스크에서 삭제할 수 있다.


언제 어떤 타입의 Peristence옵션을 사용해야 하는가?


가장 좋은 옵션은 디폴트 옵션인  MEMORY_ONLY  옵션이다. 가장 빠르다.

다음으로 메모리가 모자를 경우에는  MEMORY_ONLY_SER 옵션을 이용하면, Seriazlied 된 형태로 저장하기 때문에 메모리 공간은 줄일 수 있으나, 대신 CPU 사용률이 올라간다. 그래도 여전히 빠른 방식이다.

데이타 양이 많을 경우에는 DISK에 저장하는 옵션보다는 차라리 Persist 를 하지 않고, 필요할때 마다 재계산 하는 것이 더 빠를 수 있다.

빠른 응답이 필요한 경우에 Persist 된 데이타에 대한 유실을 방지하려면, replicated storage 옵션을 사용하는 것이 좋다. (MEMORY_ONLY2 등).  다른 스토리지 타입 역시, 장애로 인해서 데이타가 유실되더라도 재계산을 통하여 복구가 가능하지만, 재계산 하는 것 보다는 RDD 의 복제본을 저장해 놓고, 장애시 페일오버 하는 것이 빠르기 때문에, 빠른 응답시간을 요구로 하는 웹 애플리케이션의 경우 이 스토리지 타입이 유리하다. (단, 메모리 사용량은 복제본을 저장하는데도 사용되기 때문에 상대적으로 일반 스토리지 옵션에 비해서 메모리 여유가 적다.)



참고 

Learning Spark

Spark document - https://spark.apache.org/docs/latest/programming-guide.html#rdd-persistence


참고 - 이 글은 제가 스파크를 혼자 공부하면서 문서만을 통해서 정리한 글이기 때문에, 실무적인 경험이 많이 녹아 들어 있지 않습니다. 

Apache Spark RDD 이해하기 #2


스파크에서 함수 넘기기 (Passing function to Spark)


조대협 (http://bcho.tistory.com)


Passing function
스파크는 개발자가 비지니스 로직을 함수로 정의한 후에, 이 함수를 스파크 클러스의 노드들로 보내서 수행할 수 있다. 스파크가 지원하는 프로그래밍 언어에 따라서, 이 함수를 넘기는 방법이나 특성이 다른데, 여기서는 Python을 이용하는 방법에 대해서 설명한다. (스칼라는 아직 공부를 못해서)

파이썬으로 함수 넘기기
파이썬으로 함수를 넘기는 방법은 크게 3가지가 있다.
  • 람다 표기법을 이용하는 방식
  • 모듈 상의 탑레벨 함수(Top-level function)
  • 파이썬 클래스 안에서 함수를 정의하여, 클래스 안에서 그 함수를 스파크로 넘기는 방식
각각의 방식에 대해서 살펴보도록 하자.

첫번째로 람다 표기법을 사용하는 방식이다.  람다 표기법이란, 함수를 정의하는데, 있어서 함수명이 없이 간략하게 함수의 기능만을 저장하는 표기법으로, 주로 간단한 결과를 구현할때, 코딩을 간결하게 (양이 적게) 표현하고자 하는데 사용한다.
다음은 필터 연산에서 함수를 람다 표기법으로 정의해서 넘기는 예제이다. “l” 이라는 RDD에 Apache 라는 문자열이 있는지 없는지를 행마다 체크하는 로직이다.


같은 로직을 함수를 정의해서 함수 자체로 넘길 수 가 있는데, 위의 람다 표현식으로 된 함수를 containsApache라는 함수로 정의하여 필터로 넘기는 예제이다.

마지막으로는 파이썬 클래스에서 클래스의 함수를 넘기는 방식이다.
다음은 MyClass 클래스를 정의한 다음. 클래스를 생성할때, filter에 사용할 문자열을 받은후, count라는 메서드에서, 그 문자열로 필터링을 한 후, 라인 수를 count하여 리턴하는 방법이다.


이 때 주의할 점은 self.query를 filter에 인자로 넘겼는데, 이 경우에 스파크로 넘어가는 것은 self.query 변수 내용 하나가 아니라, 이 객체 전체가 스파크로 넘어가게 된다. 동작상에는 문제가 없지만, 전체 객체가 스파크로 넘어가기 때문에 메모리 사용률이 많아지고, 전체 객체를 넘기는 과정 역시 인자만 넘기는 방식에 비해서 상대적으로 시간이 많이 걸리기 때문에 좋은 방법은 아니다. 
이런 문제를 피하는 방법은 클래스내에서 무슨 값을 넘길때는 self.xxx식으로 스파크에 넘기는 것이 아니라, 그 값을 복사하여 넘기는 방법을 사용하면 된다. 예를 들어서 위의 예제의 경우에는 아래와 같이 변경하면 된다.

즉 스파크에 self.query를 넘기는 것이 아니라 이 값은 로컬 변수인 x에 x=self.query로 저장한 후, 스파크에는 이 x 값을 넘기게 되면, 실제 모든 객체가 스파크에 전달되지 않고, 이 로컬 변후 x 만 넘어가기 때문에 메모리가 과 사용되는 것을 예방할 수 있다. 







Spark의 전체적인 스택 구조

조대협 (http://bcho.tistory.com)

스파크의  전체적인 스택 구조를 보면 다음과 같다.





  • 인프라 계층 : 먼저 스파크가 기동하기 위한 인프라는 스파크가 독립적으로 기동할 수 있는 Standalone Scheudler가 있고 (그냥 스팍만 OS위에 깔아서 사용한다고 생각하면 된다). 또는 하둡 종합 플랫폼인 YARN 위에서 기동될 수 있고 또는 Docker 가상화 플랫폼인 Mesos 위에서 기동될 수 있다.
  • 스파크 코어 : 메모리 기반의 분산 클러스터 컴퓨팅 환경인 스팍 코어가 그 위에 올라간다. 
  • 스파크 라이브러리  : 다음으로는 이 스파크 코어를 이용하여 특정한 기능에 목적이 맞추어진 각각의 라이브러리가 돌아간다. 빅데이타를 SQL로 핸들링할 수 있게 해주는 Spark SQL, 실시간으로 들어오는 데이타에 대한 리얼타임 스트리밍 처리를 해주는 Spark Streaming, 그리고 머신러닝을 위한 MLib, 그래프 데이타 프로세싱이 가능한 GraphX가 있다.

현재 글에서 설명하고 있는 부분은 먼저 스파크에 대한 기본을 이해하기 위해서 Spark Core 부분을 중점적으로 설명하고 있다.