[AWS] Kinesis

Kinesis는 Kafka의 aws 버전이라고 할 수 있다.

용도 자체는 카프카를 그대로 올려주는 AWS MSK와 동일하다.
다만 AWS에서 자체적으로 관리해주고 추상화된 부분이 많기 때문에 더 쉽고 편하게 사용할 수 있다. 게다가 다른 AWS 리소스들과의 연계도 잘 지원된다.

기본적으로 클러스터와 브로커를 관리할 필요가 없는데다, 게다가 오토스케일링 증가/감소가 둘다 자동으로 된다. MSK는 증가만 된다.

온디맨드 요금이 있단것도 장점이다.

구조 및 제한

stream은 kafka의 topic에 대응되는 용어고, shard는 kafak의 partition에 대응된다고 보면 된다.

프로비저닝 모드의 경우에는 샤드 단위로 스케일링이 가능하다. 사용량이 늘면 샤드를 더 띄워서 부하 분산을 하는 것이다.
샤드는 계정당 200개가 지원되는데, 요청하면 늘릴 수 있고. 샤드당 1000TPS로 데이터를 때려박을 수 있다.
샤드를 5개 띄워두면 1초에 5000개까지 박을 수 있는 것이다.

온디맨드 모드는 200000TPS로 데이터를 박을 수 있다. 그 이상 규모로 늘어날 수 있다고 가정한다면 프로비저닝 모드를 사용하는 편이 좋을 것이다.

비용

자세한건 아래 페이지를 참조한다.
https://aws.amazon.com/ko/kinesis/data-streams/pricing/

프로비저닝 모드는 이런식으로 샤드당 요금과 데이터를 몇개나 넣었는지를 기준으로 측정한다.

생각보다 그렇게 비싸지는 않았다. 샤드가 하나만 떠있다면 샤드 비용은 한달에 14달러 정도만 나온다.

온디맨드는 스트림 단위, 그리고 얼마나 데이터를 넣고 읽어왔는지로 비용을 부과한다.

이건 규모가 매우 작을때는 온디맨드가 오히려 비쌀 것 같다.

스트림 만들기

기본 데이터 스트림으로 먼저 만들어보겠다.

이름 잘 짓고, 온디맨드로 할지 프로비저닝으로 할지만 정하면 된다.

나는 프로비저닝으로 하겠다.

그래서 이런 모양으로 만들어지면 일단 생성은 된 것이다.

데이터 쌓아보기 (Producer)

aws-sdk의 putRecord를 이용하면 별로 어려울 것 없이 데이터를 쌓을 수 있다.
나는 Lambda로 작성해서 돌렸다.

const aws = require('aws-sdk');

const kinesis = new aws.Kinesis();

async function putEvent(event) {
    await kinesis.putRecord({
      Data: JSON.stringify(event),
      PartitionKey: '1', 
      StreamName: 'test_queue',
    }).promise();
}

exports.handler = async (event) => {
    await putEvent({
      user_id: 1,
      product_id: 1,
    });

    await putEvent({
      user_id: 2,
      product_id: 1,
    });

    await putEvent({
      user_id: 3,
      product_id: 1,
    });

    await putEvent({
      user_id: 1,
      product_id: 2,
    });

    const response = {
        statusCode: 200,
        body: JSON.stringify('Hello from Lambda!'),
    };
    return response;
};

파티션 키는 해싱해서 어떤 샤드에 넣을지를 결정할때, 그 해시 키값으로 사용되는 값이다.
중복되지만 않게 넣으면 되고, 그때그때 만든 requestId같은걸 넣어도 된다.

그러면 이렇게

콘솔에서도 확인을 할 수 있다. "시작 위치"는 트림 호라이즌으로 해야 전체목록을 볼 수 있다.

데이터 읽어보기 (Consumer)

이번엔 Lambda를 이용해서 쌓아둔 것들을 읽어보겠다. 쓰는것보다 읽어들이는게 좀더 복잡하다.

const aws = require('aws-sdk');

const kinesis = new aws.Kinesis();

async function getShardIterator(ShardId) {
    return (await kinesis.getShardIterator({
        ShardId, 
        ShardIteratorType: 'TRIM_HORIZON',
        StreamName: 'test_queue',
    }).promise())?.ShardIterator;
}

async function getRecords(shardIterator) {
    return (await kinesis.getRecords({
      Limit: 10,
      ShardIterator: shardIterator,
    }).promise())?.Records;
}

async function getShards() {
    return (await kinesis.listShards({
        StreamName: 'test_queue',
    }).promise()).Shards;
}

exports.handler = async (event) => {
    const shards = await getShards();
    console.log(shards);

    const shardIterator = await getShardIterator(shards[0].ShardId);
    const records = await getRecords(shardIterator);

    for(const record of records) {
        let data = Buffer.from(record.Data).toString();
        console.log(data);
    }

    const response = {
        statusCode: 200,
        body: JSON.stringify('Hello from Lambda!'),
    };
    return response;
};

먼저 읽을 Shard를 결정하고, ShardIterator를 만든 다음에 getRecords를 통해서 가져올 수 있다.

그래서 consumer는 shard를 각자 하나씩만 담당해서 처리하는게 일반적이다.
근데 shard에 맞춰서 consumer도 오토스케일링을 하려면.. 그것도 꽤 손이 많이 간다. 그래서 일반적으로 Lambda로 트리거를 걸어서 처리하는게 일반적인듯하다.

ShardIteratorType 몇가지 타입이 있는데, 대충 이렇다.

1. AT_SEQUENCE_NUMBER: 전달된 시퀀스 번호부터 순차적으로 읽어옴.
2. AFTER_SEQUENCE_NUMBER: 전달된 시퀀스 번호 이후부터 순차적으로 읽어옴.
3. AT_TIMESTAMP: 전달된 타임스탬프 시점부터 스트리밍을 순서대로 읽어옴
4. TRIM_HORIZON: 가장 오래된 데이터부터 읽어옴
5. LATEST: 샤드의 가장 최근 레코드 바로 다음부터 읽어옴. (Iterator를 만든 시점부터 새로 들어온 것만 받음)

레코드는 임의로 삭제할 수가 없다. 설정된 보존 기한이 지나면 자동으로 삭제될 뿐이다.
그래서 현실적으로 TRIM_HORIZON 같은건 쓸일이 딱히 없을 것이다.

어쨌든 저대로 실행하면

가져오긴 잘 가져온다.

Lambda로 Consumer 구축

AWS Lambda를 사용하면 데이터 consume 처리를 꽤나 간편하고 쉽게 처리할 수 있다.
consumer용 함수를 먼저 만들고,

함수 역할 권한에 키네시스 접근권한을 넣어준다.

그리고 트리거로 키네시스를 추가해주면 된다.

스트림 고르고, 배치 크기를 정한다. 저러면 한번에 100개씩 모아서 들어온다.
그리고 최신 항목들만 순서대로 들어오도록 했다.

그럼 이렇게 추가될 것이다.

받아먹을 consumer 코드를 작성해보자.
일단 로그만 찍게 했다.
Records라는 항목으로 들어오고, data는 base64로 들어와서 디코딩을 해줘야 한다.

exports.handler = async (event) => {
    const records = event.Records;

    console.log('레코드 들어옴!!');

    for(const record of records) {
       record.kinesis.data = Buffer.from(record.kinesis.data, "base64").toString('utf8');

       console.log(record);   
    }

    // TODO implement
    const response = {
        statusCode: 200,
        body: JSON.stringify('Hello from Lambda!'),
    };
    return response;
};

그리고 producer로 레코드를 넣어주면

트리거로 람다가 호출되면서 실행이 될 것이다.

저걸 필요에 따라 적절히 사용하면 된다.

다른 방법: KCL

KCL은 Kinesis Client Library의 축약인데, 이런 shard별 consumer의 관리 같은 정형화된 구조들을 어느정도 제공해주는 라이브러리다.
잘 지원되는건 Java, Python 정도인 것 같다.

이런건 추후 별도 포스트로 다뤄보겠다.

참조
https://jaemunbro.medium.com/aws-%EB%A9%94%EC%8B%9C%EC%A7%95%EC%84%9C%EB%B9%84%EC%8A%A4-%EB%B9%84%EA%B5%90-kinesis-sqs-sns-ab397a07cb1d
https://docs.aws.amazon.com/ko_kr/streams/latest/dev/service-sizes-and-limits.html