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


Archive»


 
 


한시간에 만드는 대용량 로그 수집 시스템

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


정정 및 참고 내용

2017.1.24 몇가지 내용을 정정합니다.

https://cloud.google.com/logging/quota-policy 를 보면 스택드라이버 로깅에 쿼타 제한이 초당 500건/계정으로 잡혀있어서. 일반적인 경우는 최대 500 TPS의 성능을 낼 수 있습니다. 그 이상의 성능이 필요하면, 여러 계정을 사용해야 합니다 또는 구글에 별도의 쿼타 증설 요청을 해야 합니다.

하루에, 최대 2천5백만건의 로그를 하나의 프로젝트를 통해서 수집이 가능합니다.


또한 프리티어의 경우에는 한달에 로그를 5GB  까지 수집이 가능한데, 이게 넘으면 로그가 더이상 수집되지 않습니다. 그래서 아래 내용 처럼 빅쿼리로 Export를 해서 로그가 5GB 이상 스택드라이버에 저장되지 않도록 해야 합니다. (차기전에 데이타를 퍼나르는)

애플리케이션 로그 이외에도, VM 로그등도 이 5GB의 용량을 공유하기 때문에, VM 로그등도 차기전에 GCS로 퍼 나르거나 또는 구글 Support 티켓을 통하여 애플리케이션 로그 이외의 로그를 수집하지 않도록 별도 요청해야 합니다. (로그 저장 용량에 대해서 비용을 지불하면, 이런 제약은 없음)


백앤드 시스템에서 중요한 컴포넌트중의 하나가, 클라이언트로 부터 로그를 수집 및 분석하는 시스템이다.

오늘 설명할 내용은 500 TPS (Transaction Per Sec)가 넘는 대용량 로그 수집 및 분석 시스템을  managed 서비스를 이용하여, 쉽고 빠르게 구축할 수 있는 방법에 대해서 소개하고자한다.


일반적인 로그 수집 및 분석 시스템 아키텍쳐

일반적으로 클라이언트에서 로그를 수집하여 분석 및 리포팅 하는 시스템의 구조는 다음과 같다.


  • 앞단의  API 서버가 로그를 클라이언트로 부터 수집하고 데이타를 정재한다.

  • 로그 저장소가 순간적으로 많은 트래픽을 감당할 수 없는 경우가 많기 때문에, 중간에 Message Q를 넣어서, 들어오는 로그를 Message Q에 저장하여 완충을 한다.

  • 이 Message Q로 부터 로그를 Message Consumer가 순차적으로 읽어서 Log Storage에 저장한다.

  • 저장된 로그는 Reporting 툴을 이용하여 시각화 한다.


이런 구조 이외에도 API 서버에서 파일로 로그를 저장한 후,  Fluentd나, LogStash 등의 로그 수집기를 이용하는 방법등 다양한 아키텍쳐가 존재한다.


이런 시스템을 구축하기 위한 일반적인 솔루션들은 다음과 같다.


컴포넌트

솔루션


API 서버

node.js, ruby, php 등 일반적인 웹서버


Message Q

Rabbit MQ와 같은 일반적인 큐
Kafaka 와 같은 대량 큐

AWS SQS나 구글 Pub/Sub 같은 클라우드 큐


Message Consumer

Multi Thread(or Process) + Timer를 조합하여 메세지를 폴링 방식으로 읽어오는 애플리케이션 개발


Log Storage

Hadoop, HBase 와 같은 하둡 제품

Drill,Druid와 같은 SQL 기반 빅데이타 플랫폼

Elastic Search


Reporting

Zeppeline, Jupyter 와 같은 노트북류

Kibana



구조나 개념상으로는 그리 복잡한 시스템은 아니지만, 저러한 솔루션을 모두 배우고, 설치하고 운영하는데 시간이 들고, 각각의 컴포넌트를 구현해야하기 때문에 꽤나 시간이 걸리는 작업이다.


그러면 이러한 로그 수집 및 분석 작업을 클라우드 서비스를 이용하여 단순화 할 수 없을까?

스택 드라이버

스택 드라이버는 구글 클라우드의 모니터링, 로깅 및 애플리케이션 성능 분석등 모니터링 분야에서 다양한 기능을 제공하는 서비스 이다.

그중에서 스택드라이버 로깅은 구글 클라우드나 아마존 또는 기타 인프라에 대한 모니터링과, Apache, MySQL과 같은 써드 파티 미들웨어에 대한 로그 수집 및 모니터링을 지원하는데, 이 외에도, 사용자가 애플리케이션에서 로깅한 데이타를 수집하여 모니터링할 수 있다.



스택 드라이버와 빅쿼리를 이용한 로그 수집 분석 시스템 구현

스택 드라이버 로깅의 재미있는 기능중 하나는 로그 EXPORT 기능인데, 로그 데이타를 구글 클라우드 내의 다른 서비스로 로그 데이타를 내보낼 수 있다.


  • GCS (Google Cloud Storage)로 주기적으로 파일로 로그 데이타를 내보내거나

  • Pub/Sub이나 Big Query로 실시간으로 데이타를 내보낼 수 있다.


그렇다면 스택 드라이버를 통해서 빅쿼리에 로그 데이타를 직접 저장한다면 복잡한 Message Q나, Message Consumer 등의 구현도 불필요하고, 로그 저장도 복잡한 오픈 소스를 이용한 개발이나 운영도 필요 없이, 매니지드 서비스인 빅쿼리를 이용하여 간략하게 구현할 수 있다.

스택 드라이버 로깅을 이용한 로그 수집 시스템 구현


스택 드라이버 애플리케이셔 로깅 기능을 이용하여 클라이언트로 부터 로그를 수집하여 분석하는 시스템의 아키텍쳐를 그려 보면 다음과 같다.




API 서버를 이용하여 클라이언트로 부터 로그를 수집하고, API 서버는 스택 드라이버 로깅 서비스로 로그를 보낸다. 스택 드라이버 로깅은 Export 기능을 이용하여, 수집된 로그를 실시간으로 빅쿼리로 전송한다. 빅쿼리에 저장된 로그는 구글 데이타 스튜디오 (http://datastudio.google.com)이나 제플린, 파이썬 주피터 노트북과 같은 리포팅 도구에 의해서 시각화 리포팅이 된다.

API 서버쪽에서 스택 드라이버 로깅으로 로그를 보내는 부분을 살펴보자

아래는 파이썬 Flask 를 이용하여 로그를 스택 드라이버로 보내는 코드이다.


import uuid

from flask import Flask

from google.cloud import logging


app = Flask(__name__)

logging_client = logging.Client()

tlogger = logging_client.logger(‘my-flask-log’)

slogger = logging_client.logger('struct_log')

@app.route('/')

def text_log():

   logstring = "This is random log "+ str(uuid.uuid4())

   tlogger.log_text(logstring)

   return logstring


@app.route('/slog')

def struct_log():

   struct  = "This is struct log "+ str(uuid.uuid4())

   slogger.log_struct({

               'name':'myterry',

               'text':struct,

               'key' : 'mykey'})      

   return struct


if __name__ == '__main__':

   app.run('0.0.0.0',7001)

   

google.cloud 패키지에서 logging 모듈을 임포트한 다음에, 로깅 클라이언트로 부터

tlogger = logging_client.logger(‘my-flask-log’)

slogger = logging_client.logger('struct_log')

로 각각 “my-flask-log”와 “struct_log”라는 이름을 가지는 logger 둘을 생성한다.

(뒤에서 언급하겠지만, 이 로거 단위로, 로그를 필터링 하거나, 또는 이 로거 단위로 로그 메세지를 다른 시스템으로 export 할 수 있다.)


다음, 로그를 쓸 때는 이 logger를 이용하여 로그를 써주기만 하면 된다.

   tlogger.log_text(logstring)

는 텍스트로 된 한줄 로그를 쓰는 명령이고,

   slogger.log_struct({

               'name':'myterry',

               'text':struct,

               'key' : 'mykey'})  

는 JSON과 같이 구조화된 계층 구조를 로그로 쓰는 방식이다.

이렇게 개발된 로그 수집용 API 서버의 코드는 직접 VM을 만들어서 Flask 서버를 깔고 인스톨 해도 되지만  앱앤진을 사용하면 코드만 배포하면, Flask 서버의 관리, 배포 및 롤백, 그리고 오토 스케일링등 모든 관리를 자동으로 해준다. 앱앤진을 이용한 코드 배포 및 관리에 대한 부분은 다음 문서 http://bcho.tistory.com/1125 를 참고 하기 바란다.

스택 드라이버에서 로그 확인

코드가 배포되고, 실제로 로그를 기록하기 시작했다면 스택 드라이버에 로그가 제대로 전달 및 저장되었는지 확인해보자. 구글 클라우드 콘솔에서 스택 드라이버 로깅으로 이동한 다음 아래 그림과 같이 리소스를 “Global” 을 선택한 후, 앞에 애플리케이션에서 남긴 “my-flask-log”와 “struct-log” 만을 선택해서 살펴보자





다음과 같이 로그가 출력되는 것을 확인할 수 있으며, struct_log의 예를 보면 로그의 내용은 time_stamp  와 프로젝트 정보와 같은 부가 정보와 함께, 애플리케이션에서 남긴 로그는 “jsonPayload” 앨리먼트 아래에 저장된것을 확인할 수 있다.



빅쿼리로 Export 하기

스택 드라이버로 로그가 전달되는 것을 확인했으면, 이 로그를 빅쿼리에 저장해보자. Export 기능을 이용해서 가능한다. 아래와 같이 스택 드라이버 로깅 화면에서 상단의 “CREATE EXPORT”  버튼을 누른다.

다음 리소스 (Global)과 로그 (struct_log)를 선택한 다음에,



Sink Name에 Export 이름을 적고 Sink Service는 BigQuery를 선택한다. 다음으로 Sink Destination에는 이 로그를 저장할 Big Query의 DataSet 이름을 넣는다.

마지막으로 Create Sink를 누르면, 이 로그는 이제부터 실시간으로 BigQuery의 structlog라는 데이타셋에 저장이 되면 테이블명은 아래 그림과 같이 strcut_log_YYYYMMDD와 같은 형태의 테이블로 생성이 된다.




테이블 프리뷰 기능을 이용하여 데이타가 제대로 들어갔는지 확인해보자. 아래와 같이 위의 코드에서 저장한 name,key,text는 테이블에서 jsonPayload.name, jsonPayload.key, jsonPayload.text 라는 필드로 각각 저장이 되게 된다.



빅쿼리는 실시간으로 데이타를 저장할때는 초당 100,000건까지 지원이 가능하기 때문에 이 시스템은 100,000TPS 까지 지원이 가능하고, 만약에 그 이상의 성능이 필요할때는 로그 테이블을 나누면(Sharding) 그 테이블 수 * 100,000 TPS까지 성능을 올릴 수 있다. 즉, 일별 테이블을 10개로 Sharding 하면, 초당 최대 1,000,000 TPS를 받는 로그 서비스를 만들 수 있으며, 이 테이블 Sharding은 빅쿼리 테이블 템플릿을 사용하면 쉽게 설정이 가능하다. (정정 빅쿼리는 100K TPS를 지원하나, 스택 드라이버가 500 TPS로 성능을 제한하고 있음)


이렇게 저장된 로그는 빅쿼리를 지원하는 각종 리포팅 툴을 이용하여 시각화가 가능하다.

시각화 도구는

을 참고하기 바란다.


이렇게 간단하게, 코드 몇줄과 설정 몇 가지로 100,000 500 TPS 를 지원하는 로그 서버를 만들어 보았다.

스택 드라이버를 이용한 로그 분석 수집 시스템의 확장

이 외에도 스택 드라이버는 빅쿼리뿐 아니라 다른 시스템으로의 연동과 매트릭에 대한 모니터링 기능을 가지고 있어서 다양한 확장이 가능한데, 몇가지 흥미로운 기능에 대해서 살펴보도록 하자.


실시간 스트리밍 분석 및 이벤트 핸들링

스택 드라이버 로깅의 Export 기능은, 하나의 로그를 여러 연동 시스템으로 Export를 할 수 있다. 앞에서는 빅쿼리로 로그를 Export 하였지만, 같은 Log를 Dataflow에 Export 하였을 경우, 로그 데이타를 실시간 스트림으로 받아서, 실시간 스트리밍 분석이 가능하다.


구글 데이타 플로우에 대한 설명은 아래 링크를 참고하기 바란다.


또는 실시간 스트리밍이 아니라, 로그 메세지 하나하나를 받아서 이벤트로 처리하고자 할 경우, Pub/Sub 큐에 넣은 후에, 그 뒤에 GAE또는 Cloud function (https://cloud.google.com/functions/) 에서 메세지를 받는 구조로 구현이 가능하다.


로그 모니터링

스택 드라이버 로깅은 단순히 로그를 수집할 뿐만 아니라 훨씬 더 많은 기능을 제공한다.

앞에서 스택 드라이버 로깅을 이용한 로그 수집 시스템을 만드는 방법을 알아보았지만, 부가적인 몇가지 기능이 같이 제공되는데 다음과 같다.

필터를 이용한 특정 로그 핸들링

logger를 통해서 수집된 로그에는 필터를 걸어서 특정 로그만 모니터링할 수 있다.

예를 들어서 text 문자열에 “error” 가 들어간 로그나, latency가 1초이상인 로그와 같이 특정 로그만을 볼 수 있다.

다음은 jsonPayload.text 로그 문자열에 “-a”로 시작하는 문자열이 있는 로그만 출력하도록 하는 것이다.



이 기능을 사용하면, 로그 메세지에서 특정 로그만 쉽게 검색하거나, 특정 에러 또는 특정 사용자의 에러, 특정 ErrorID 등으로 손쉽게 검색이 가능해서 로그 추적이 용이해진다.

매트릭 모니터링

다음은 메트릭이라는 기능인데, 로그를 가지고 모니터링 지표를 만들 수 있다.

예를 들어 하루 발생한 에러의 수 라던지, 평균 응답 시간등의 지표를 정의할 수 있다.

이렇게 정의된 지표는 대쉬보드에서 모니터링이 가능하고 또는 이러한 지표를 이용하여 이벤트로 사용할 수 있다. 응답시간이 얼마 이상 떨어지면 오토 스케일링을 하게 한다던가 또는 이메일로 관리자에게 ALERT을 보낸다던가의 기능 정의가 가능하다.


매트릭 생성

지표 정의는 로그 화면에서 필터에 로그 검색 조건을 넣은 채로, CREATE METRIC 버튼을 누르면 사용자가 지표를 매트릭으로 정의할 수 있다.



대쉬 보드 생성


이렇게 정의된 매트릭은 스택 드라이버 대쉬 보드 화면에서 불러다가 그래프로 시각화가 가능한데, 다음 그림은 struct_log의 전체 수와를 나타내는 매트릭과, struct_log에서 log text에 “-a”를 포함하는 로그의 수를 나타내는 메트릭을 정의하여 차트로 그리는 설정이다.



위에 의해서 생성된 차트를 보면 다음과 같이 전체 로그 수 대비 “-a”  문자열이 들어간 로그의 수를 볼 수 있다.


지금까지 스택드라이버 로깅과 빅쿼리를 이용하여 간단하게 대용량 로그 수집 서버를 만드는 방법을 살펴보았다. 두개의 제품을 이용해서 로그 수집 시스템을 구현하는 방법도 중요하지만, 이제는 개발의 방향이 이러한 대용량 시스템을 구현하는데, 클라우드 서비스를 이용하면 매우 짧은 시간내에 개발이 가능하고 저비용으로 운영이 가능하다. 요즘 개발의 트랜드를 보면 이렇게 클라우드 서비스를 이용하여 개발과 운영 노력을 최소화하고 빠른 개발 스피드로 개발을 하면서, 실제로 비지니스에 필요한 기능 개발 및 특히 데이타 분석 쪽에 많이 집중을 하는 모습이 보인다.


단순히 로그 수집 시스템의 하나의 레퍼런스 아키텍쳐에 대한 이해 관점 보다는 전체적인 개발 트렌드의 변화 측면에서 한번 더 생각할 수 있는 계기가 되면 좋겠다.


데이타 스트리밍 분석 플랫폼 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가 된다.




구글 데이타 스트리밍 데이타 분석 플랫폼 dataflow - #1 소개


조대협 (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와 같은 런타임의 개념이다.


런타임

Apache Beam으로 작성된 코드는 여러개의 런타임에서 동작할 수 있다. 구글 클라우드의 Dataflow 서비스에서 돌릴 수 도 있고, Apache Flink나 Apache Spark 클러스터 위에서도 그 코드를 실행할 수 있으며, 로컬에서는 Direct Pipeline이라는 Runner를 이용해서 실행이 가능하다.


여러 런타임이 있지만 구글 클라우드의 Dataflow 런타임을 사용하면 다음과 같은 장점이 있다.


매니지드 서비스로 설정과 운영이 필요 없다.

스트리밍 처리는 하나의 노드에서 수행되는 것이 아니라, 여러개의 노드에서 동시에 수행이 되기 때문에, 이 환경을 설치하고 유지 보수 하는 것만 해도 많은 노력이 들지만, Dataflow는 클라우드 서비스이기 때문에 별도의 설치나 운영이 필요없고, 작성한 코드를 올려서 실행 하기만 하면 된다.

Apache Spark등을 운영해본 사람들은 알겠지만, Spark 코드를 만드는 것 이외에도, Spark 클러스터를 설치하고 운영 하는 것 자체가 일이기 때문에, 개발에 집중할 시간이 줄어든다.

오토 스케일링을 지원하기 때문에, 필요한 만큼 컴퓨팅 자원을 끌어다가 빠르게 연산을 끝낼 수 있다.

클라우드 컴퓨팅의 장점은 무한한 자원을 이용하여, 워크로드에 따라서 자원을 탄력적으로 배치가 가능한 것인데, Dataflow 역시, 이러한 클라우드의 장점을 이용하여, 들어오는 데이타량이나 처리 부하에 따라서 자동을 오토 스케일링이 가능하다.


그림처럼 오전에 800 QPS (Query per second)의 처리를 하다가 12시경에 부하가 5000 QPS로 늘어나면 그만한 양의 리소스 (컴퓨팅)를 더 투여해서 늘어나는 부하에 따라서 탄력적으로 대응이 가능하다.

리밸런싱(Rebalancing)기능을 이용하여 작업을 골고루 분배가 가능하다.

Spark이나 Hadoop Map & Reduce와 같은 대용량 분산 처리 시스템의 경우 문제가 특정 노드의 연산이 늦게 끝나서 전체 연산이 늦게 끝나는 경우가 많다. 예를 들어 1000개의 데이타를 10개씩 100개의 노드에서 분산하여 처리를 한후 그 결과를 모두 모아서 합치는 연산이 있다고 할때, 1~2개의 노드가 연산이 늦게 끝나더라도 그 결과가 있어야 전체 값을 합칠 수 있기 때문에, 다른 노드의 연산이 끝나도 다른 노드들은 기다려야 하고 전체 연산 시간이 느려 진다.


Dataflow의 경우는 이런 문제를 해결 하기 위해서, 리밸런싱(rebalancing)이라는 메카니즘을 발생하는데, 위의 그림(좌측의 그래프는 각 노드의 연산 시간이다.) 과 같이 특정 노드의 연산이 느려진 경우, 느려진 노드의 데이타를 다른 연산이 끝난 노드로 나눠서 재 배치하여 아래와 같이 전체 연산 시간을 줄일 수 있다.