데이터 분석 플랫폼 구축과 활용.md

서평

• 책을 읽게 된 계기는 2가지가 있습니다. 이커머스에서 애플리케이션 서버를 운영할 때 로깅을 Fluentd를 사용하면서 Input, Parse, Filter, Buffer, Output을 설정하였고 Fluentd를 만든 treasure data사에 대해서 궁금한 부분이 몇 가지 있어서 읽게 되었습니다. treasure data사는 데이터 처리의 플랫폼으로 오픈 소스의 스트리밍 형 전송 도구인 Fluentd와 벌크 형 전송 도구인 Embulk를 개발한 곳입니다. 현재 treasure data 사의 CDP(Customer Data Platform)를 운영하면서 플랫폼에 대해서 공부를 하다가 이 책을 읽게 되었습니다.
• 책은 이론을 설명하면서 그림과 실습형 소스코드 위주로 이루어져 있는데 로그에 대해서 자세하게 다룰 수 있어서 좋았습니다. 이 책은 서버 엔지니어나 인프라엔지니어, 데이터 엔지니어를 대상으로 하고 있습니다.
• 이 책은 데이터를 수집하는 방법부터 보존, 데이터의 시각화를 대상으로 설명하고 있으며 그 외에 Emublk & Digdag에 대해서도 간략하게 소개하고 있습니다. 책을 읽으면서 Fluentd 뿐만 아니라 Elasticsearch와 Kibana에 대해서 자세하게 알게 되어서 좋았습니다.
• 책 초반에는 데이터 분석 기반을 만들 때 생각할 요소로 수집,변환,보존,분석,표시,운영에 대해 전반적으로 시작하면서 로그에 대해 범용성을 살펴볼 수 있었습니다. Fluentd 뿐만 아니라 다른 로그 수집 미들웨어에 대해서도 특징에 대해 알려주고 있어서 좋았습니다.

데이터 분석 개요

• 데이터 분석 기반을 만들 때 생각할 요소
  • 수집 : 어떤 데이터를 모을 것인가. 어떤 데이터를 보관할 것인가
  • 변환 : 데이터를 어떻게 전처리해서 분석하기 쉬운 형태로 바꿀 것인가
  • 보존 : 데이터를 어디에 저장할 것인가
  • 분석 : 어떤 기반에 넣어서 데이터를 분석하고 활용할 것인가. 데이터를 넣는 것 뿐만 아니라 분석을 위한 좋은 환경을 만들기 위해서 어떻게 해야할까
  • 표시 : 어떻게 데이터를 시각화하고, 결과를 전달할까
  • 운영 : 데이터를 분석하기위한 기반을 장기적으로 운영하기 위해서는 어떻게 해야할까. 엔지니어가 아니라도 활용할 수 있는 환경을 만들기 위해서는 어떻게 해야할까
• 대상이 되는 데이터
  • 마스터 데이터 : 시간이 지나도 바뀌지 않는 데이터. 예를 들어 상품명과 같은 것은 시간이 지나도 바뀌지 않음.
  • 시계열 데이터 : 어떤 시간에 발생하는 사건을 기록하는 데이터. 로그 데이터와 같은 경우
  • 설정 데이터 : 상황에 따라 애플리케이션의 동작을 변경하기 위해서 보존하는 데이터
• SoR(System of Record) : 기록(Transaction)을 위한 시스템. 입출력의 내용을 문제없이 바르게 처리하기를 기대하는 시스템. 예를 들어 신용카드에 의한 결제나 항공권의 예약시스템 등의 경우
• SoE(System of Engagement) : 관계강화(Engagement)를 위한 시스템. 고객의 이용을 촉진하기 위한 시스템.예를 들어 상품을 추천하는 시스템이나 고객에게 검색결과를 보여주는 시스템의 경우
• 의사결정을 위한 데이터 분석
  • 데이터에 근거한 의사결정
    • 데이터와 정보의 관점에서 보면 의사결정이란 것은 불확실성 그리고 그에 관련한 불이익과 이익을 관리하는 것이다.
  • 데이터 분석의 단계와 엔지니어링
    • 분석의 목적은 수치를 해석하는 것이 아니라 의사결정을 하는 것에 있음. 그리고 그 결과로 사업에 공헌하거나 서비스를 개선할 수 있음. 시각화도 마찬가지로 결과를 내는 수단이며, 사고의 도구
  • 웹상의 시각화
    • 특징
      • 인터랙티브한 데이터의 취급이 가능하므로 드릴다운, 드릴업이 가능함
        • 인터랙티브한 데이터의 취급이란 포인팅 디바이스(대체적으로 마우스를 말함)에 의한 조작이나 조건 검색에 의해서 현재 표시된 데이터를 가공하고 양방향으로 데이터를 취급하는 것
        • 드릴다운은 특정 상태의 데이터에서 조건을 더 지정하여 범위를 좁히는 것
        • 드릴업은 조건검색을 뺌으로써 좀더 넓은 범위의 데이터를 보는 것을 의미함
      • 임의의 데이터 또는 분석 상태에 대한 URI를 발행할 수 있음
      • URI를 공유하는 것으로 간단히 여러 사람이 같이 분석할 수 있음

로그 데이터의 기초

• 서비스 개선을 위한 로그
  • hot data : 발생한 직후의 데이터. 발생하고 바로 이용됨. 또는 고빈도로 조회되는 데이터를 말함
    • hot data는 준 실시간의 분석을 가능케 함
      • 지금 이 수간에 트래픽에 어떤 변화기 있는가
      • 특정 캠페인을 실시간 타이밍에 어떤 상품의 주문이 증가 했는가
      • 특정 콘텐츠에 갑자기 request가 집중되고 있는데, 어떤 경로에서 들어오고 있는 것인가
    • 액세스가 빈번하게 일어나는 데이터를 hot data라고 함
  • cold data : 작성되고 난 뒤 이용될 때까지 일정 시간이 걸리는 데이터
    • cold data는 일단 데이터를 스토리지에 저장한 뒤에 이용될 때까지 일정기간 조회되지 않음, 장기적으로 백업을 위한 데이터
  • warm data : hot data와 cold data의 사이에 있는 데이터
• 펀더멘탈 매트릭스(기초 지표) : 행동의 종류 * 고객의 속성 * 상품의 속성 * 시간단위 * 기본 통계량

데이터 분석 기반 구축

• 인덱스 : Elasticsearch에서 도규멘트를 저장하는 단위
• Mapping이라는 것은 어떤 인덱스에 대해서 각 필드의 형과 인덱싱의 방법을 지정하는 기능

데이터 분석 기반 운영

• ETL
  • Extract : Fluentd에서는 Input계의 플러그인에 해당함
  • Transform : Fluentd에서는 filter 플러그인을 이용함으로써 이 Transform의 사양을 처리하는 것이 가능함
  • Load : Fluentd에서는 out_elasticsearch 플러그인이 담당
• 데이터 분석 시스템 가용성
  • 데이터 멱등성
    • At most once
      • 메시지는 바로 송신할 것. 만약 메시지의 송신이 성공했다면 다시는 송신하지 말 것. 다만. 저장소가 꽉 차거나 해서 메시지를 잃어버릴 수 있음
    • At least once
      • 각각의 메시지는 적어도 한 번은 송신됨. 실패한 경우에는 메시지는 중복될 수도 있음
    • Exactly once
      • 각각의 메시지를 정확하게 한 번만 전송함
    • Fluentd에서는 At most once와 At least once를 지원하고 있음. 비동기로 데이터를 전송하는 것으로 높은 throughput를 실현할 수 있기 때문
  • Exactly once의 실현은 어려움
    • 비동기이면서 분산시스템에서 Exactly once를 실현하는 것은 매우어려운 과제
  • retry(재시도)
    • Load의 처리는 몇 가지의 요인으로 실패할 가능성이 있음
      • 쓰기를 할 곳의 데이터 저장소 용량이 꽉 차버렸음
      • 쓰기를 할 곳에 네트워크가 끊어졌음
      • Fluentd가 동작하고 있는 머신의 전원이 갑자기 끊어졌음
  • 더욱 가용성을 높히기 위해서 : 큐를 사용함
    • Fluentd에서도 멱등성과 재시도에 대해서 시도하고 있지만 Fluentd 노드 자체에 장애가 발생하거나, 데이터를 전부 잃어버렸을 때는 Fluentd의 가용성을 높이는 것이 어려움
    • Amazon Kinesis
      • AWS에서 스트림 처리용의 메시징 기반. 분산 메시지큐라고 하는 시스템. 분산 메시지큐를 이용하면 메시지의 다중화가 가능하기 때문에 하나의 노드에서 데이터를 가지는 것보다 데이터의 안전성을 향상할 수 있음
      • 분산 메시지큐에 넣은 데이터는 거기에서 데이터를 꺼내는 워커가 정기적으로 추출하고 재이용함
• 데이터 저장소 비교
  • 관리형 데이터 웨어하우스
    • 운영에 할당할 리소스와 각 데이터 저장소의 튜닝에 시간을 절약하면서 이 데이터 저장소와 미들웨어를 활용하고 싶은 경우에는 호스팅 서비스를 이용하는 것도 검토해볼 수 있음
    • Fluentd를 이용하는 경우라면 Treasure Data를 자연스럽게 사용할 수 있음
    • Google Cloud Platform이 제공하는 완전관리형 서비스인 BigQuery는 Dremel이 베이스 가 되는 대규모 컬럼지향형 데이터구조에 대해서 쿼리가 가능함
  • 스토어, 서비스의 검토 항목
    • 데이터 저장소, 서비스, 스트림처리를 나누어 사용하는 이유
      • 코스트
        • TCO(Total Cost of Ownership - 총보유비용)와 분석에 드는 비용. TCO는 데이터의 저장에서부터 파기할 때까지 필요한 시간과 지출을 의미함
        • 데이터가 늘어나면 늘어날수록 일반적으로 데이터의 보유 비용은 늘어남
      • 정보의 신선도
        • 분석 대상이 되는 데이터에서 준실시간으로 결과를 알고 싶은지, 어느 정도 시차를 두어도 되는지에 따라 분석방법이 달라짐
      • 확장성(scalability)
        • 각각의 스토어에 관한 확장성과 스트림처리의 아키텍처의 확장성 양쪽을 의미함
      • 스키마의 유연성
        • 어떤 타이밍에 데이터의 스키마를 결정할 것인가라는 점을 다루고 있음
        • Schema on write : 데이터의 스키마가 처음부터 결정되어 있다면, 쓰기를 할 때 데이터의 스키마가 고정되어 있어도 문제가 없음. 쓰기를 할 때 스키마가 결정되는 방식. 관계형 데이터 베이스에 데이터를 넣을 때는 테이블이 필요하기 때문에 최초에 테이블을 만들고 데이터를 넣는 것을 말함
        • Schema on read : 데이터를 읽을 때 스키마를 결정하는 방식. 데이터를 저장할 때에 스키마를 결정할 필요가 없음. 데이터의 스키마가 자주 변할 때는 이 방법을 다루기 좋음
        • Schema on write에서는 스키마를 변경하기 위해 시간이 걸리지만, 인터렉티브한 쿼리에 대해서 고속으로 답할 수 있는 이점이 있음
      • 중간데이터의 유지
        • 행동모델을 도출하기 전에 전단계에서 필요한 데이터를 산출하거나 집계하는 것으로도 데이터는 생겨남. 이 데이터를 중간 데이터라고 함
  • 저장소의 검토 포인트
    • 장기간 데이터를 유지해야할 피룡가 있고, 10테라바이트 정도의 데이터를 매시간 분석대상으로 할 필요가 있다면 현 시점에서는 Hadoop이나 BigQuery 또는 몇 가지의 MPP(Massively Parallel Processing)데이터베이스가 선택지가됨
    • Elasticsearch가 유효한 선택지
      • 데이터를 어느 정도 기간만 유지해도 됨
      • 스트림에서 데이터를 처리하면서, 직전1시간의 1분단위 데이터를 집계할 필요가 있음
      • 어느 정도의 확장성을 필요로 함
    • 스트림 처리를 한 뒤에 집계 데이터만 유지할 필요가 있고, 매초의 집계가 필요한 경우라면 CEP(Complex Event Processing) 엔진과 같은 스트림처리를 하는 아키텍처가 맞음
    • 복수의 데이터베이스와 미들웨어를 조합해서 분석 기반을 구축하는 것이 코스트 효율이 좋고 효과적으로 분석할 수 있는 기반을 만드는 방법. 배치 처리와 스트림처리를 병용하는 아키텍처, 그 중에서도 분산컴퓨팅 환경에서의 제안은 람다(lambda) 아키텍처라고 함

로그 수집 미들웨어

• 서비스 개선에 없어서는 안될 로그 수집
  • AARRR 모델
    • 서비스 개선으로 이어지는 요소로서의 로그를 검토할 때는 참고할 수 있는 사고방식
    • 서비스 이용에서 유저 행동의 단계를 5개로의 요소로 나누어서 단계(phase)별로 기표를 만들고 개선 정책을 세우기 쉽도록하는 프레임워크
    • 분석결과를 보고 다음 행동을 선택하는, 데이터주도(data driven) 경영을 지향하는 기업 또는 팀에게는 꼭 필요한 사고방식
    • AARRR 각각의 요소에 드는 비용과 성과 그리고 필요한 컨버전율, 컨버전 비용의 분석이라는 프레임워크는 유용함
      • Acquisition : 유저획득
        • SEO(검색엔진 최적화. 검색엔진에서 검색결과의 상위에 나올 수 있도록 하는 작업을 말함)나 성과보수형 광고(affiliate). 소셜이나 TV, 웹 등에서 광고를 목적으로 하는 정책에서 얼마나 첫방문이 늘었는지를 측정한 지표. 투입한 코스트를 방문 유저로 나누어서 비율로 측정함
      • Activation : 활성화
        • 처음으로 이용한 유저가 어느 정도 활성하였는가를 측정하는 지표
        • 활성화한 유저(액티브 유저) 수를 방문 유저 수로 나누어서 비율을 측정함
      • Retention : 지속
        • 반복 이용을 독촉하는 메일이나 스마트폰 알림통지, 리타게팅 광고 등에서 어느 정도의 활성화 유저를 획득했는지를 계측하는 지표
        • 정책별로 비활성화 유저가 재방문해서 재활성화 유저가 되기까지의 전환율
      • Referral : 소개
        • 기존 유저가 다른 누군가에게 서비스를 소개해서 얼마나 신규 유저를 얻을 수 있는지 측정하는 지표
        • 소개 기능의 이용율과 그로부터 실제로 얻게 되는 신규 유저의 전환율로 측정함
      • Revenue : 수익화
        • 유저가 서비스 안에서 어느 정도 수익에 공헌했는지 어느 정도 과금을 했는지를 측정하는 지표
        • 금액에 상관없이 수익에 이른 비율과 ARPU(Average revenue per user : 유저당 평균매출, 한 사람의 유저가 특정기간 동안 발생시키는 평균 매출액)라고 하는 유저당 평균 금액으로 측정함
        • AARRR 모델을 사용함으로써 서비스의 어떤 부분에 어떤 숙제가 있는지 명확해지고, 현재 상황에서 개선점과 마케팅의 전략을 세우기 쉽게 됨
• 현대적인 로그 수집
  • 빈번하게 변화하는 데이터 구조에도 유연하게 대응할 수 있음
  • 단시간에 집계할 수 있고, PDCA(Plan-Do-Check-Act. 계획, 실행, 평가, 개선의 4단계를 반복하는 사이클을 의미함) 사이클을 빠르게 돌릴 수 있음
  • 액세스 증가에 따라 데이터 양이 급증하더라도 서버 수와 처리양을 늘려서 스케일링이 가능함
  • 로그의 추출과 집계
    • JSON Lines(JSONL) : 한 행에 하나의 JSON 오브젝트를 저장하는 로그 형식
    • LabeledTSV 형식(LTSV) : 값의 이름인 라벨과 값을 하나의 콜론으로 구분하고, 각각의 요소들은 탭으로 구분하는 데이터 포맷. LTSV 형식은 JSON 형식에 비해서 자유도가 떨어지지만, 단순한 형식이기 때문에 복잡한 정규표현식은 사용하지 않아도 됨
  • 로그 출력의 동기처리
    • 애플리케이션에서 직접 로그 파일을 남길 때는 배타처리와 동기처리가 필요함. 배타처리라는 것은 같은 파일에 동시에 쓰기를 하면 깨진 문자열이 남은 것을 방지하기 위해서 파일락(file lock)을 FLOCK 등으로 락을 점유하여 중복쓰기를 방지하는 것을 말함
    • 멀티스레드나 멀티프로세스 프로그램에서 하나의 파일에 로그를 출력하는 경우에는 쓰기 순서가 바뀌지 않도록 동기처리도 필요함
  • 로그의 로테이트(rotate) 처리
    • 일반적인 미들웨어에서 로그 로테이트를 하기 위해서 보통 HUP(Hang Up) 시그널을 사용함. 로그 파일에 쓰기를 일단 중지하고, 새로운 파일에 쓰기를 시작하도록 하는 처리
  • 로그 전송 시의 네트워크 부하
    • 유지관리면에서 좋은 방법
      • 미세하게 sleep 처리를 넣어서 네트워크 점유시간을 짧게 유지함
      • 네트워크 통신 속도에 제한하는 커맨드를 같이 사용함
      • 실행 스케줄을 장비별로 미묘하게 다르게 함
• 로그 수집 미들웨어
  • Scribe - C++ - 2008년 - Facebook
  • Flume - Java - 2010년 - Apache Project
  • Fluentd - C + Ruby - 2011년 - TreasureData Inc.
  • Logstash - JRuby - 2011년 - elasticsearch Inc.
  • 로그 수집 미들웨어 특징
    • 해석 가능할 정도의 짧은 소요시간
      • 로그 수집 제품의 특징인 비동기통신을 활용하면 시계열 데이터 처리에 어울려 거의 실시간으로 로그를 수집해서 활용할 수 있게 됨
      • 실시간으로 신선한 데이터를 다루는 것의 장점
        • 데이터 스트림 처리의 데이터 수집 대기 시간을 줄임으로써 신선하고, 정밀한 집계
        • 유저의 신선하고 상세한 행동 로그를 가지고 높은 정밀도의 추천과 매칭
        • 마케팅 정책의 압도적으로 빠른 효과 측정(TV 광고나 광고메일, 웹 광고 등)
        • 시스템 리소스의 실시간 트러블 모니터링으로 문제의 빠른 해결
        • 센서 데이터를 활용한 스트림 컴퓨팅(에너지 최적화, 재난방지 등)
        • 차량의 위치 정보와 사람의 행동, 도로별 정체 상황을 분석한 교통제어
        • 신용카드 등의 웹사이트에서의 부정이용의 검출
        • 소셜데이터의 수집 등을 통한 주식 알고리즘 트레이딩
        • 행동 로그를 가지고 관객의 위약 예측을 해서 위약 예방 마케팅의 실시
    • 우수한 리소스 사용
      • 로그 수집 미들웨어의 특징인 비동기통신, 즉 데이터스트림에 의한 순차송신방식은 쇼트패킷이 아니라 롱패킷방식으로 우수한 네트워크 전송방식
      • 1건의 레코드를 하나의 파일로 버퍼링하면 잦은 파일액세스가 발생하므로, 복수의 레코드를 묶어서 chunk라는 묶음으로 버퍼링해서 비동기로 상위서버에 전송함 . 로그전송량이 급증하더라도 랜덤액세스가 잘 일어나지 않고 비교적 큰 블럭액세스가 되기 때문에 디스크I/O 처리를 점유함으로 인한 응답속도저하를 방지할 수 있음
      • 부하가 줄어들고, 시스템의 응답성능을 어느 정도 확보할 수 있음
    • 비동기화 처리에 의한 빠른 처리
      • RDBMS(Relational DataBase Management System)와 로컬 파일에 배타락(Exclusive Lock)을 걸고 트랜잭션 처리를 하면 확실성은 늘어나지만, 쓰기에 대한 성능이 점점 한계가 옴
      • 로그 수집 미들웨어를 이용해서 TV나 메일링리스트, 시기적 요인 등에 의해 액세스 집중으로 초당 레코드 건수가 급증하더라도 큐잉으로 대응할 수 있음. 유저의 응답시간에 거의 영향을 주지 않고 로그 수집을 할 수 있게 됨
    • 통신의 예외 처리 / 재시도 처리
      • 직접 구현하기 매우 까다로운 예외처리와 재시도 처리를 맡길 수 있음
        • 통신중에 네트워크가 순단(네트워크가 간혈적으로 끊기는 현상)하는 경우
        • TCP(Transmission Control Protocol)층에서는 서버에 도달해서 소켓 버퍼에 들어와 ACK 응답이 오지만 애플리케이션층에서 문제가 발생해서 소실하는 경우
        • TCP층에서는 문제가 없었지만, 애플리케이션층에서 도착한 ACK 응답이 소실되었을 때
        • 폭주에 의해서 응답이 없고, 상대 서버에 바로 보낼 수 없는 경우

처음으로 만나는 Fluentd

• Fluentd는 로그 수집 미들웨어. 저장 장소가 분산되어 있는 데이터와 로그의 수집을 간단하고 스마트하게 해결해 줌으로써 데이터로부터 가치를 창출하기 위한 비용을 최소화 할 수 잇음
• 데이터 구조
  • Fluentd에서 하나의 메시지는 [tag, time, record]라는 3개의 요소로 구성되어 있음.
    • tag : 레코드의 라우팅에 사용하는 문자열
    • time : UNIX 타임스탬프로 저장
    • record : 객체형으로 Key-Value 형식의 연관 배열을 저장함
  • record는 중첩 구조도 다루긴 합니다만 최종적으로 데이터를 저장하는 곳이 지원하지 않는 경우는 플랫 Key-Value 형식으로 메시지를 구성하든지 필터처리로 변환할 필요가 있음.
• 아키텍처
  • 로그/메시지에 임의의 태그를 붙여서 순차수집을 하면서 필터/버퍼/태그 또는 라벨을 사용한 라우팅을 거쳐서 각종 데이터 출력 장소에 보존하는 것을 안정적으로 비동기 처리할 수 있는 아키텍처
  • 기본적인 데이터의 흐름
    • 각종 데이터 소스에서 로그/메시지에 태그를 붙여서 수집함(Input 플러그인)
    • 필터라고 하는 데이터 가공과 집계 처리를 필요에 따라 실행함(Filter 플러그인, Filter계 Output 플러그인)
    • 각종 데이터 저장소로 출력함(Output 플러그인)
  • Input 플러그인으로 schemaless한 구조화 메시지를 받음(또는 데이터를 추출함). 필요에 따라 Filter 플러그인과 Filter계 Output 플러그인을 여러개 조합해서 데이터의 가공 처리를 함
  • 무엇인가 에러가 발생하면 적절한 재시도 처리를 하면서 Output 플러그인을 사용해서 최종적인 데이터 저장소에 보냄
  • 파일버퍼 기능을 사용하면 데이터의 저장소에 통신 에러가 발생하더라도 버퍼에 쌓고 재시도하는 구조로 되어 있기 때문에 디스크 용량이나 예외처리 관련 기능을 생략할 수 있음
  • 높은 성능을 내기 위해서 Fluentd의 코어부분에는 C언어 네이티브로 작성된 cool.io(이벤트 루프 라이브러리)와 MessagePack(직렬화 라이브러리)을 사용함
• 배치처리와 다른 점
  • FTP에 연결된 대용량 CSV를 마이크로배치화하는 경우에는 Fluentd의 배치 처리만 특화한 embulk라든가 스케줄 관리와 네트워크 플로우 툴인 digdag를 추천함
  • Fluentd와 embulk 양쪽 다 입력, 필터 처리, 출력이라는 데이터의 파이프라인 처리는 설정 파일로 실현할 수 잇기 때문에 개념이 비슷하고 범용적으로 채택하는 경우가 들어나고 있음
• 도입의 간편함 : td-agent 패키지를 사용하는 장점은 기존의 시스템에 영향이 적은 것
• 플러그인 에코 시스템
  • 플러그인의 세 종류
    • 각종 소스로부터 데이터를 입력하는 Input 플러그인
    • 데이터를 가공하는 Filter 플러그인
    • 데이터를 출력하는 Output 플러그인
  • Fluentd와 플러그인이 Ruby 언어로 작성되어 있기 때문
• 동작 프로세스 확인
  • Fluentd는 Supervisor 프로세스에서 Worker를 기동하기 때문에 부모와 자식 프로세스, 즉 2개의 Ruby 프로세스로 구성되어 있음
• 번들 프러그인
  • Output 플러그인
    • td : Hadoop/Presto 데이터분석 연계 기반의 TreasureData에 레코드를 송신함
    • td_counter : TreasureData의 모니터 서비스에 매트릭스 데이터를 송신함

Fluentd 설계 요령

• tail 플러그인 구조
  • Fluentd의 표준 플러그인인 tail 플러그인은 파일에 한 줄에 하나의 레코드 또는 여러 줄에 하나의 레코드를 남긴 로그를 읽을 때 이용함
• Fluentd 노드 디자인패턴
  • Fluentd는 여러 개의 노드로 분산 처리함으로써 하루 100억개 이상의 메시지를 처리할 수 있는 고가용성 로그수집 시스템으로 설계됨
  • 유연한 로그수집을 가능케 하며 Fluentd 노드 전체의 성능과 가용성, 유지관리의 용이성을 높이기 위해서 각 Fluentd 서버가 원하는 처리와 역할을 단순하게 하는 것이 중요함
  • Fluentd 쪽에서 시간단위로 집계를 하고 싶을 때
    • HTTP 응답코드를 datacounter 플러그인으로 카운트하고, 그것을 클라우드 서비스의 Mackerel이나 datadog으로 보내서 그래프를 만들거나 감시 조건으로 이용하는 경우에는 Aggregator가 필수
    • 초당 수만건의 상당한 대규모의 데이터의 후처리를 하는 경우에는 fluent-plugin-kinesis를 사용해서 Amazon Kinesis Streams와 Amazon Kinesis Firehose를 이용하는 사례도 늘고 있음
  • 각 노드의 역할
    • Forwarder Node(포워드 노드) : 수집한 로그를 Aggregator로 전송하는 말단 노드
    • Aggregator Node(애그리게이터 노드) : Forwarder로부터 로그를 전송받아서 집적하는 노드. 가벼운 데이터처리나 집계처리를 함
    • Processor Node(프로세서 노드) : CPU 리소스를 필요로하는 데이터처리와 단위시간당 집계처리를 하는 집계 노드
  • 노드 구성 예
    • 싱글구성으로 가능한 것
      • 애플리케이션의 로그를 그대로 데이터베이스에 보관함
      • 센서데이터를 시리얼포트에서 순차적으로 취득해서 데이터베이스에 보관함
      • syslog를 전송, 수신해서 데이터베이스에 보관함
      • http / tcp / udp로 로그와 메시지를 수신해서, S3 등의 스토리지에 보관함
      • 로컬이나 API 폴링(Twitter의 스트림 등)으로 수집한 데이터를 가공해서 보관함
      • 웹 애플리케이션의 행동로그를 큐잉해서, 스트림처리 서비스인 Amazon kinesis에 전송함
      • 애플리케이션과 시스템의 매트릭스 정보를 취득해서 그래프로 만들고 대시보드 서비스(Mackerel, Datadog, Librato Metrics 등)의 API로 송신함
      • AWS의 ELB의 로그를 모아서 필터 처리한 뒤에 데이터 집계와 해석을 하는 데이터웨어하우스에 보관함
  • 범용 구성
    • 여러 대의 Forwarder 노드에서 데이터를 수집함
    • 단위 시간별로 모아서 보관함
    • 단말의 Forwarder 노드의 처리를 단순화하고, 데이터의 가공과 집계를 일원화함
    • 로그 데이터를 여러 곳의 데이터 보관소에 보관함
    • 분산처리를 지원하는 스토어에 데이터를 출력함
    • 윈도우로 구분하여 단위 시간별로 SQL로 집계하는 Norikra에 일단 보낸 뒤에 결과를 출력 장소에 보관함
  • 응용 구성
    • 로그의 양이 초당 1Gbps또는 초당 100만건의 메시지를 넘는 대규모의 환경이라면 스트림데이터 분산처리 기반의 Apache Storm이나 클라우드 서비스의 힘을 빌림
    • AWS를 사용하면 Processor 노드를 사용하지 않고, fluent-plugin-kinesis를 사용해서 Aggregator에서 버퍼링해서 Amazon Kinesis Firehose나 KPL(Kinesis Producer Library)로 보냄
    • Fluentd를 사용하는 것으로 물론이고 API 호출 회수도 줄이고, 안정성 향상과 비용절감이란느 장점
    • 시각화 도구에 대해서 자신의 서버 환경에 GrowthForecast나 Graphite, Elasticsearch로 대응할 수도 있음.
    • 유료인 클라우드서비스도 고려한다면 Redash, Mackerel, datadog, GoogleDataStudio, Geckoboard, Metric Insights라는 그래프화가 가능한 대시보드 서비스를 사용하여 운영의 수고를 덜 수 있음
    • datadog는 일반적으로 서버의 모니터링 툴로 알려져 있지만 그래프 작성수에 제한이 없기 때문에 KPI 지표 등의 데이터를 Fluentd 플러그인이나 API로 보내면 대시보드 도구로도 사용할 수 있고, 감시에 의해서 에러통지도 메일이나 전화로 받을 수 있음
• 유스케이스
  • 액세스 로그를 수집해서 사이트(도메인)별로 파일명을 나눠서, gzip으로 압축한 로그를 Amazon S3에 아카이브함
  • 단위시간당 액세스수와 상태코드별 집계 결과를 사이트별로 그래프화함
  • 애플리케이션 로그/액세스 로그를 수집해서 히스토그램으로 만들고 완전/부분 일치 검색과 모니터링이 가능한 대시보드를 Grafana와 Redash 또는 Elasticsearch와 Kibana 등과 함께 구성해서 만듬
  • IP 주소를 가지고 GeoIP를 사용해서 위도와 경도 정보를 추가하고 유저 정보와 함께 지도에 표시함
  • PC와 스마트폰, RSS 기능, API 기능의 액세스 볼륨의 시각화
  • 액세스 로그의 응답시간을 사용해서 페이지 응답 속도의 최소/평균/최대의 변동폭의 추이를 계측함
  • HTTP 상태코드가 404, 500 에러일 때의 추이를 계측함
  • Googlebot 등의 클롤러의 액세스 현황을 시각화함
  • 로그인 성공수/로그인 실패수를 계측함
  • 버튼의 효과를 측정하는 A/B 테스트의 결과를 시각화함
  • 자사 광고의 노출(Impression) 회수, 클릭 이벤트수의 추이를 계측함
  • 웹서버와 DB 서버의 에러 로그를 전체적으로 확인함
  • MySQL의 슬로우쿼리를 시각화함
  • GoogleAnalytics를 대체하는 자신만의 Web비콘을 만듬
  • 로그 양의 스파이크 등 이상치가 검출되면 채팅 도구, HipChat, Slack 등에 통지함
  • Norikra를 사용해서 스트림 데이터의 단위시간당 데이터의 집계를 SQL로 하고, 그 결과를 데이터 베이스와 그래프화 미들웨어에 전송하여 대시보드 도구에서 사용함
  • iOS, Android로 푸시통지를 하는 큐형 메시지 서버로 이용함
  • 메일의 전송실패 로그를 상태코드와 함께 수집하여 MySQL에 보관하고, 전송금지 리스트 등록까지의 지연을 최소화함
  • 메일의 전송 이력을 수집해서 수신자 도메인별로 메일의 비도달률을 집계하여 메일주소 스크린 현황을 모니터링함
  • 메일의 전송이력을 수집해서 메일주소를 암호화(SHA1)한 뒤에 보관하고, 검새가능한 대시보드를 Elasticsearch + Kibana로 만듬
  • munin과 dstat, 시리얼포트에서 얻은 시스템가동 현황을 데이터베이스에 저장함
  • 공통적이고 가벼운 전처리를 한 뒤에 Amazon Kinesis 보내서 실시간 이벤트처리를 구현함
• 노드 구성과 사용할 플로그인
  • 애플리케이션 서버의 로그를 S3에 저장함
    • 복수의 애플리케이션 서버에 보관된 Apache 액세스 로그와 애플리케이션 로그를 수집해서 Aggregator 노드로 전송함
  • ELK(Elasticsearch, Fluentd, Kiana)의 구성
    • 여러 개의 웹서버로부터 수집한 Apache 액세스 로그에 지역 정보를 부여한 뒤에 Amazon S3와 Elasticsearch에 저장함

Fluentd 운영 Tips

• Fluentd 감시
  • 안정된 로그 수집을 하기 위해서는 정기적으로 리소스와 성능의 감시, 시각화를 하여 장애의 징후를 미리 알아내야 함
  • Fluentd의 운영을 함에 있어서 필요한 프로세스와 CPU의 점유율의 감시, 포트 감시, 에러 로그 감시, End-to-End 감시, 버퍼의 리소스 감시, 성능 감시의 방법
  • @ERROR이라는 라벨은 Fluentd의 에러 스트림으로 불리는 플러그인 내부의 레코드에서 에러가 발생할 때 사용함. @ERROR 라벨이 붙은 fluentd,warn 태그로 레코드가 옴
• 로그 누락을 막기 위한 8가지 포인트
  • 시스템 설정으로 로그 누락 막기
  • 적절한 버퍼 설정하기
  • 네트워크 단절에 대비하기
  • 프로세스 다운에 대비하기
  • 로그 누락을 막기 위한 forward 플러그인의 설정
  • 로그 Aggregator의 이중 구성
    • Aggregator가 다운되었을 때에도 최소한의 지연으로 중계를 계속하기 위해서는 Aggregator 2대를 준비해서 , SPoF(Single Point of Failure)가 없도록 이중구성을 함. 고가용성을 실현하기 위해서는 SPoF라는 단일장애 지점을 없애야함
  • 종료 전에 버퍼를 플러시하기
  • 메모리 버퍼가 있는 언어를 이용하기
• 청크라고 하는 큐에 로그 이벤트를 추가함, Fluentd 프로세스 전체에서 동시에 여러 파일의 수가 파일 디스크립터의 상한을 넘게 되면 그 이상의 청크를 만들 수 없다는 것도 주의해야 함

데이터 저장소 입문

• 데이터 저장소 요건
  • 로그의 데이터 저장소 요건
    • 여러 가지 형식의 데이터를 보존
      • 로그의 형식은 여러 가지 데이터 형식이 존재함. 액세스 로그, 애플리케이션 로그 이벤트 데이터 등
    • 시간의 경과에 따른 증가에 대응
      • 로그 데이터는 기본적으로 추가만 되는 데이터. 데이터 저장소로써 시간이 경과할수록 증가하는 데이터를 보존하여, 스케일링할 수 있어야 함
      • 간단하게 시간의 범위를 기준으로 데이터를 액세스할 수 있어야 함
    • 유연한 데이터 변경에 대응
      • 로그의 형식은 개발하면서 변경될 가능성이 있음. 로그 데이터가 변경되더라도 유연하게 대응할 수 있는 점이 데이터 자장소에게 필요함
• 대표적인 데이터 저장소
  • Hadoop
    • 대용량의 데이터를 HDFS(Hadoop Distributed File System)에 저장하여, 로그의 수집과 변형 등을 MapReduce으로 병렬처리할 수 있음.
    • Hive와 Pig 등, HDFS 상의 데이터를 MapReduce로 구현할 필요없이 처리할 수 잇는 미들웨어도 나와 있음
    • 더욱 빠른 데이터 처리를 위한 Presto 등도 개발되어 있음. Hadoop은 주로 장기간에 걸쳐서 저장된 데이터를 가지고, 배치처리로 집계할 때 효율적으로 처리하기에 용이함
  • InfluxDB
    • 시계열 데이터, 이벤트, 메트릭스를 저장하는 목적으로 개발된 데이터 저장소
    • 저장된 데이터를 얻는 HTTP 인터페이스도 가지고 있음. InfluexDB는 관리용의 WebUI
  • Prometheus
    • OSS 모니터링 시스템. 모니터링 대상이 되는 서버에서 매트릭스를 취득(Pull)하는 Pull형 모니터링 시스템
    • 내부에 독자적인 시계열 데이터베이스를 가지고 있어서, 쓰기와 읽기를 고속으로 처리할 수 있음. PromQL이라는 독작적인 쿼리언어로 시계열 데이터 처리를 하기 쉬움

Elasticsearch 기초

• Elasticsearch는 OSS(Open Source Software)로 발표한 분산형 전문 검색 서버
• Apache Lucene라는 전문 검색 라이브러리를 코어로 이용하고 있음
• 특징
  • OSS(Apache License v2) : Github상에서 오픈소스로 공개되어 있어, 수정에 대한 요청과 패치도 간단히 작성할 수 있음
  • 도큐멘트 지향 : 도큐멘트 단위로 필드 정의를 할 수 있기 때문에 유연한 데이터 등록이 가능함
  • 분산 시스템 : 인덱스를 분산해서 저장하고, 검색함. 스케일아웃을 처음부터 생각한 설계
  • 멀티테넌시 : 여러 개의 인덱스를 등록할 수 있음
  • RESTful API : 데이터의 조작과 설정, 감시 등의 필요한 기능이 HTTP 인터페이스로 이용가능함
  • 근실시간 : 데이터를 근실시간으로 검색가능함
• Elastic Stack은 데이터의 취득도구인 Logstash, Beats, 데이터 저장소인 Elasticsearch, 데이터의 시각화 도구인 Kibana를 아우리는 이름
• 아키텍처
  • 노드
    • Elasticsearch의 하나의 프로세스에 해당함. 하나의 서버 안에 여러 개의 노드를 구동할 수 있음. 노드는 개별 이름과 ID를 가지고 식별할 수 있음
    • 노드명은 디폴트로 UUID(Universally Unique Identifier)의 앞자리 7자리의 문자를 사용함(내부에서 노드의 ID로는 UUID를 이용함)
  • 클러스터
    • Elasticsearch는 여러 개의 노드를 하나의 Elasticsearch로서 동작시킬 수 있음. 그 노드군을 클러스터라고함. 클러스터를 구성함으로써 대량의 데이터를 여러 개의 노드에 분산해서 유지할 수 있음
    • 가용성과 검색 성능의 향상을 위해서 클러스터 안에 데이터 복제본인 레플리카를 둘 수도 있음. 클러스터로의 데이터 등록, 검색은 각 노드들에게 요청으로 변환되어 각 노드에서 처리함
  • 도큐멘트
    • Elasticsearch가 취급하는 데이터의 최소 단위를 도큐멘트라고 함. 일반적으로 JSON형식의 데이터
    • RDBMS의 레코드 하나에 해당함. Elasticsearch는 스키마프리이기 때문에 각 도큐멘트는 서로 다른 구조를 가질 수 있음. 도큐멘트는 복수의 필드로 구성되고 공통의 항목(필드)은 같은 형(타입)을 가져야 함
  • 필드
    • RDBMS의 컬럼에 해당함
  • 인덱스
    • 인덱스는 도큐멘트의 집합. Elasticsearch는 기본적으로 인덱스 단위로 데이터를 관리함
    • 인덱스가 클러스터의 노드에 분산되어 저장되기 때문에 대량의 데이터를 다룰 수 있게 됨. 인덱스를 노드에 분산하기 위한 단위가 샤드
  • 타입(인덱스타입)
    • 인덱스에 등록하는 도큐멘트를 논리적으로 분류하기 위한 기능
    • 타입을 사용하면 인덱스에 연러 가지 종류의 도큐멘트를 저장하더라도 관리하기 쉬움
  • 샤드(세그먼트)
    • 작은 단위로 분할한 인덱스를 샤드
    • 다른 데이터 저장소에서는 파티션 등으로 불리기도 함. Elasticsearch는 이 샤드를 클러스터의 각 노드에 할당해서 분산시킴
  • 프라이머리 샤드/레플리카 샤드
    • Elasticsearch는 데이터 등록의 요청이 오면, 프라이머리샤드에 데이터를 저장하고나서 레플리카 샤드에 데이터를 복사함. 레플리카 샤드는 프라이머리 샤드의 복사본
  • 매핑
    • 인덱스에 저장되는 데이터의 구조를 정의하기 위해서 매핑(Mapping)이란 것을 이용함
  • 역인덱스(Inverted Index)
    • Elasticsearch의 인덱스는 Apache Lucene을 사용해서 데이터를 관리함. Apache Lucene은 전문 검색을 하기 위해 데이터를 역인덱스로 변환해서 저장하고 검색함
    • 역인덱스는 입력한 문장을 특정 기준으로 분할한 뒤, 분할한 단어가 어느 도큐먼트에 있는지를 쉽게 알 수 있도록 단어와 도큐먼트 ID를 매핑한 색인
• 로그 데이터를 저장하는 단위
  • 전문 데이터(문장 등)를 검색하기 위해서 검색하려는 데이터의 묶음별로 인덱스를 만들어서 보관함

Kibana 특징

• 주요기능
  • Kibana에는 시각화, 분석도구로서 기능
    • 저장되어 잇는 로그의 검색,필터링
    • 여러 가지 형식의 시각화
    • 대시보드의 작성, 저장, 공유
    • 시계열, 그래프 등의 분석 기능
  • Kibana에서는 Elasticsearch를 백엔드로 사용하고 있기 때문에 검색, 추출에는 Lucene이 지원하는 검색쿼리를 사용함. Web UI를 사용해서 집계처리 등의 설정을 하기 때문에 특수한 쿼리를 사용하지 않고도 간단한 집계, 그래프를 만들 수 있음
• Kibana 시스템 구성
  • Kibana와 Elasticsearch
    • Elastic Stack
      • 데이터 저장소 : Elasticsearch
      • 로그 수집, 가공, 중계 : Logstash
      • 경량 데이터 수집 에이전트 : Beats
      • 시각화, 분석 도구 : Kibana
  • 데이터 저장
    • 현재 상태를 가지고 있지 않기 때문에 이중 구성이나 업그레이드도 비교적 간단함. 하나의 Elasticsearch 클러스터에 대해서 여러 개의 Kibana를 실행해두고, 로드밸런서에서 요청을 나누는 것만으로도 이중화 구성을 할 수 있음
• 그 외의 시각화, 분석 도구의 비교
  • Redash
    • 여러 가지 데이터소스를 백엔드로 사용할 수 있는 분석 대시보드
    • RDBMS와 Presto,Hive 그리고 Redshift, BigQuery라는 서비스를 데이터소스로 사용하는 것도 가능하며, 쿼리 결과를 시각화하여 대시보드로 표시할 수 있음
    • Redash와 Kibana의 큰 차이점은 특정 데이터소스의 의존성. Kinbana는 Elasticsearch의 데이터만 대상으로 하지만 Redash는 여러 가지 데이터소스를 대상으로 쿼리를 실행할 수 있고 그 결과를 Redash의 데이터베이스에 저장함
    • Redash는 여러 가지 데이터소스를 대상으로 하지만, 한 편 쿼리를 실행하고 시각화하기 위해서 각각의 데이터소스에 맞는 쿼리를 작성할 필요가 있음. RDBMS에는 SQL 쿼리, Presto에는 PrestoQL을 직접 작성함. Elasticssearch도 백엔드로 사용할 수 있지만 JSON 기반의 쿼리를 직접 작성해야 함
    • Kibana는 Elasticsearch와 직접 연동하기 때문에 거의 쿼리를 작성할 일ㅇ 없이 브라우저상에서 집계, 시각화를 할 수 있음. 실시간이며 인터랙티브한 분석에는 Kibana가 더 어울림. SQL로 작성하고 정기적으로 결과를 뽑는 처리를 구현하기에는 Redash가 더 좋음
  • BI 도구
    • BI 도구는 Business Intelligence, 경영상의 분석, 판단을 위한 도구로 발전해 온 분석 애플리케이션
    • 예전에는 독립적으로 동작하는 데스크탑 애플리케이션인 경우가 많았고, 높은 전문성으로 인해 비싼 가격의 제품이 많은 분야였음. 최근에는 서버에서 동작하는 애플리케이션과 클라우드 서비스의 기능으로 제공되는 도구도 많아져서 웹 서비스의 분석에도 이용될 정도로 턱이 낮아졌음
    • Tableau
    • Qlik View
      • Qlik사에서 제공하는 인메모리형 BI 도구. Kibana와 같은 웹 기반의 UI를 가지고 있음
    • Domo
      • Domo사가 제공하는 인메모리형 BI 도구. BI 도구 뿐만 아니라 콜라보레이션웨어 등의 기능도 있음
    • Google Data Studio
      • Google사가 제공하는 BI 도구. Google BigQuery, Cloud SQL과 Google Analytics, Google 스프레드시트 등 Google의 각종 서비스와 간단하게 결합해서 사용함
    • Amazon QuickSight
      • Amazon사에서 제공하는 BI도구. AWS 각 서비스와 통합해서 이용할 수 있음

Kibana 분석 워크플로우

• 매트릭 시각화
  • 서비스 개선을 하기위해서는 일단 각종 매트릭을 시각화할 필요가 있음
    • 페이지뷰
    • 유니크유저수
    • 투고수
    • 특정 키워드의 검색회수
  • Kibana의 차트 설정은 크게 [Metrics]와 [Buckers]으로 나눔. [Metrics]는 차트에 표시하는 값을, [Buckets]는 그것을 어떻게 분류할 것인지를 지정함

대시보드

• 대시보드 공유
  • 공유하는 방법으로는 대시보드 자체의 URL을 공유하는 방법(saved dashboard)과 현재 선택되어 있는 기간 등의 상태를 공유하는 방법(snapshot)이 있음
  • 대시보드의 링크를 전달할 때는 saved dashboard를 어떤 순간의 결과를 공유할 때는 snapshot의 URL을 이용하도록 함

Fluentd 플러그인 사전

• Fluentd는 로그/메시지의 수집을 하는 동시에 태그와 라벨을 붙이고, 조건에 따라 필터 플러그인으로 데이터를 가공,집계(태그의 변경에 동반하는 처리는 디렉티브를 사용하는 Output 플러그인이 하고, 태그의 변경을 사용해서 라우팅을 함)하며, 최종적인 저장소에 로그/메시지를 저장함
• 빅데이터
  • td - 클라우드형 Hadoop 서비스인 TreasureData에 레코드를 보냄
• Shadow Proxy를 사용한 에러의 조기 감지
  • 가동중인 시스템의 HTTP 요청의 내용을 가지고, 실제 서비스 환경에 반영하기 전에 스테이징 서버에 적용하는 것을 Shadow Proxy(셰도우 프록시)라고 함
  • 자동테스트로 품질관리를 하고 있지만, 실제 환경에서는 웹 서버 뿐만 아니라, 데이터베이스와 캐시 서버 등으로 복잡하게 구성되어 있기 때문에 새로운 배포에서 예상치 못한 에러가 발생할 수도 있음. 반영하는 타이밍에 발생하는 장애에 대응하기 위해서는 Shadow Proxy 서버가 도움이 됨

Embulk & Digdag

• 현대의 데이터 처리 워크플로우 만드는 방법
  • 일반적인 데이터 처리 워크플로우
    • 수집(Ingest/Collect) - 애플리케이션 로그, 유저 속성 정보, 광고의 인상, 서드파치쿠키
    • 전처리(Enrich) - 봇의 액세스 로그 제외. IP 주소로 위치 정보 추가, user-agent의 구조화, 마스터 데이터를 사용해서 로그에 유저 속성 추가
    • 분류, 집계, 분석(Model) - 데이터베이스에 추가, 분석 처리 시스템으로 전송, 압축해서 스토리지에 저장(아카이브), 통계 데이터로 기록
    • 활용(Utilize) - 추천 엔진 API의 참조 데이터, 실시간 거래, BI 애플리케이션을 사용한 시각화
• Fluentd는 준실시간의 로그 활용을 할 때 첫 장애물인 로그 수집의 과제를 해결하기 위해 개발됨
• 각종 데이터의 입력과 수집을 Input 플러그인으로 지원하고, 데이터 가공을 Filter 플러그인으로 처리하며, Output 플러그인으로 여러가지 미들웨어나 스토리지로 저장할 수 있음
• 2가지 과제
  • Fluentd의 설정 파일의 거대화. 가끔 일어나는 사업적인 사양 변경에 따라 설정 파일의 라인수가 증가하여 유지보수가 힘들게 되었음
  • 스트리밍 처리에 특화된 Fluentd는 정기적으로 벌크로드를 하여 데이터 처리 워크플로우를 만들기에는 적합하지 않음
• 데이터 워크플로우를 지원하는 도구
  • 스트리밍 데이터 컬렉터 - fluentd - 액세스 로그 / 앱로그 / 서버로그
  • 벌크 데이터 로더 - embulk - csv파일/ S3 / MySQL / PostgreWQL 등
  • 워크플로우 관리 - digdag - ETL 처리의 자동화
• Embulk
  • Embulk는 Fluentd와 같이 Input/Filter/Output 플러그인을 조합해서 설정 파일을 정의함. 병렬 분산 처리에 대응한 성능과 재시도 제어 등에 안정성이 우수한 데이터 전송 파이프라인을 만들수 있음
  • 다량의 데이터를 효율적으로 읽어서 CPU 코어를 최대한 사용해서 배치 처리하는 데 특화되어 있음
  • 데이터베이스와 스토리지에서 데이터를 읽어서, 임의의 처리를 한 뒤에 다른 보관 장소로 보내는 데이터의 대용량처리에 특화된 ETL 처리 도구
  • Fluentd와 다른 특징으로 고속성과 트랜잭션 제어, 스키마를 사용한 데이터의 검사 기능이 있음
  • 대용량의 데이터를 마이크로 배치처리로 Redshift, BigQuery, Elasticsearch로 저장하는 경우라면 Fluentd 보다는 병렬처리, 처리량, 저장타이밍을 자융롭게 컨트롤할 수 있는 Embulk로 저장하는 편이 확실히 안정적
• Digdag
  • Digdag는 워크플로우의 정의를 설정 파일로 하고 있음. Embulk와 임의의 셸스크립트에 임의의 변수를 넣어가며 ,의존 관계순으로 직렬 및 병렬 처리로 Job을 실행할 수 있음
  • 워크플로우관리 도구로서 ETL 처리의 자동화에 도움이 됨
  • 여러 단계에서의 처리의 의존관계와 순서 ,병렬실행 등을 프로그램 가능한 YAML 설정 파일을 통해 제어할 수 있는 아키텍처
  • 여러 개의 데이터소스로부터 병렬 또는 직렬로 데이터를 읽고, 날짜별로 테이블을 만들고 저장하며 지속적인 1차 집계를 한 뒤에 그 결과를 저장하는 처리를 직관적으로 설정 파일에 설정할 수 있음
  • 기본기능
    • 작업을 의존관계순으로 실행
    • 과거분의 일괄실행(backfill)
    • 정기 실행
    • 시간 등의 변수를 포함해서 실행
    • 파일이 생성되면 실행
  • 에러처리
    • 실패하면 통지
    • 실패한 위치에서 재시작
  • 상태 감시
    • 실행 시간이 일정 이상이면 통지
    • 작업의 실행시간의 시각화
    • 실행 로그의 수집과 저장
  • 고속화
    • 작업을 병렬로 실행
    • 동시 실행 작업 개수의 제어
  • 개발지원
  • 워크플로우의 버전 관리
    • GUI로 워크플로우 개발
    • 정기처리를 간단하게 실행할 수 있는 라이브러리
    • Docker 이미지를 사용해서 작업 실행
  • Digdag에는 스케줄러를 내장하고 있어서 데몬으로 동작하는 서버 모드와 커맨드라인에서 임의로 실행하는 로컬 모드의 2가지가 있음. 서버 모드에서는 기밀 정보와 폴리시 파일을 커맨드라인으로 등록함
  • Digdag을 실행하는 커맨드

    • digdag run stage1_load_assets.dig

  • 병렬 실행 수의 상한을 설정하여 Digdag의 워크플로우를 실행

    • digdag run example.dig --max-task-threads 4

  • 편리한 오퍼레이터
    • 파일이 나타날 때까지 계속하는 s3_wait>:와 gcs_wait>:라는 오퍼레이터
    • AWS S3에 파일이 생성될 때까지 기다렸다가 생기면 다음 태스크로 가서 Redshift에서 가져오는 Digdag의 설정. BigQuery에서 가져오는 것도 bq_load>: 오퍼레이터를 사용하여 만들 수 있음
    • rb>: 오퍼레이터와 py>: 오퍼레이털르 사용하면 데이터를 읽는 것외에도 데이터의 가공도 할 수 있음

    • $ cat tasks/__init__.py
      # coding:utf-8
      import digdag
      import pandas as pd
      
      class Convert(object):
        def __init(self):
          pass
        
        def transform(self, session_time = None, query_result=' 0'):
          input = pd.read_csv(digdag.env.params['input_csv']
          user_info = pd.read_csv(digdag.env.params['user_info']))
          combined = pd.merge(input, user_info, how='left', on=['id'])
          combined.to_csv(digdag.env.params['output_csv'], index=False, encoding='utf-8')
      
      # Python으로 데이터 가공 처리를 하는 예제
      $ cat conver_csv.dig
      _export : 
        input_csv: path/to/input.csv
        output_csv: path/to/output.csv
        user_info: path/to/users.csv
        td:
          database: www_access
      
      +extract:
        td>: queries/sample_query.sql
        download_file: ${input_csv}
        
      +transform:
        py>: Convert.transform
      
      +load:
        # 남은 csv 파일을 전송함(Embulk 설정은 생략)
        sh>: embulk run embulk_load_csv_bigquery.yml.liquid