일단 써보기

Kafka Streams의 stateful 프로세싱은 다른 시간에 도착하는 관련 이벤트들을 그룹화하여 관리하고 저장할 수 있는 기능을 제공한다. State Store는 중간 상태를 로컬 또는 원격으로 저장하는 구조로, Kafka 체인지 로그 토픽을 기반으로 한 내결함성을 갖추고 있다. 이러한 구조 덕분에 Kafka Streams 애플리케이션은 인스턴스 간 데이터 일관성을 유지하면서도 효율적인 스케일 아웃을 달성할 수 있다. 이번 글에서는 Kafka Streams의 state store 분산 방식, 데이터 일관성 유지 방법, 그리고 로컬 및 원격 상태 조회 방식을 중점으로 다룬다.

1. State Store의 역할과 필요성

Kafka Streams에서 State Store는 스트림 처리 중 발생하는 중간 상태를 로컬에 저장한다. 예를 들어 사용자 행동 데이터를 처리할 때, 각 사용자의 총 구매 횟수를 집계한다고 가정해보자. 각 이벤트가 들어올 때마다 사용자의 기존 구매 횟수를 가져와 업데이트해야 하는데, 이 중간 데이터는 빠르게 접근할 수 있는 로컬 저장소에 저장하는 것이 효율적이다.

• 사용자의 구매 이벤트가 Kafka 토픽으로 수신될 때마다, Kafka Streams 애플리케이션은 구매 이벤트를 기반으로 사용자의 총 구매 횟수를 누적하여 업데이트한다.
• 각 Kafka Streams 인스턴스는 사용자 ID를 기준으로 파티셔닝된 데이터를 관리하며, 해당 파티션에 속한 사용자의 구매 횟수를 state store에 저장하여 빠르게 접근할 수 있다.

2. 스케일아웃 상황에서 State Store와 Changelog Topic의 동작 방식

Kafka Streams의 State Store와 Changelog Topic은 인스턴스 간 분산 처리와 데이터 일관성을 지원하기 위해 다음과 같은 방식으로 동작한다.

2-1. 파티셔닝과 할당

Kafka Streams는 데이터를 파티셔닝하여 관리하며, 각 인스턴스는 특정 파티션을 처리하는 방식으로 상태를 유지한다. 스케일아웃 시 인스턴스가 추가되면, 기존 인스턴스에 할당된 파티션을 새 인스턴스에 자동으로 재할당하여 데이터 처리가 더욱 분산된다. 각 인스턴스는 할당된 파티션에 대한 독립적인 State Store를 유지하게 된다.

2-2. State Store 데이터 복제와 일관성 유지

각 인스턴스의 State Store는 Changelog Topic에 변경 사항을 기록하여 다른 인스턴스에서도 해당 상태를 복구할 수 있도록 한다. 인스턴스가 확장되어 파티션이 새로운 인스턴스에 재배치될 때, 새로운 인스턴스는 Changelog Topic을 구독하여 필요한 데이터를 로드하고, 이후 업데이트 사항을 받아 데이터 일관성을 유지한다.

2-3. 스케일아웃 과정에서의 성능 최적화와 복구 지연 완화

스케일아웃 중 State Store의 데이터 크기가 클 경우 복구에 시간이 지연될 수 있다. Kafka Streams는 이를 완화하기 위해 상태 스냅샷(State Snapshot)과 상태 저장소 압축(State Store Compaction) 기법을 제공한다. 상태 스냅샷을 통해 특정 시점의 데이터를 저장하여 복구 시 전체 데이터를 다시 적용할 필요 없이 빠르게 복구할 수 있다. 압축 기법을 통해 불필요한 데이터를 정리하여 복구 시간을 단축할 수 있다.

2-4. 분산 환경에서의 데이터 일관성 보장

Kafka Streams는 분산 환경에서도 exactly-once 또는 at-least-once 처리 방식을 통해 데이터 일관성을 보장한다. State Store는 트랜잭션 방식으로 이벤트를 처리하며, 이벤트 처리 후 상태를 Changelog Topic에 커밋하여 장애 시 중복 처리를 방지한다. 이를 통해 스케일아웃 상황에서도 데이터 일관성을 유지할 수 있다.

예시 상황

쇼핑 애플리케이션에서 사용자 구매 이벤트를 처리하는 Kafka Streams 애플리케이션을 가정한다.

1. 초기 인스턴스 구성
   사용자의 구매 횟수를 집계하는 애플리케이션이 초기에는 두 개의 인스턴스를 사용한다. 각 인스턴스는 사용자 ID를 기준으로 파티셔닝된 데이터를 할당받아 각 사용자의 총 구매 횟수를 State Store에 기록한다.
2. 스케일아웃으로 인스턴스 추가
   트래픽이 증가하면서 새로운 인스턴스를 추가하여 스케일아웃한다. Kafka Streams는 기존 두 인스턴스에 할당된 파티션 중 일부를 새 인스턴스에 자동으로 재할당한다. 새 인스턴스는 할당된 파티션의 Changelog Topic을 구독하여 해당 파티션의 모든 상태 데이터를 로드하고, 이후 들어오는 구매 이벤트를 처리한다.

3. State Store 데이터 일관성 유지 및 장애 복구

Kafka Streams 애플리케이션은 상태가 필요한 스트림 처리 작업을 수행할 때 State Store를 사용한다. 이때 State Store에 저장된 데이터는 로컬에 유지되기 때문에, 특정 인스턴스에 장애가 발생하면 해당 인스턴스의 상태 정보를 잃어버릴 위험이 있다. Changelog Topic은 이러한 문제를 해결하기 위해 상태 변경 사항을 Kafka의 특별한 토픽에 기록해 두며, 다른 인스턴스가 해당 파티션을 맡게 되었을 때 이 기록을 바탕으로 State Store를 복구할 수 있게 해준다.

• 웹사이트에서 발생하는 사용자 클릭 이벤트를 집계하는 Kafka Streams 애플리케이션을 가정한다. 각 인스턴스는 특정 파티션의 데이터를 처리하며, 클릭 수 집계를 state store에 기록한다.
• 인스턴스 장애 시, 다른 인스턴스가 해당 파티션을 할당받아 state store를 복구하는데, 이때 필요한 데이터는 Changelog Topic에서 불러와 복구한다. 이를 통해 중단 없이 데이터 일관성을 유지하면서 집계를 이어갈 수 있다.

Changelog Topic은 Kafka의 파티션 구조를 활용해 각 State Store의 변경 사항을 토픽 파티션에 저장한다. Kafka Streams 애플리케이션이 여러 파티션을 사용하는 경우, 각 파티션별로 Changelog Topic도 같은 수의 파티션을 가지게 된다. 이를 통해 Changelog Topic의 데이터는 애플리케이션 파티션과 동일한 구조로 분산 저장되어 있어 복구할 때 효율적이다.

• 데이터 일관성 유지: State Store는 기본적으로 RocksDB 같은 로컬 데이터베이스에 저장된다. State Store가 변경될 때마다 로그 커밋이 Changelog Topic으로 전송되어 데이터 일관성이 보장된다. Kafka Streams의 at-least-once 또는 exactly-once 처리 보장에 따라, 중복 데이터가 발생할 수 있는 상황에서도 Changelog Topic을 통해 마지막으로 커밋된 상태를 기준으로 일관성을 유지할 수 있다.

Kafka Streams의 인스턴스가 장애로 인해 중단될 때, Kafka는 해당 인스턴스가 맡고 있던 파티션을 다른 인스턴스에 할당한다. 이 새로운 인스턴스는 Changelog Topic에 저장된 데이터를 기반으로 State Store를 복구한다.

• 복구 과정
  • 새 인스턴스가 할당된 파티션의 Changelog Topic을 구독하여 모든 변경 기록을 가져온다.
  • 각 이벤트를 순차적으로 적용해 State Store를 다시 생성한다.
• 예를 들어, A 인스턴스가 클릭 수를 관리하던 도중 장애가 발생했다면, B 인스턴스가 해당 파티션을 할당받고, Changelog Topic에서 A 인스턴스의 상태 변경 사항을 받아 State Store를 복구해 클릭 수 집계를 이어간다.

4. Changelog Topic의 장점과 고려 사항

장점:

• 내결함성(Fault Tolerance): Changelog Topic은 State Store의 상태를 Kafka 클러스터에 백업하는 역할을 하기 때문에, 로컬 데이터 손실이 발생해도 안전하게 복구할 수 있다.
• 데이터 일관성: Kafka Streams는 장애 시 재처리를 통해 일관성을 유지하기 위해 Changelog Topic을 사용하며, 정확히 한 번(exactly-once) 또는 적어도 한 번(at-least-once) 처리 옵션을 설정할 수 있다.

고려 사항:

• 디스크 사용량: Changelog Topic은 모든 State Store 변경 사항을 저장하므로, 대규모 데이터를 다루는 애플리케이션에서는 저장 용량이 커질 수 있다. 필요 시 TTL(Time-to-Live) 설정을 통해 오래된 데이터를 삭제하거나, 압축을 통해 용량을 줄일 수 있다.
• 복구 지연 시간: 대량의 데이터를 가진 State Store를 복구할 때 지연이 발생할 수 있다. 복구 시간이 중요한 경우, State Store를 주기적으로 백업하거나 상태 스냅샷 기능을 사용해 복구 시간을 단축하는 것이 도움이 된다.

결론 : Terraform으로 Public ECR을 정의할때는 provider의 region을 us-east-1으로 설정해야 한다.

resource "aws_ecrpublic_repository" "example_ecrpublic" {
  provider = aws.us-east-1
  repository_name = "example_ecrpublic"
}

K8s 클러스터를 구성하기 위해 AWS Container registry를 Terraform으로 정의하고 있었는데, 다음과 같은 오류가 발생했다.

no such host라기에 웹 콘솔에서는 정상적으로 생성되는지 확인해 보았는데 정상적으로 생성되었다.

관리자도구를 켜서 생성시 API요청이 어떻게 날아가는지 확인해보니, terraform을 통해 생성 요청 URI경로와, 웹 콘솔 생성 요청 URI를 의 region이 서로 달랐다.

Terraform : api.ecr-public.ap-northeast-2.amazonaws.com

AWS console : ecr-public.us-east-1.amazonaws.com/

private repository를사용할때는 ap-northeast-2로 정상적으로 리소스가 생성이 되었는데,

public repository를 사용하니 us-east-1로 region을 지정해야 리소스가 정상적으로 생성된다.

https://github.com/hashicorp/terraform/issues/5137

관련 내용을 terraform issue에서 확인할 수 있었는데, aws는 public ecr을 us-east-1에서만 지원하기 때문에, terraform에서는 provider의 region을 us-east-1로 지정해야 한다.

바이너리 로그의 포맷에 대해 이야기하기에 앞서서 바이너리 로그에 대해 정의해야 한다.

바이너리 로그(Binary Log)

바이너리 로그(binary log)는 MySQL에서 사용하는 기능 중 하나로, 데이터베이스에서 수행된 모든 데이터 변경 작업(INSERT, UPDATE, DELETE 등)을 기록하는 데 사용된다.

바이너리 로그에는 다음과 같은 중요정보들이 포함된다.

• SQL 쿼리문(또는 변화의 결과물 - ROW)
• 쿼리의 실행시각 및 쿼리 대상 데이터베이스에 대한 정보

그렇다면 왜 바이너리 로그를 사용할까?

• 복제(Replication): 바이너리 로그는 MySQL Replication의 핵심 요소이다. Master 서버의 바이너리 로그에 쓰인 내용이 Slave 서버에 반영되어 복제 프로세스가 이루어진다.
• 데이터 복구(Point-In-Time Recovery): 예기치 못한 데이터 손실이 발생했을 경우, Binary Log에 기록된 작업들을 재실행 함으로써 특정 시점으로 데이터베이스 상태를 되돌릴 수 있다.

바이너리 로그를 사용함으로써 데이터 복구와 복제라는 두 가지 중요한 일을 수행할 수 있지만, 디스크 공간을 소비하고 I/O 성능에 약간의 부하를 준다는 점을 유의해야한다.

바이너리 로그 포맷(Binary Log Format)

MySQL의 바이너리 로그(Binary Log)는 세 가지 주요 로깅 포맷을 제공한다.

Statement-based (--binlog-format=STATEMENT)

• 데이터 변경 작업을 수행하는 SQL 쿼리문을 바이너리 로그에 저장한다.
• Replication시 복제본 서버는 원본 서버의 바이너리 로그를 읽어들여 그 안에 기록된 SQL 문장들을 순서대로 재실행하여 Replication을 수행한다.
• SQL 쿼리문을 바이너리 로그에 저장하므로 감사에 용이하다.
• 트랜잭션 격리수준이 Repeatable Read 이상이어야 한다.

Row-based (--binlog-format=ROW (기본값))

• 바이너리 로그에서 개별 테이블 행이 어떻게 영향을 받는지를 로그에 기록한다.
• Replication시 SQL 문장을 해석하거나 재실행하는 것이 아니라, 단순히 특정 레코드의 변경 사항을 적용하여 Replication을 수행한다.

바이너리 로그의 내용은 일반 텍스트 형태로 기록되지 않고 바이너리 형식으로 보관되기 때문에, 직접적으로 내용을 읽을 수는 없지만, mysqlbinlog 도구를 사용하면 바이너리 로그 파일의 내용을 텍스트 형태로 변환하여 확인할 수 있다.

Statement-based

# at 141
#191024 15:58:08 server id 1  end_log_pos 236   Execute    load data local infile '/tmp/sbtest1.txt' REPLACE into table sbtest1 fields terminated by '\t'​

위와 같이, Statement-based 로깅의 경우 실행된 SQL 쿼리 문장(load data local infile '/tmp/sbtest1.txt' REPLACE into table sbtest1 fields terminated by '\t')이 직접 로그에 기록된다.

Row-based

### UPDATE `test`.`sbtest1`
### WHERE
###   @1=7292
###   @2='33381779362-68792378251-25015017183-69472446453-71422697075'
### SET
###   @1=7292
###   @2='33381779362-68792378251-97250707473-96610169832-15182108387'

Row-based 로깅의 경우 SQL 쿼리의 변화된 결과가 바이너리 로그에 기록된다.

위 예시에서는 test.sbtest1 테이블의 한 레코드가 UPDATE 되었음을 표시하고 있다. WHERE 절 이후에 원래 값(@1=7292, @2='33381779362-68792378251-25015017183-69472446453-71422697075')이 나타나며, SET 절 이후에는 변화된 값(@1=7292, @2='33381779362-68792378251-97250707473-96610169832-15182108387')이 나타난다.

이와 같이 바이너리 로그의 형태는 설정된 모드에 따라 크게 달라진다.

Mixed (--binlog-format=MIXED)

• 기본적으로 Statement-based 로깅을 사용하지만, Statement-based로 안전하게 복제되지 않을 수 있을 때에는 자동으로 Row-based 방식으로 전환하여 로깅을 진행한다.
• 안전하게 복제되지 않을수 있을 때란 비확정적 쿼리(non-deterministic query)가 발생했을 때가 대표적이다. 비확정적 쿼리는 항상 같은 결과를 반환하지 않는 쿼리를 말하는데, 예를 들어 UUID()나 RAND()와 같은 함수를 사용하는 쿼리, 혹은 데이터베이스의 특정 상태(예: AUTO_INCREMENT 값, 시스템 변수 등)에 의존하는 쿼리 등이 해당한다.

MySQL의 Mixed 방식에서의 로깅 방식 결정

다음의 경우 Statement-based 바이너리 로깅이 아닌 Row-based 바이너리 로깅을 사용한다.

• 함수에 UUID()가 포함되어 있을 때.
• AUTO_INCREMENT 열이 있는 하나 이상의 테이블이 업데이트되고 트리거나 저장된 함수가 호출될 때.
• 뷰의 본문이 row-based 복제를 필요로 할 때, 뷰를 생성하는 문장도 그것을 사용합니다. 예를 들어, 뷰를 생성하는 문장이 UUID() 함수를 사용할 때 이런 경우가 발생합니다.
• 로드 가능한 함수 호출이 포함되어 있을 때.
• FOUND_ROWS() 또는 ROW_COUNT()가 사용될 때.
• USER(), CURRENT_USER(), 또는 CURRENT_USER가 사용될 때. (참조: Bug #28086)
• 관련된 테이블 중 하나가 mysql 데이터베이스의 로그 테이블일 때.
• LOAD_FILE() 함수가 사용될 때.
• 명령문이 하나 이상의 시스템 변수를 참조할 때. 

https://dev.mysql.com/doc/refman/8.0/en/binary-log-mixed.html

비확정적 쿼리가 문제가 되는경우 row-based나, mixed-based를 사용하는것이 좋다는것은 알겠다. 그럼 mixed-based를 사용하면 되지 왜 row-based를 사용해야 할까?

핵심은 STATEMENT 포맷은 SQL 쿼리 문장 그대로를 기록하는 반면, ROW 포맷은 쿼리의 결과(변경된 레코드)를 기록한다는 점에 있다. 

복제(Replication)

앞서 바이너리 로그는 MySQL Replication의 핵심요소라고 설명했다. MySQL에서 복제는 어떻게 수행되길래 바이너리 로그를 필요로 하는걸까?

데이터베이스 복제작업은 다음과 같은 과정을 거친다.

1. 소스 MySQL서버의 전체 데이터 스냅샷을 레플리카 서버에 이관한다. (mysqldump와 같은 툴 활용)
2. 스냅샷이 생성되는 동안 소스 서버에서 발생하는 모든 데이터 변경 작업은 바이너리 로그에 기록된다.
3. 스냅샷 이관이 완료되면 레플리카 서버는 소스 서버의 바이너리 로그를 읽어 릴레이 로그에 저장하고, 릴레이로그에서 읽어 DB에 반영한다.

복제과정에서의 바이너리 로그 포맷

Statement-based 바이너리 로그 포맷의 경우 복제과정에서 소스서버에서 수행된 쿼리를 그대로 실행하게 된다. 때문에 소스서버에서 불필요하게 많은 락을 유발한 쿼리들이 그대로 동일하게 수행될 수 있다.

반면 Row-based 바이너리 로그포맷은 변경된 데이터 자체가 전달되어 복제가 수행되기 때문에 소스서버에서 발생한 락을 재생산하지 않는다.