스트리밍 데이터 (Kinesis Data Streams, Data Firehose, MSK)

무엇을 고르나

Firehose 는 2024-02 에 Kinesis Data Firehose 에서 Amazon Data Firehose 로, Flink 서비스는 2023-08 에 Kinesis Data Analytics 에서 지금 이름으로 바뀌었습니다. 오래된 블로그 글을 볼 때 이름이 달라도 같은 서비스입니다.

Kinesis Data Streams

샤드와 용량

스트림은 샤드로 나뉘고, 레코드는 파티션 키의 해시로 샤드에 배정됩니다. 같은 파티션 키의 레코드는 같은 샤드에 들어가 순서가 지켜집니다.

데이터 보존 기간은 기본 24 시간이고 최대 365 일까지 늘릴 수 있습니다(늘리면 추가 요금). 재처리가 필요하면 7 일 정도로 늘려 두는 것이 안전합니다.

파티션 키 설계

파티션 키가 한쪽으로 쏠리면 샤드 하나만 ProvisionedThroughputExceededException 을 내고 나머지는 놉니다. 사용자 ID, 기기 ID 처럼 값이 많은 키를 쓰고, 순서가 필요 없다면 무작위 값을 씁니다.

import boto3, json, uuid
kds = boto3.client("kinesis", region_name="ap-northeast-2")
kds.put_records(
    StreamName="clickstream",
    Records=[{"Data": json.dumps(e).encode(), "PartitionKey": e.get("userId") or str(uuid.uuid4())} for e in events],
)
# 응답의 FailedRecordCount 가 0 이 아니면 실패한 레코드만 골라 재시도해야 합니다

PutRecords 는 일부만 실패해도 HTTP 200 을 돌려줍니다. 응답의 FailedRecordCount 를 확인하지 않으면 데이터가 조용히 사라집니다.

Lambda 로 소비하기

Lambda 의 이벤트 소스 매핑(ESM)은 샤드마다 배치를 읽어 함수를 호출합니다. 운영에서 꼭 정하는 설정은 다음과 같습니다.

def handler(event, context):
    failures = []
    for r in event["Records"]:
        try:
            process(r)
        except Exception:
            failures.append({"itemIdentifier": r["kinesis"]["sequenceNumber"]})
    return {"batchItemFailures": failures}

주의 재시도 설정을 기본값(레코드가 만료될 때까지 재시도)으로 두면, 독이 든 레코드 하나 때문에 그 샤드의 처리가 보존 기간 내내 멈춥니다. IteratorAge 지표가 계속 오르면 이 상황을 먼저 의심합니다.

Amazon Data Firehose

Firehose 는 샤드도 소비자 코드도 없습니다. 소스(Direct PUT, KDS, MSK)에서 받아 버퍼링한 뒤 목적지에 씁니다.

• 목적지: S3, Apache Iceberg 테이블(S3 Tables 포함), Redshift, OpenSearch, Splunk, HTTP 엔드포인트, 여러 SaaS.
• 형식 변환: JSON 을 Glue 카탈로그 스키마에 맞춰 Parquet, ORC 로 바꿔 저장합니다. Athena 비용이 크게 줄어듭니다.
• 동적 파티셔닝: 레코드 안의 필드(customerId, 날짜)로 S3 접두사를 나눕니다.
• Lambda 변환: 필드 정리, 마스킹 같은 간단한 변환을 끼울 수 있습니다.
• 버퍼는 크기(MB)와 시간(초) 중 먼저 차는 쪽에서 비웁니다. 버퍼가 작으면 작은 파일이 많이 생겨 분석이 느려집니다.

aws firehose create-delivery-stream --delivery-stream-name alb-logs-to-s3 \
  --delivery-stream-type DirectPut \
  --extended-s3-destination-configuration '{
    "RoleARN":"arn:aws:iam::111122223333:role/firehose-to-s3",
    "BucketARN":"arn:aws:s3:::lake-prod",
    "Prefix":"raw/alb/dt=!{timestamp:yyyy-MM-dd}/",
    "ErrorOutputPrefix":"errors/alb/!{firehose:error-output-type}/",
    "BufferingHints":{"SizeInMBs":128,"IntervalInSeconds":300},
    "CompressionFormat":"GZIP"}'

실패한 레코드는 ErrorOutputPrefix 아래에 쌓입니다. 이 접두사를 정하지 않거나 아무도 보지 않으면 데이터 손실을 늦게 알게 됩니다.

Amazon MSK

• 인증은 IAM 액세스 제어를 기본으로 씁니다. 토픽, 그룹 단위 권한을 IAM 정책으로 줄 수 있어서 Kafka ACL 을 따로 관리하지 않아도 됩니다.
• 클러스터는 세 AZ 에 브로커를 고르게 두고, 토픽은 복제 계수 3, min.insync.replicas=2 로 만듭니다.
• 브로커 간 복제 트래픽은 AZ 간 전송이지만 MSK 는 이 복제 트래픽에 전송 요금을 받지 않습니다. 반면 클라이언트와 브로커가 다른 AZ 에 있으면 일반 AZ 간 전송 요금이 붙습니다. 소비자에 client.rack 을 설정해 같은 AZ 의 복제본에서 읽게 하면 이 비용이 줄어듭니다.

증상, 원인, 해결

비용 팁

• 하루 몇 번만 오는 소량 데이터에 프로비저닝 KDS 를 상시 켜 두지 않습니다. 온디맨드나 Firehose Direct PUT 이 싸고 단순합니다.
• 분석용 적재만 필요하면 KDS 를 거치지 말고 Firehose 로 바로 넣습니다.
• 향상된 팬아웃은 소비자, 샤드 시간당 요금이 붙습니다. 지연에 민감한 소비자에게만 씁니다.
• MSK 는 보존 기간이 길면 계층형 스토리지로 오래된 세그먼트를 싼 저장소로 옮깁니다.