[AWS_5] AWS kinesis

AWS

AWS kinesis?

  • Kinesis Stream: 대규모 데이터 레코드 스트림을 실시간 수집/처리(Amazon Kinesis is an AWS service that collects, processes, and analyzes data in real time.)

(이벤트/log 등)수집 단계에서 유의점

  • lambda 사용하지 말자
    • 프로그램(coding)없이 기존 서비스로 저장 가능하다면 해당 서비스 사용
    • eg. API Gateway, Kinesis Stream, Firehose -> 수집하여 -> S3에 넣기(RDBMS (X))
  • 수집은 실시간(event 등)
  • cf. ETL Tool: 테이블 옮기기 위해서는 DMS tool사용

  • Data Collect: Kinesis, Firehose, Amazone API Gateway
    • 온프레미스 경우, DB에 다 넣음(RDBMS) by Sqoop -> 이후 병렬작업해서 S3로 내림
    • 수집pipline은 한번 Set-up해두면 새로운 channel 추가는 빠름
    • 협력업체 데이터도 수집필요
    • S3에 쌓기
  • 전처리: EMR + Glue/S3 -> 의미있는 값의 집합(전처리된 데이터). 이후 S3에 저장.
    • Spark 처리하려면 S3안에 있어야 함
      • Why? RDB에 넣어두면 여러명이 접근하면 부하걸림 S3에 다 넣기
      • eg. SQL join 등등
  • 분석 및 시각화:
    • Data Mart: 각 부서에서 원하는 데이터 구성(A single subject or functional organization area)
    • 하둡 테이블로 저장해두면 red shift, AWS Athena 등이 S3에 있는 데이터를 읽어옴 -> tableau, Periscope data, Apache Superset, Zeppelin

2) Kinesis, Firehose, Amazone API Gateway & S3

  • 상기 조합의 장점
    • S3 저장공간 무제한
    • kinesis Stream에서 데이터 24시간 보존 -> 바로 가공 가능
    • S3에 반정형화된 json형식으로 데이터 저장 -> 가변적 데이터 수집, 유연한 분석 가능

AWS 실습

1) kinesis

  • shard 숫자 설정
  • Delivery stream name = kinesis name
    • 운영시 압축확장자: gzip

  • kinessis firehose에 S3 버킷 지정: raw-data/bhr-dees2/

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    # send sth to kinesis from ec2(S3에 저장)
    $ aws kinesis put-record --stream-name [kineisis name:bhr-dee2] --partition-key [option: 123] --data [data:bhr_testdata123] aws configure

    # data 확인
    $ SHARD_ITERATOR=$(aws kinesis get-shard-iterator --shard-id shardId-000000000000 --shard-iterator-type TRIM_HORIZON --stream-name bhr-dee2 --query 'ShardIterator')
    aws kinesis get-records --shard-iterator $SHARD_ITERATOR

    # 하기 링크에사 base 64 디코딩
    http://www.convertstring.com/ko/EncodeDecode/Base64Decode
    >> result: [data:bhr_testdata123]

  • record하면 ec2 -> kinesis로 쏴주는 것.

  • 조회는 kinesis로 작성한 내용 조회
  • kinesis monitoring: get_records 그래프로 작성여부 확인가능

2) API gateway

  • 모든 event를 API gateway 통해서 받으면 가격이 비쌈
  • kinesis agent 가 log 정리/취합
  • create resource
  • data receive method: http 방식
    • 어떤 method? post 방식
      • get은 파일 크기가 정해져 있음. 이슈발생 가능성 있음.

  • send data to kinesis from API gateway

    • 헤더값을 보고 인식
    • intergration: 들어온 데이터 어떻게 처리/관리 (언어: velocity(context handling))
    • deploy 적용
      1
      2
      # json형식 data를 kinesis로 발송
      $ curl -d '{"key1":"value1", "key2":"value2"}' -H "Content-Type: application/json" -X POST  https://7ud79aqdo7.execute-api.ap-northeast-2.amazonaws.com/prod/v1

  • cloud watch: API ID 사용(eg. 7ud79aqdo7)

    • API 설정시 enable cloudwatch log 설정
    • throattling setting 필요: 10k개 이상되면 뱉어내기 때문.
  • similar to “horizontal partitioning” of DB
    • horizontal partitioning이외에 여러개 instance에 대해서도 작업 가능 (does this across potentially multiple instances of the schema.)
  • Horizontal partitioning: DB table을 column이 아닌 row 단위로 partitioning하여 따로 관리(Horizontal partitioning is a design principle whereby rows of a database table are held separately, rather than splitting by columns (as for normalization))
  • 각 partition이 shard의 부분(Each partition forms part of a shard, which may in turn be located on a separate database server or physical location)
  • search perfomance 향상
  • consumer 중 하나로서, 자동으로 데이터를 S3/redshift(AWS의 DW)/AWS ES(elastic search)/Splunk로 보내기 위한 관리형 서비스
  • Amazon Kinesis Data Firehose is a fully managed service for delivering real-time streaming data to destinations such as Amazon Simple Storage Service (Amazon S3), Amazon Redshift, Amazon Elasticsearch Service (Amazon ES), and Splunk.
    • 이전에는 Lambda가 위 역할수행
  • Firehose가 없다면 큐는 인덱싱없이 FIFO만 구현
  • Firehose는 큐의 인덱싱을 관리하는 역할

  • Destination: S3, RedShift, AWS ES, splunk
  • 유용한 기능
    • 데이터를 받아서 서로 다른 bucket에 저장 가능
    • 여러 consumer로부터 데이터 수집/관리 가능

질문 1)
서버에 있는 로그는 어떻게 수집?

  • kinesis는 서버에 쌓여있는 로그를 읽어와서 다시 저장.
  • 폴더를 바라보고 있으면 데이터가 쌓일 때마다 쏴줌
  • EC2가 kinesis 바라보고 있으면, kinesis로 계속 쏴줌

질문 2)

  • 실시간 분석은 DB가 아니라 큐(kafka: 하드디스크 queue…)에 넣어야 함.
  • 큐에 넣어두면, 큐에 데이터가 존재하는 동안 여러 consumer(ec2 instance -> S3/Rdeshift/EMR …)가 사용할 수 있음
  • 그렇지 않으면, DB connection, JDBC등 사용하여 접속하여 다시 접근하므로 cost발생.

질문 3)
비정형데이터를 정형화 데이터로 바꾼다는 의미?

  • raw Data -> transformed Data

질문 4)

  • AWS 안쓰고 온프레미스라면? 최근 2년 기간 데이터는 oracle, 나머지는 하둡.

질문 5)
왜 큐에다 쏘는가?

  • 큐는 데이터 처리의 완충작용
  • 앱 구동시, RDBMS에 insert하면서 5초 소요된다면? 큰 문제.
  • 때문에 로그 처리는 바로 큐로 날려줌 -> 앱 사용에 딜레이 발생X
  • 큐에 던지면 큐가 일단 받음.
  • 앱에서만 이용하는 이벤트도 큐로 받음.

질문 6)
AWS의 S3와 Spark 조합을, 온프레미스 MySQL과 Spark 조합으로 대체 가능한가?

  • 성능차이 발생. S3를 수백대의 Slave가 바라봐도 문제없음. MySQL은 과부하 발생

기타 1)

  • AWS 서비스별 API 존재. kinesis Stream의 Shard 자동 조절 가능.
  • raw data는 json형식으로. Parquet은 바로 안됨.
< !-- add by yurixu 替换Google的jquery并且添加判断逻辑 -->