2년 전 개선작업 시작때 기록

무려 2년 전 입사한지 얼마 지나지 않았을 때 작성한 건데 생각보다 잘 썼다..?? 저때 열심히 살았나보다.

서비스에 라이브 송출 플레이어가 추가되면서 실시간 데이터 처리를 맡게 되었다. 이전에 있던 다른 일반 VOD 재생 데이터들은 관계형 데이터베이스에서 DynamoDB를 사용하도록 프로세스가 수정된 상태였다.

구조

VOD 재생 데이터도 실시간 집계라면 실시간 집계였지만 라이브 서비스의 경우 현재 시청자수를 구해야해서 정말로 실시간 집계가 필요했다. 재생 데이터는 5초에 한 번 계속해서 들어오도록 되어있다.
기존에는 시간, 날짜 별로 데이터를 쌓고 있었다. 현재 시청자 수의 경우 더 짧은 집계 단위가 필요했고 이를 30초로 정해서 데이터를 쌓기 시작했다.
여기서 문제가 발생했는데, 순서가 보장되지 않는 SQS 일반 큐에서 재생 데이터를 저장하는 조건은 해당 uuid에 마지막 재생 시점보다 새로운 재생 데이터의 재생 시점이 이후일 경우이다.
날짜, 시간별 데이터의 경우 데이터가 유실될 경우가 없다고 봐도 됐는데 초 단위로 집계되는 데이터는 몇 초 차이로 집계가 되지 않아 미세한 오차가 발생했다.
다섯 개의 플레이어를 띄워놓고 재생했음에도 현재 시처자 수가 4~5를 왔다갔다 하는 현상이 발생하는 것이다.
그리고 작은 단위로 집계하다보니 DB 부하가 생겨 하필 휴가중에 ‘현재 시청자수가 안나와요!’하는 전화를 받기도 했다. 🥲..
그래서 이번에 로직을 변경했다. 바꾼 뒤로 현재까지는 아직 문제가 생기지 않은 상태다.

변경한 프로세스 로직

DynamoDB에는 Time To Live(TTL) 기능이 있다. 기능을 활성화 하면 지정한 타임스탬프가 지나면 48시간 이내에 해당 아이템을 삭제한다. 속성은 Epoch 시간 형식의 숫자 데이터 형식을 사용해야 한다. 이 기능을 활용해보기로 했다.

현재 시청자수를 위한 테이블을 생성했다.

1
2
3
4
5
6
7
8
const putCurrentPlaydata: DynamoDB.DocumentClient.PutItemInput = {
  TableName: process.env.LIVE_AGG_CURRENT_TABLE_NAME,
  Item: {
    PK: streamId,
    SK: uuid,
    expDate: expDateForSeconds,
  },
};

PK는 streamId, SK는 uuid로 설정했다. 이렇게하면 uuid별(재생 기록 단위)로 item을 구분할 수 있고 streamId(집계 미디어 단위)로 쿼리가 가능하다.
TTL은 expDate 속성을 보도록 설정했다.

1
2
3
4
5
6
7
8
9
10
11
12
export const queryCurrentViewParam = async (streamId: string) => {
    const current = Math.floor(moment().valueOf() / 1000);
    const queryParam = {
        TableName: process.env.LIVE_AGG_CURRENT_TABLE_NAME,
        KeyConditionExpression: "PK = :streamId",
        FilterExpression: "expDate >= :currentDate",
        ExpressionAttributeValues: {
            ":streamId": streamId,
            ":currentDate": current,
        },
    }
}

느낀 점

5초마다 들어오는 데이터 스트림을 어떻게 처리해야되나 정말 고민이 많았다. 중간에는socket.io를 써 볼까 싶어 했는데 단지 실시간 집계 때문에 소켓을 도입하는 건 과하다는 느낌을 받아서 도입하지 않았다. 데이터 스트림을 처리니까 AWS Kinesis를 사용해볼까도 생각해봤는데 Kinesis는 데이터를 사용하는 타겟 소스가 여러개인 경우에 사용하는게 적합한 서비스였다. 아예 큰 틀을 바꿔서 프로세스를 개선 시켜보려고도 했지만 결국에는 AWS SQS, DynamoDB Stream 조합이 가장 적합하다는 것을 다시 한 번 느끼고 새로운 개선 방안을 모색한 결과 생각보다 단순하게 문제를 해결할 수 있었다. 이것도 정답이 아닐지도 모르지만.. 문제가 생기면 또 해결하면 되니까 🙃