개시하다

Kiness Data Firehose로 새로운 프로젝트의 로그 통계를 설계했지만
아직 안 써봐서 써봤다면

Kiness Data Firehose

말하자면 Kinesis는 이렇다.
Kiness는 스트리밍 데이터를 실시간으로 수집, 처리 및 분석할 수 있습니다.
그중에도 네 개가 더 있어요.

Kiness Data Streams: 스트리밍 데이터 수신 서비스

Kiness Data Firehose: 흐르는 데이터를 S3 또는 Redshift 등 분석 도구에 전송하는 서비스

Kiness Data Analytics: 스트리밍 데이터 분석 서비스

Kinesis Video Streams: 간단하고 안전하게 영상을 전송하는 서비스

Kiness Data Streams와 Firehose의 차이점

Streams는 쉽게 말하면 흐르는 데이터만 받는 서비스입니다.
수신 후 Lambda나 어떤 활동에 불을 붙여 흐르는 데이터를 수신하게 할 뿐이다.
Firehose는 데이터를 받은 후 S3과 Redsift로 데이터를 흘립니다.
Streams는 받아들일 뿐이기 때문에, 예를 들면 Streams->Firehose->S3
또 Firehose가 1분 뒤 S3에 뱉는 등 사용 상황에 따라'데이터가 어느 정도 축적됐다'는 임계값을 받을 수 있고, Streams는 1초 이하면 데이터를 로드할 수 있다.
Firehose는 엔드포인트(S3의 경로)와 데이터(덤프된 json 등)만 지정하면 재생이 가능하기 때문에 GUI에서 쉽게 설정할 수 있다.

별말 말고 해 보아라

1. 콘솔 화면에서 Kinesis와 검색

2. 대시보드를 사용하여 "메시지 흐름 만들기" 누르기

3. Delivery stream name의 임의의 이름으로 Choosea souece 유형을 선택합니다.

Streams->Firehose 프로세스가 아니므로 Direct Put or other sources 에서 확인
pytohon에서boto3까지Firehose에 데이터를 투입할 때other sources입니다.

4. Transform source records with AWS Lambda, Convert record format 설정

Transform source records with AWS Lambda: 스트림 데이터를 저장하기 전에 Lambda를 통해 편집할 수 있습니다.
예를 들어 Hello라는 문자열을kinsis Firehose로 보내고 lambda로 수신하면'world'의 보충 기록으로 저장할 수 있다.
이번에 그대로 저장하려면 Disable 확인
Convert record format은 Enable에서 두 가지 형식으로 출력을 결정할 수 있습니다.

Apache Parquet

Apache ORC

다른 점은 ↓ 사진 참조

참조 URL: https://speakerdeck.com/chie8842/karamunahuomatutofalsekihon-2
데이터 분석 용도라면 특정 열에 초점을 맞춰 고속 통계를 내기 때문에 이런 실러캔스 형태가 적용된다.
Athena의 경우 Parquet 형식 통계를 작성하기 때문에 중요합니다.
한 번Athena 성능 튜닝 Tips 상위 10행복해요?그렇게 지도 모른다, 아마, 아마...
이번에는 데이터를 보내려고 하기 때문에 다이사블로 직접 보냅니다.

5. Select a destination 설정

아마존 S3를 넣을 예정이라 S3를 체크한다

6.S3의 구간 설정, Prefix 설정

S3의 세그먼트 지정 제작의 세그먼트 이름
프리픽스가 패스를 결정할 수 있는 것도 참고"Athena 성능 튜닝 Tips TOP10"로, 의식적으로 파티별로 나누면 사용자 정의로 설정된다.
예:
s3://athena-examples/flight/parquet/year=1991/month=1/day=1/
s3://athena-examples/flight/parquet/year=1991/month=1/day=2/

정교한 Prefix참조 공식
Prefix를 사용자정의할 때 Error Profix를 설정해야 한다는 점을 잊지 마십시오.

7. S3 buffer condition 및 압축/암호화 설정

S3 Put 1 파일의 용량 제한을 설정하거나 파일 토출을 위한 간격(Interval)을 설정할 수 있습니다.
이번에는 5MB 또는 60초로 파일 토출이 진행됐다.
이밖에 압축 방법은 GZIP, 스냅피, Zip 등 다양하고 S3의 encryption도 설정할 수 있다.

8.Role 설정

S3에 대해 Put을 수행해야 하므로 필요한 권한 설정을 수행합니다.
Create new Role을 클릭하여 새로운 IAM 역할을 만듭니다.(제작 시 선택)


설정 후 마지막 확인 화면이 되므로 "Create delivery stream"설정이 종료됨

마지막에 넣어볼게요.

pip install boto3에aws의client 라이브러리인boto3를 넣습니다.

import boto3
import uuid
import json

client = boto3.client(
    'firehose',
    aws_access_key_id='xxxxxxxxxxxxxxx',
    aws_secret_access_key='xxxxxxxxxxxxxxxxxxxx',
    region_name='ap-northeast-1'
)
data = {
    "musicId": 'xxxxxxxxxxxxxxxxxxxx',
    "userId": 1,
    "artistId": 1
}
response = client.put_record(
    DeliveryStreamName='a-honda-test-kinesis-firehose',
    Record={
        'Data': f'{json.dumps(data)}\n'
    }
)
print(response)

이렇게 보내주시면...

a-honda@hoge:~
$ python test.py
{'RecordId': 'ls1DjmHBiOXPtNe/7zpYc/6Zk8o8j6JVz5QkBeyGC3I8aFjf7Dv/CAQ7JnRw913ovghFSIMuEn3MOkkR5GK8QVRJCjvH6AmRw0SPUqcTE8jm5kcDxWu+91HeFr+PJlyFCgIG259Ig+DH7rsXTfHRNMFPt3/G7GquA0WUmSXDTBIOzRxnsQ1bNyN0mEHFrhOkbiXElZ1rL6s1WvJLddHF+qDpggjUCX2i', 'Encrypted': False, 'ResponseMetadata': {'RequestId': 'e5724431-a382-2e82-b3c8-8bbaf8b786fe', 'HTTPStatusCode': 200, 'HTTPHeaders': {'x-amzn-requestid': 'e5724431-a382-2e82-b3c8-8bbaf8b786fe', 'x-amz-id-2': 'zWNAiCZio6tiW5gGbNggHSLjA4EHKnu3RXE5umbYbdVx/ZraOY8bEj2H4AlF/C7R0rtk5fLt0Bm+aPyNiMFqZ9ihKmOxrCDO', 'content-type': 'application/x-amz-json-1.1', 'content-length': '257', 'date': 'Mon, 10 Feb 2020 03:26:03 GMT'}, 'RetryAttempts': 0}}

이런 느낌으로 보내주세요.
S3에 저장된 거 확인했어.

최후

솔직히 flentd와의 차이점을 조금 조사해서 사용하는 게 좋을 것 같아요.
flentd라면 aggregator를 따로 준비해야 하는데 관리로 유지하는 장점이 있다고 생각합니다.
참조 URL: 로그 관리의 모범 사례
fluentd -> kinesis firehose
Gunsey 씨 케이크에 이렇게 써있어요.
이 흐름을 전혀 몰라요(쓴웃음)
왜 flumentd를 끼웠을까...
또 고객으로부터 스트림 데이터를 직접 던질 수 있지만 부당한 짓을 하면 비용이 들 수 있다.
설정은 의외로 간단하지만, S3의 경로가 정해져 있기 때문에 S3의 카탈로그를 자세히 적어놓으면 딜리버리 스트림을 만들어야 하기 때문에 번거롭다(땀)
많지만 로그 관리가 깊다