안녕하세요, 데이터분석플랫폼 조직의 루이스입니다.

시리즈의 첫 글인 아파치 플링크와 CDC의 만남. 플링크 CDC 맛보기에 이어 두 번째 글을 쓰게 되었습니다. 첫 번째 글에서는 아파치 플링크로 CDC(Change Data Capture)를 수행하여 MySQL 테이블을 다른 MySQL 테이블로 연동하는 내용을 공유드렸었는데요, 이번 글에서는 플링크를 사용하여 MySQL 테이블을 아파치 아이스버그(Apache Iceberg)로 CDC를 수행하고 운영하면서 얻은 경험들을 독자분들과 공유하고자 합니다.

먼저 저희 팀의 미션을 간략하게 말씀드리면, 서비스 팀들의 데이터를 취합하여 일단위의 지표를 추출하고 제공하는 것이라고 할 수 있습니다. 이를 위해 다양한 소스에서 데이터를 수집하고 있으며, 그 과정에서 서비스 팀의 데이터베이스에 있는 데이터를 활용해야 하는 경우도 있습니다. 그러나 서비스 팀의 데이터베이스는 실제 서비스에 사용되고 있기에, 지표 추출을 목적으로 서비스 팀의 데이터베이스에 접근하여 데이터를 가져오는 방식은 실서비스의 운영에 부담을 줄 수 있습니다. 따라서, 실서비스 데이터베이스가 아닌 별도의 데이터베이스에서 지표 추출을 목적으로 데이터를 가져와야 합니다.

이를 위해 서비스 팀의 데이터베이스가 저희 팀 내 데이터베이스에 실시간 연동이 될 필요가 있으며, 이런 실시간 연동 작업을 CDC라고 칭합니다. 다만 첫 번째 글에서 소개했던 것처럼, 타겟 시스템으로 MySQL 데이터베이스를 사용하면, 지표 추출을 위해 데이터베이스에서 데이터를 가져올 때 데이터베이스에 부하가 있어 성능 상의 제약이 발생하는 일부 비효율적인 과정이 존재합니다. 저희 팀에서는 이 문제를 해결하기 데이터 레이크하우스 기술 중 하나인 아파치 아이스버그(Apache Iceberg)를 도입하였습니다.

본 글은 아파치 아이스버그에 대한 소개로 시작하여, 플링크에서 아이스버그로 데이터 적재 시 필요한 준비 과정을 먼저 소개합니다. 이후 플링크의 데이터스트림 API를 사용하여 아이스버그로 데이터가 적재되는 과정, 적재 과정에서 생성되는 파일들의 정보 및 활용 방법을 실제 예시와 함께 설명합니다. 또한 글의 마무리에서는 샤딩 테이블들을 하나의 아이스버그 테이블로 운영하는 가능성을 시험하기 위한 구현 및 테스트 결과를 공유드리며 글을 마무리하고자 합니다.

다만 본 글에서는 CDC와 아파치 플링크에 대한 기본적인 내용은 다루지 않습니다. 관련한 정보는 첫 번째 블로그 글인 아파치 플링크와 CDC의 만남. 플링크 CDC 맛보기에 잘 정리되어 있습니다. 해당 배경 지식이 없으신 경우 본 글을 이해하기 어려울 수 있으니, 첫 번째 블로그 글을 먼저 읽으시는 것을 권장드립니다. 만약 CDC, 아파치 플링크 및 아이스버그에 대해 어느 정도 이해하고 계신다면, 본 글에서 아이스버그에 대한 개념을 설명한 이후의 파트인 플링크에서 아이스버그로 준비 과정부터 읽으셔도 내용을 이해하시는 데 무리가 없을 것입니다.

아울러 본 글에서 언급되는 주요 시스템과 라이브러리의 버전은 아래와 같으며, 참고해 주시기 바랍니다.

아파치 플링크: v1.17.1
아파치 아이스버그: v1.5.0
아파치 하이브: v2.3.2
플링크 CDC 라이브러리: v2.4.1

아파치 아이스버그(Apache Iceberg)

아파치 아이스버그는 넷플릭스(Netflix)가 개발한 오픈 테이블 포맷을 지원하는 데이터레이크 기술입니다. 아이스버그는 테이블의 형태로 데이터가 추상화되어, 사용자가 일반적인 관계형 데이터베이스의 테이블에 CRUD 쿼리를 수행하는 것과 동일한 방식으로 데이터 조회 및 수정이 가능한 것이 특징입니다. 또한 트랜잭션을 지원하여 원자성, 일관성, 격리성 및 지속성이라 불리는 ACID(Atomicity, Consistency, Isolation, Durability)가 보장되며 다수의 사용자에게 일관성 있는 데이터 뷰(View)를 제공합니다. 아울러 아이스버그는 모든 정보를 파일로 저장하며, 파일을 저장하기 위한 객체 저장소(Object storage)로 S3와 하둡 파일 시스템(Hadoop File System)등을 지원합니다.

아이스버그의 주요 기능에는 변화분에 대한 증분 업데이트 기능이 있습니다. 아이스버그 테이블은 앞서 설명드린 것처럼 일반적인 관계형 데이터베이스와 동일하게 CRUD를 지원하기 때문에, 다른 관계형 데이터베이스의 변경분을 아이스버그에 반영하는 게 가능합니다. 아이스버그는 이러한 증분 업데이트 기능 덕분에 CDC 연동이 가능하여, 데이터베이스를 CDC로 연동해 대규모 데이터 분석을 위한 적재용 타겟 시스템으로 많이 선택됩니다.

반대로 아이스버그가 제공하는 관계형 데이터베이스와 차별화되는 기능으로는 과거 시점의 테이블 상태를 조회하는 타임 트레블(Time Travel) 기능이 있습니다. 아이스버그는 주기적으로 테이블에 커밋(commit)이라는 작업을 수행합니다. 이 커밋 작업은 아이스버그 테이블에 일정 시간 동안 인입된 데이터들을 반영하여 새로운 상태를 나타내는 스냅샷을 생성합니다. 즉, 커밋을 통해 새로운 스냅샷이 생성되고 스냅샷은 특정 시점의 테이블 상태를 나타냅니다. 아이스버그는 이렇게 생성된 스냅샷들의 이력을 파일에 계속 유지하기 때문에, 이러한 스냅샷을 기반으로 과거 시점의 테이블 상태를 조회하는 타임 트레블 기능을 제공할 수 있습니다.

마지막으로 소개드릴 아이스버그의 주요 기능은 히든 파티셔닝(Hidden Partitioning)입니다. 아파치 하이브(Apache Hive)와 아이스버그 간 자주 비교되기도 하는 기능인데요, 하이브의 경우 수행할 쿼리를 작성할 시 파티션 칼럼을 쿼리의 조건 구문에 항상 명시해야 하는 특징이 있습니다. 물론 하이브의 설정을 변경하여 파티션 칼럼이 없이도 쿼리를 수행할 수 있기는 하지만, 이 경우 성능 하락을 감수해야 합니다. 하지만 아이스버그의 경우, 쿼리 작성 시 사용자가 파티션을 따로 명시하지 않아도 됩니다. 아이스버그 테이블에 파티션이 적절히 설정되어 있다면, 메타데이터에 저장된 파티션 정보를 참고하여 자동으로 최적화된 데이터 접근을 제공하기 때문입니다.

아이스버그 도입 필요성

기존에 저희 팀은 플링크를 통해 MySQL 데이터베이스의 테이블을 CDC 연동하고, 연동된 테이블의 전체 데이터를 일 단위 배치(Daily batch)로 하둡 파일 시스템에 적재하고 있었습니다. 이후 적재된 데이터를 스파크로 가져온 후, 데이터 변환 및 지표 계산을 마지막으로 수행했습니다. 하지만 이 과정에서도 두 가지 아쉬운 점이 있었습니다.

첫 번째는 비효율적인 작업이 매일 반복된다는 점입니다. MySQL 데이터베이스에서 일 배치로 테이블 전체 데이터를 가져오는 작업은 테이블의 변화분이 많거나 적은 것과 무관하게 동일하게 수행됩니다. 따라서 변화분이 많고 적은 상태와 상관없이 테이블 전체 데이터를 매일 가져오는 작업은 비효율적일 수밖에 없었습니다.

두 번째는 MySQL 데이터베이스에 발생하는 부하가 스파크(Spark)의 성능을 제약한다는 점입니다. 일례로 MySQL 데이터베이스에서 테이블의 전체 데이터를 빠르게 가져오기 위해 스파크 애플리케이션(Spark Application)에 더 많은 자원을 할당하게 되면, 데이터베이스 서버의 디스크 사용률이 순식간에 100%에 도달하는 것을 경험할 수 있습니다. 따라서, 스파크 애플리케이션에 더 많은 자원을 할당하는 것이 불가능하기 때문에, 테이블을 소싱(Sourcing) 하기 위해 병렬로 동작하는 각각의 스파크 애플리케이션들 또한 MySQL 부하를 고려하여 성능에 제한을 두고 있습니다.

이 상황에서 아이스버그를 도입하면 두 가지 장점이 있습니다. 첫째, 테이블의 전체 데이터를 하둡 파일 시스템에 일 배치로 일괄 적재하는 단계가 필요 없어집니다. CDC를 통해 하둡 파일 시스템에 존재하는 아이스버그 테이블에 실시간으로 변화분을 반영할 수 있기 때문입니다. 둘째, 기대하는 성능만큼 스파크 애플리케이션에 충분한 자원을 할당하는 것이 가능해집니다. 기존에는 MySQL 데이터베이스에서 테이블 전체 데이터를 가져올 시, 발생하는 부하 때문에 관련 조직과 협의된 수준에서만 자원을 할당할 수 있었습니다. 하지만 하둡 파일 시스템에 적재된 아이스버그 테이블에는 부하가 없어 기대하는 성능만큼 자원 할당이 가능한 장점이 있습니다.

아이스버그 카탈로그 계층

그림 1. 카탈로그 계층

처음 소개할 아이스버그의 주요 구성 요소는 카탈로그(Catalog)입니다. 카탈로그는 네임스페이스(namespace)로 그룹화된 테이블들을 관리하며, 테이블 생성, 제거 및 변경과 같은 모든 종류의 작업을 처리합니다. 이를 위해 카탈로그는 테이블의 현재 메타데이터에 대한 포인터(Current Metadata Pointer)를 가지고 있어, 사용자가 테이블에 작업을 수행할 시 테이블의 최신 메타데이터를 알려 주는 일종의 진입점 역할을 합니다.

또한, 카탈로그는 테이블에 수행 중인 트랜잭션들의 상태를 확인할 수 있습니다. 이를 통해 테이블에 대한 일관성 있는 뷰(View)를 유지할 수 있기 때문에, ACID의 무결성과 안정성에 필수적인 요소입니다. 단, 트랜잭션 확인은 동일한 유형의 카탈로그 내에서만 가능합니다. 만약 하나의 아이스버그 테이블에 여러 유형의 카탈로그를 사용하면, 서로 다른 유형의 카탈로그간에는 수행 중인 트랜잭션 상태 확인이 불가하여 일관성 있는 뷰를 보장할 수 없습니다. 즉, 한 유형의 카탈로그를 통해 테이블을 커밋하면, 다른 유형의 카탈로그는 테이블의 과거 상태를 바라볼 수도 있다는 것을 의미합니다.

아이스버그와 호환되는 카탈로그 유형은 크게 서비스 카탈로그(service catalog)와 파일 시스템 카탈로그(file-system catalog)로 분류됩니다.

먼저 서비스 카탈로그는 온-프레미스(On-premise) 또는 클라우드 관리형 서비스(예: AWS)를 의미합니다. 서비스 카탈로그는 백업 저장소를 사용하여 아이스버그 테이블에 대한 모든 참조를 유지하며, 락(Lock) 메커니즘을 통해 ACID를 보장합니다. 대표적인 카탈로그에는 깃(Git)과 유사하게 버전 관리 기능에 목적을 두는 네시(Nessie)가 있습니다. 또 다른 카탈로그에는 하이브 메타스토어(Hive Metastore)가 있으며, 이미 하이브 환경에 익숙한 조직은 이를 쉽게 활용이 가능하다는 이점이 있습니다. 상기의 이유로 저희 팀은 현재 하이브 메타스토어를 사용 중입니다. 이 외에도 아마존 웹 서비스의 글루(Glue), 스노우플레이크(Snowflake) 및 JDBC 등도 카탈로그를 사용할 수 있습니다.

한편, 대표적인 파일 시스템 카탈로그로는 하둡(Hadoop) 카탈로그가 있습니다. 하둡 카탈로그는 파일 시스템에 version-hint.txt 파일을 사용하여 테이블의 최신 버전을 추적합니다. 서비스 카탈로그는 네시가 버전 관리 기능에 특화된 기능을 제공되는 것처럼, 각각의 목적에 맞추어 특화되고 편리한 기능(예: 동시성 제어)들을 제공합니다. 하지만 파일 시스템 카탈로그는 단순한 저장소로서의 기능만 제공하기에, 실서비스 환경에서는 권장되지 않는 편입니다.

아이스버그 메타데이터, 데이터 계층

다음으로 설명드릴 요소는 아이스버그의 메타데이터 계층(metadata layer)에 존재하는 메타데이터 파일(metadata file), 매니페스트 리스트(manifest list) 그리고 매니페스트 파일(manifest file)입니다. 메타데이터 계층에는 실제 데이터를 제외한 모든 정보들이 존재하며 아이스버그의 주요 기능들에 핵심이 되는 부분입니다.

그림 2. 메타데이터 계층의 메타데이터 파일

메타데이터 계층의 구성 요소들 중 제일 먼저 설명드릴 요소는 메타데이터 파일입니다. 앞서 카탈로그는 현재 메타데이터 포인터를 가지고 있어, 최신의 메타데이터 파일을 가리키고 있다고 설명드렸습니다. 메타데이터 파일은 아이스버그 테이블에 커밋이 성공할 때마다 새로 생성되며, 현재의 메타데이터 포인터도 새로 생성된 메타데이터 파일을 가리킵니다. 이때, 커밋은 원자적(Atomic)으로 수행되어 동시성이 있는 환경에서도 유실이 발생하지 않습니다. 이 특성을 활용하여 새로운 메타데이터 파일이 이전 버전의 메타데이터 파일을 기반으로 생성되고 교체되도록 보장할 수 있습니다.

메타데이터 파일에는 테이블에 대한 기본적인 정보들과 추적 중인 스냅샷들에 대한 정보가 존재합니다. 기본적인 정보에는 테이블의 유니크 아이디, 테이블에 적용된 설정, 스키마 정보 및 관련 파일들의 저장 경로 등이 포함됩니다. 또한 생성된 스냅샷들의 상대 적인 나이를 나타내는 시퀀스 넘버(Sequence Number)가 있습니다. 이외에도 레코드 수나 파일 개수와 같은 통계 정보 그리고 연관된 매니페스트 리스트의 저장 경로 등이 존재합니다. 아이스버그는 이런 정보들을 활용하여 사용자가 특정 시점의 테이블을 조회할 경우, 해당 스냅샷의 매니페스트 리스트를 확인하고 필요한 파일들을 읽어 테이블 형태로 제공합니다.

그림 3. 메타데이터 계층의 매니페스트 리스트, 파일

다음으로 메타데이터 계층에서 설명드릴 요소는 매니페스트 리스트입니다. 매니페스트 리스트는 특정 스냅샷에 대한 정보를 담고 있는 파일입니다. 이를 스냅샷과 비교하여 설명드리면, 스냅샷은 특정 시점의 테이블 상태를 의미하고 매니페스트 리스트는 스냅샷에 대응되는 실제 정보들을 가지고 있는 물리적인 파일입니다. 매니페스트 리스트는 커밋 후 생성된 모든 매니페스트 파일들에 대한 정보를 리스트 형태로 가지고 있으며, 매니페스트 파일 관련 정보에는 매니페스트 파일의 타입, 추가 또는 삭제된 레코드 수와 같은 통계 및 파티션 정보 등이 존재합니다.

메타데이터 계층에서 마지막으로 설명드릴 요소는 매니페스트 파일로, 이는 데이터 파일들과 삭제 파일들에 매칭되는 파일입니다. 매니페스트 파일은 매칭되는 파일의 타입 정보를 가지고 있으며 파일 타입에는 데이터 파일(data file), 동등 칼럼 기준의 삭제 정보를 가지는 동등 삭제 파일(equality delete file), 파일 경로 및 위치 기반의 삭제 정보를 가지는 포지션 삭제 파일(position delete file)이 존재합니다. 매니페스트 리스트와 마찬가지로 각 매니페스트 파일의 경로나 칼럼 별 최소 및 최대 값, null 값 개수 등의 통계 정보가 존재하며, 이 정보들은 이후 테이블 조회 시 필요한 특정 매니페스트 파일만을 확인하는 데 사용됩니다.

데이터 계층에는 실제 데이터들과 변화분들이 파일로 저장됩니다. 데이터 파일에는 삭제 메시지를 제외한 모든 데이터들이 존재하며, 삭제 파일의 경우 동등 삭제 파일은 동등 칼럼의 값이, 포지션 삭제 파일에는 파일 경로 및 포지션 정보가 저장됩니다. 아래 예시 1이 데이터 파일과 두 종류의 삭제 파일에 저장된 정보를 출력했을 때의 예시입니다. 이때 특이 사항으로 동등 삭제 파일에는 데이터 파일에 있는 모든 동등 칼럼(id)의 값이 들어있는 점을 꼽을 수 있는데요, 후술 될 쓰기 연산자 파트와 스캔 플래닝 파트에서 특이 사항에 대한 이유를 설명드리겠습니다.

// 데이터 파일
   id first_name last_name                email  phone_number job_title    salary  department_id  is_active
0  23   seungmin       lee seungmin@example.com 010-4321-9876  Employee  30000.00              1          1
1  24    Unknown   Unknown                 null          None      None      0.00             -1          0
2  25    gildong      hong  gildong@example.com 010-1234-5678  Employee  50000.00              1          1


// 동등 삭제 파일
   id
0  23
1  24
2  25


// 포지션 삭제 파일
                                                                                                                           file_path   pos
0   hdfs://hadoop-cluster/.../.../namespace/source_table/data/id_bucket=0/00000-0-6215b83a-023c-409e-8c6d-d9460bdf23a7-00009.parquet     1

<예시 1. 데이터 파일 및 두 종류의 삭제 파일>

아이스버그 테이블 주요 설정

아이스버그 테이블에는 다양한 설정이 존재합니다. 이번 파트에서 모든 설정을 다룰 순 없으나 테스트 및 조사 과정에서 파악한 내용, 그리고 현재 팀에서 사용 중인 것들 중 중요하다 판단되는 2가지 설정을 설명드리겠습니다. 이외에 실제로 사용 중인 다른 설정과 값들은 본 글의 플링크에서 아이스버그까지 준비 과정 > 테이블 설정 파트에서 자세히 공유드리겠습니다. 주요 설정 2가지를 설명드리기 전 저희 팀의 아이스버그 테이블의 사용 과정을 다시 한번 설명드리면 아래와 같습니다.

플링크를 통해 아이스버그 테이블을 생성
MySQL 테이블 데이터를 아이스버그 테이블로 적재
스파크를 통해 아이스버그 테이블을 조회

이제 본격적으로 아이스버그 테이블의 주요 설정을 설명드리겠습니다.

첫 번째는 설명드릴 설정은 테이블의 쓰기 모드 설정입니다. 쓰기 모드는 write.update.mode, write.delete.mode, write.merge.mode 설정에 개별적으로 설정 가능하며, 설정 가능한 값으로 COW(Copy-on-Write), MOR(Merge-on-Read)가 있습니다. 이 설정에 따라 데이터 적재 및 조회 방식이 달라집니다. 예를 들어 COW 모드로 설정되면 데이터 적재 시 기존 파일을 갱신하는 방식으로 동작하는데, 적재 시점에 데이터 변경사항이 기존 파일에 반영됩니다. 따라서 데이터를 기록하는 데 많은 비용이 들며 테이블 조회에는 이미 변화분 반영이 완료된 파일들을 읽기에 상대적으로 비용이 낮습니다. MOR로 설정하는 경우 실제 데이터와 변화분(삭제)이 개별 파일에 저장됩니다. 해당 파일들은 테이블 조회 시점에 합쳐져 보이기 때문에, 데이터를 기록하는 시점의 비용은 낮으나 테이블 조회 시 데이터를 합쳐야 해서 더 많은 비용이 듭니다. 즉, 어느 시점에 더 많은 비용을 지불할 것인지를 결정할 수 있도록 아이스버그 테이블 사용 환경을 명확히 한 후 적절한 설정을 선택해야 합니다.

두 번째는 파티션입니다. 아이스버그는 특정 칼럼에 파티션을 설정할 수 있으며, 파티션이 설정되면 파티션 별로 물리적으로 구분된 경로에 데이터가 저장됩니다. 지원되는 파티션의 종류로는 아래와 같으며 상황과 목적에 맞추어 적절한 방식을 선택하는 것을 권장합니다.

Bucket: 파티션 칼럼의 값을 해싱하고 사용자가 설정한 모듈로 값 기준으로 모듈러 연산을 수행하여 파티션을 구분합니다. 현재 팀에서 사용 중인 파티션 방식입니다.
Identity: 파티션 칼럼에 존재하는 고유 값들을 각각의 파티션으로 구분합니다.
Truncate: 파티션 칼럼의 값을 사용자가 넘겨준 정수 길이만큼 자른 후, identity 파티션을 수행합니다.
Hour, Day, Month, Year: 파티션 칼럼의 시간 정보 값을 기준으로 파티션을 구분합니다.

파티션 관련 두 가지 사항을 짚고 넘어가면, 첫번째 사항으로 파티션은 단순히 데이터를 물리적으로 구분하여 저장하는 것 이상으로 아이스버그의 성능에 중요한 영향을 끼칩니다. 특히 MOR 모드로 데이터가 적재되어 테이블 조회 시점에 변화분들을 반영해야 한다면, 파티션을 설정하는 것이 주요한 성능 개선 포인트가 될 수 있습니다. 파티션을 설정하게 되면, 테이블 조회 시 각 파일들을 비교 및 합치는 작업이 파티션 별로 수행되어 조회 성능을 향상할 수 있습니다. 뿐만 아니라, 이후 설명될 아이스버그 주요 최적화 기능인 컴팩션(Compaction)도 파티션 별로 수행되기에 더욱더 빠르게 컴팩션을 실행할 수 있습니다. 관련해서는 후술 될 아이스버그 테이블 조회 방식 및 최적화 파트에서 자세히 설명드리겠습니다. 다만, 너무 세분화된 파티션은 너무 많은 수의 작은 크기의 파일을 만들기 때문에 성능에 부정적인 영향을 줄 수 있으니, 적절히 파티션을 설정하는 게 중요합니다.

두번째 사항은 아이스버그가 동일 칼럼에 여러 파티션을 허용하지 않는다는 점입니다. 일부 사용자들이 시간 관련 칼럼에 여러 레벨(예: /ts_column_year=.../ts_column_month=.../)로 파티션을 설정하는 경우가 있는데, 이 경우 관리가 더 복잡해지고 앞서 말씀드린 파티션 세분화에 따른 성능 저하가 발생할 수 있습니다. 하지만 다행히도 아이스버그 개발팀에서 이를 인지하고 막아 놓았는데요(Improve partition spec builder · apache/iceberg · GitHub), 다만 라이브러리에서 제공하는 API가 아닌, 스파크 SQL 등의 방식으로 ALTER TABLE 쿼리를 수행하여 설정이 가능하지만, 설정 이후 테이블 조회 시 에러가 발생할 수 있기에 동일 칼럼에 여러 파티션을 설정하지 않는 것을 권한다고 합니다.

플링크에서 아이스버그로 준비 과정

저희 팀에서는 플링크를 통해 아이스버그 테이블에 데이터를 적재합니다. 따라서 아이스버그 외에도 플링크 및 연관된 일부 시스템들을 추가로 설정해야 했었습니다.

이번 파트에서는 저희 팀에서 플링크를 통해 MySQL 테이블의 데이터를 하둡 파일 시스템의 아이스버그 테이블로 적재하는 과정에서 적용한 세부 설정들 일부와, 간략한 코드들을 공유드리겠습니다. 단, 공유드리는 설정이 반드시 정답은 아니며, 상황에 따라 더 적합한 설정이 존재할 수 있기에, 아래와 같이 저희 팀의 미션과 상황을 먼저 명확히 말씀드리고 설정에 대한 설명을 이어서 기술하겠습니다.

팀의 미션은 지표 계산에 필요한 데이터 수집입니다. 지표 계산에 문제가 없는 선에서 MySQL 테이블 칼럼 타입과 아이스버그 테이블 칼럼 타입을 맵핑합니다.
하둡 파일 시스템을 저장소로 하이브 메타스토어를 카탈로그로 사용합니다. 지표 계산을 위해 사내 하둡 및 하이브를 이미 활용하고 있어 이미 친숙한 환경이었기 때문입니다.
재처리 시 멱등성을 보장하기 위해 UPSERT 모드로 데이터를 적재합니다.
하나의 플링크 잡은 오직 하나의 MySQL 테이블만을 CDC 연동합니다. 플링크 CDC 라이브러리는 하나의 플링크 잡에서 여러 테이블의 데이터와 변화분을 읽어오는 기능을 제공하나, 아이스버그의 플링크 적재 API는 여러 테이블에 데이터를 적재하는 것을 지원하지 않습니다.

플링크 설정

먼저 플링크에 설정한 일부 설정과 값들을 공유드리겠습니다. 첫 번째는 하둡 인증 관련 설정이며 두 번째는 플링크 체크포인트 주기 설정입니다.

하둡 인증 관련 설정부터 설명드리면, 팀에서는 하둡 파일 시스템을 객체 저장소로 사용하기에 하둡 인증 관련 설정을 플링크에 추가해야 합니다. 사내에서는 커버러스(Kerberos)를 통한 인증이 지원되기에, 플링크의 flink-conf.yaml 파일에 아래와 같이 설정을 합니다. 만약 플링크 버전 1.19 이상을 사용한다면 flink-conf.yaml 파일이 conf.yaml로 대체되었으므로, conf.yaml 파일에 동일하게 아래의 설정을 적용합니다.

hadoop.security.authentication: kerberos  
security.kerberos.login.principal: seungmin-lee@HADOOP  
security.kerberos.login.keytab: /../../seungmin-lee.keytab  
security.kerberos.access.hadoopFileSystems: hdfs://hadoop-cluster

<설정 1. 플링크 하둡 커버러스 설정>

다음으로는 플링크의 체크포인트 주기입니다. 체크포인트는 플링크가 주기적으로 저장하는 상태 정보로, 플링크 잡(Job)이 중단될 경우 잡 복구에 사용되는 정보입니다. 일반적으로 안정성 측면에서 체크포인트 주기는 작게 설정하는 것이 좋습니다. 하지만 아이스버그로 적재하는 과정에서 고려해야 할 사항이 있습니다. 플링크에서 아이스버그 테이블로 데이터 적재 시, 플링크 체크포인트 주기에 맞춰 아이스버그 테이블에 커밋이 수행되며, 새로운 파일들이 생성된다는 점입니다. 즉, 체크포인트가 빈번하게 생성될수록 더 많은 파일들이 생성되고, 개별 파일들의 크기는 작아지게 됩니다. 이처럼 파일들이 작게 그리고 많이 생성되는 상황은 테이블 조회 시간이 늘어나는 주된 원인으로 작용하는데요, 조회 시간이 늘어나는 이유는 아이스버그 테이블 조회 방식 및 최적화 > 스캔 플래닝 파트에서 더욱 자세히 설명드리겠습니다.

그렇다고 체크포인트 주기를 무작정 길게 늘이는 것 또한 말씀드린 것처럼 안정성 측면에 좋지 않습니다. 아울러 아이스버그 테이블의 조회 관점에서, 테이블에 적재된 데이터는 커밋이 수행되어야 조회가 가능해지기 때문에 체크포인트를 주기를 늘리는 것은 실시간성이 떨어지는 문제도 있습니다. 예를 들어 15시 30분에 데이터가 적재되고 커밋이 15시 50분에 수행된다면, 15시 30분에 적재된 데이터는 15시 50분 이후에 조회할 수 있습니다. 저희 팀은 테스트를 통해 안정적으로 체크포인트를 기록할 수 있으며, 지표 계산에도 문제가 없다 판단되는 10분을 최종적인 체크포인트 주기로 설정하였습니다.

하이브 설정

하이브에서 고려해야 할 설정은 DDL(Data Definition Language)과 관련 있는 hive.metastore.disallow.incompatible.col.type.changes의 설정값 변경입니다. 이 설정은 하이브 서버에서 전역적으로 설정되며, 칼럼 타입이 바뀌는 DDL의 허용 여부를 결정합니다. 이 설정을 더 설명드리기 전 DDL 관련하여 먼저 설명드려야 할 부분이 있는데, 기본적으로 플링크는 MySQL 테이블을 아이스버그 테이블로 CDC 연동 시, DDL 이벤트를 지원하지 않습니다. 아래와 같이 MySQL의 데이터와 변환분을 이벤트 타입에 맞춰 변경하는 코드 1의 로직은 항상 before 또는 after 키를 검증합니다. 하지만 DDL 이벤트 관련 메시지에는 해당 키들이 존재하지 않으며, 대신 수행된 DDL 정보가 들어있습니다. 따라서 앞선 1편의 아파치 플링크와 CDC의 만남. 플링크 CDC 맛보기에서 설명드린 것 처럼, DDL 이벤트가 발생하면 DDL과 이후 이벤트들을 스킵하고 아이스버그 테이블에 스파크 SQL을 통한 DDL을 별도로 수행합니다.


public void deserialize(SourceRecord record, Collector out) throws Exception {
    Envelope.Operation op = Envelope.operationFor(record);
    Struct value = (Struct) record.value();
    Schema valueSchema = record.valueSchema();
    // CREATE, SELECT 이벤트 처리
    if (op == Envelope.Operation.CREATE || op == Envelope.Operation.READ) {
        GenericRowData insert = extractAfterRow(value, valueSchema);
        validator.validate(insert, RowKind.INSERT);
        insert.setRowKind(RowKind.INSERT);
        emit(record, insert, out);
    // DELETE 이벤트 처리
    } else if (op == Envelope.Operation.DELETE) {
        GenericRowData delete = extractBeforeRow(value, valueSchema);
        validator.validate(delete, RowKind.DELETE);
        delete.setRowKind(RowKind.DELETE);
        emit(record, delete, out);
    // UPDATE 이벤트 처리
    } else {
        if (changelogMode == DebeziumChangelogMode.ALL) {
            GenericRowData before = extractBeforeRow(value, valueSchema);
            validator.validate(before, RowKind.UPDATE_BEFORE);
            before.setRowKind(RowKind.UPDATE_BEFORE);
            emit(record, before, out);
        }

        GenericRowData after = extractAfterRow(value, valueSchema);
        validator.validate(after, RowKind.UPDATE_AFTER);
        after.setRowKind(RowKind.UPDATE_AFTER);
        emit(record, after, out);
    }
}

<코드 1. 플링크 CDC의 메시지 역직렬화 코드>

하지만 스파크 SQL로 DDL을 수행해도 hive.metastore.disallow.incompatible.col.type.change의 기본 값은 true이기 때문에, 칼럼 타입 변화가 기본 설정으로 지원되지는 않습니다. DDL 이벤트가 칼럼 타입 변화를 일으키는지 판단하는 기준은 아이스버그 메타데이터 파일의 관점에서 생각하면 됩니다. 예를 들어, 메타데이터 파일에 존재하는 파티션 정보에서 n번째 칼럼의 타입이 DDL 이벤트 수행 후 바뀐다면 타입 변화로 판단합니다. 칼럼 순서를 변경하는 것 또한 메타데이터 파일의 관점에서는 n번째 칼럼의 타입이 변경되는 것이라 허용되지 않습니다. 만약 동일 타입의 칼럼들로만 테이블이 구성되었다면 순서 변경이 성공할 수도 있습니다. 해당 설정 변경 없이 칼럼 타입을 변경하려면 아래의 에러 메시지 1이 발생합니다. 따라서 타입 변경을 위해서는 hive.metastore.disallow.incompatible.col.type.changes를 false로 변경해야 합니다.

The following columns have types incompatible with  the existing columns in their respective positions

<에러 메시지 1. 아이스버그 테이블에 DDL 수행 시 발생 가능한 에러 메시지>

카탈로그와 네임스페이스 설정

카탈로그에는 테이블 타입, 카탈로그 타입 및 파일들이 저장될 경로를 설정해야 합니다. 테이블 타입은 항상 iceberg이며, 카탈로그 타입은 하이브 메타스토어를 사용하기에 hive로 명시합니다. 파일들이 저장되는 경로인 WAREHOUSE_LOCATION 설정은 모든 테이블들의 데이터 및 메타데이터 관련 파일들이 저장되는 경로입니다. 따라서 네임스페이스와 테이블명을 혼합하여 테이블 별로 경로를 구분하여 설정하시는 것을 추천드립니다. 이외에도 하이브를 카탈로그로 사용하려면 하이브 메타스토어의 쓰리프트 URI를 uri 설정에 추가해야 하지만, 저희는 하둡 및 하이브 관련 설정이 플링크 클러스터에 사용되는 컨테이너 이미지에 이미 설정되어 있어 추가적인 설정을 하지는 않았습니다.

마지막으로 네임스페이스는 필수로 지정해야 할 설정이 없습니다. 다만 네임스페이스와 테이블의 소유주를 미리 지정한다면, 추후 운영 및 권한 관리 관점에서 유용할 것으로 판단하였습니다. 이러한 이유로 아래와 같이 네임스페이스와 테이블의 소유주 관련 설정을 추가하였습니다.

import org.apache.iceberg.flink.CatalogLoader  
  
hiveCatalogProps.put("type", "iceberg")  
hiveCatalogProps.put("catalog-type", "hive")  
hiveCatalogProps.put("warehouse", s"hdfs://hadoop-cluster/../../${namespace}/${table}")  
val hivecatalog = CatalogLoader.hive("catalog_name", hadoopConf, hiveCatalogProps) // 기존 카탈로그 로드 또는 생성  
  
namespaceProps.put(HiveCatalog.HMS_TABLE_OWNER, "seungmin-lee")  
namespaceProps.put(HiveCatalog.HMS_DB_OWNER, "seungmin-lee")  
hiveCatalog.createNamespace(Namespace.of("namespace_name"), namespaceProps) // 기존 네임스페이스 로드 또는 생성

<코드 2. 카탈로그와 네임스페이스 설정 및 생성 예시>

테이블 설정

아이스버그 테이블 설정은 사용하는 환경이나 목적 외에도, 팀 내부 정책에 맞추어야 하는 부분들이 있었습니다. 먼저, 저희 팀에서는 작업을 다시 수행할 경우의 멱등성 보장을 위해 UPSERT 모드로 데이터를 적재하기에, write.upsert.enabled을 true로 설정해야 했습니다. 또한 UPSERT 모드 사용 시 2 이상의 format-version을 사용해야 합니다. 글 작성 시점인 2024년 10월 기준 format-version 3은 아직 개발 단계이며 공식적으로 채택되지 않았기 때문에, 현재 format-version은 2를 사용하고 있습니다. 또한 저희 팀에서는 사내에서 제공하는 하이브나 트리노 엔진을 통한 아이스버그 테이블 조회도 염두하고 있어 engine.hive.enabled를 true로 설정했습니다.

파일 타입과 압축 방법은 기본 값인 parquet과 zstd를 사용합니다. 압축 방법은 아이스버그 버전 1.4까지는 기본값이 gzip이었는데, 아파치 트리노(Apache Trino)를 통한 조회 시 zstd가 조회나 압축 관점에서 더 좋은 성능과 높아진 GC(Garbage Collection) 안정성을 보였기 때문에, 이후 아이스버그 버전에서 압축 방법이 zstd로 변경되었습니다.

또한 메타데이터 파일은 자동으로 지워지도록 write.metadata.delete-after-commit.enabled를 true로 설정합니다. 하둡 파일 시스템을 객체 저장소로 사용하기에, 하둡 블록 사이즈보다 작은 파일들이 계속 생성되는 것은 하둡 파일 시스템의 입출력 성능에 부정적인 영향을 끼칩니다. 또한 최신의 메타데이터 파일에는 후술 될 스냅샷 만료 기능을 사용하지 않는 한, 과거의 스냅샷 정보들이 전부 남아 있기에 과거 메타데이터 파일들을 전부 유지할 필요가 없습니다. 유지 개수도 설정 가능한데 초기 값인 100을 사용하고 있습니다.

다음으로 설명드릴 부분은 커밋과 쓰기 관련 설정들로, 운영의 안정성과 직결된 부분입니다. 먼저 커밋 실패에 대한 안정성을 향상하기 위해 재시도 횟수를 초기값인 4회에서 60회로 늘렸습니다. 대신 커밋의 재시도 시간 상한을 기본값인 30분에서 5분으로 낮춰 커밋이 지속적으로 실패하더라도 횟수 기준으로 60회, 시간 기준으로는 5분의 마지노선을 두었습니다. 쓰기 관련 설정에는 격리(isolation) 레벨 설정도 있는데, 동시에 여러 연산자가 데이터를 지우거나 갱신할 경우 어느 정도로 순서를 통제할지 결정하는 설정입니다. 저희는 구현 상 CDC 연동 시 증분 스냅샷 단계를 제외하면, 아이스버그 테이블에 유효한 쓰기 연산은 항상 하나만 존재합니다. 이렇게 쓰기 연산이 항상 하나이기 때문에 쓰기 연산들의 순서가 엉키는 경우는 없겠지만, 만약의 경우를 대비하여 기본값이면서 가장 강하게 동시성을 제한하는 serializable를 사용하고 있습니다.

tableProperties.put(TableProperties.COMMIT_NUM_RETRIES, "60") // 기본값 4에서 60으로 증가
tableProperties.put(TableProperties.COMMIT_TOTAL_RETRY_TIME_MS, "300000") // 기본값 30분에서 5분으로 감소
tableProperties.put(TableProperties.DEFAULT_FILE_FORMAT, "parquet")  
tableProperties.put(TableProperties.ENGINE_HIVE_ENABLED, "true")  
tableProperties.put(TableProperties.FORMAT_VERSION, "2")  
tableProperties.put(TableProperties.METADATA_DELETE_AFTER_COMMIT_ENABLED, "true")  
tableProperties.put(TableProperties.PARQUET_COMPRESSION, "zstd")  
tableProperties.put(TableProperties.UPSERT_ENABLED, "true")

<코드 3. 아이스버그 테이블 설정 및 생성 예시>

파티션은 아이스버그 테이블 주요 설정 파트에서 설명드린 것처럼 반드시 설정해야 하는 기능입니다. 저희는 기본적으로 기본키에 대해 버킷 파티션을 설정합니다. 만약 기본키가 두 개 이상의 칼럼으로 구성된 복합키라면 카디널러티(Cardinality)가 더 높은 칼럼을 사용합니다. 내부적으로 버킷 파티션의 모듈로(Modulo) 값을 정할 때는 5, 10, 25, 50의 값을 테스트했습니다. 위에서 설명드린 내용처럼 너무 세분화된 파티션은 오히려 조회 성능에 부정적 영향을 끼칠 수 있다고 말씀드렸는데요, 실제로 테스트한 결과 모듈로의 값이 50 이상부터는 스파크를 통한 테이블 조회 시 수행시간이 늘어나는 경향을 보였습니다. 이에 너무 세분화하지 않는 것이 좋다고 판단하여 최종 값을 5로 설정하였습니다.

마지막으로 쓰기 모드 설정의 경우 저희 팀에서는 아이스버그의 플링크 API에서 테이블 생성 시 기본적으로 설정되는 COW를 사용합니다. 사용 환경이나 방식에 따라 성능에 막대한 영향을 끼치는 설정이지만, 플링크에서 아이스버그 테이블로 데이터를 적재하는 로직이 항상 MOR로 동작하기 때문입니다. 또한 저희 팀의 사용 환경은 플링크만이 쓰기 연산을 수행하며, 읽기 연산은 스파크를 통해서만 수행하기 때문에 해당 설정은 실질적인 의미가 없어 명시하지 않았습니다.

플링크에서 아이스버그로 적재 과정

플링크에서 아이스버그로 바로 데이터를 적재하게 되면, 카프카 커넥트 활용 시 사용되는 디비지움의 Change event 포맷이 아닌 RowData 포맷을 사용합니다. 이번 파트에서는 먼저 RowData 포맷과 Change event 포맷의 차이를 설명드리며, RowData 포맷을 사용하면 어떠한 이점이 있는지 설명드리겠습니다. 이후 플링크에서 아이스버그 테이블로 데이터 적재 시 필요한 플링크 다이내믹 테이블(Flink Dynamic Table)과 아이스버그 테이블을 어떻게 동적으로 생성하여 사용하는지 말씀드리겠습니다. 마지막으로 플링크에서 아이스버그 테이블로 데이터 적재시 존재하는 소스(Source), 쓰기(Writer) 및 커밋터(Committer) 3개의 연산자들의 동작 과정을 코드와 함께 설명드리겠습니다.

본격적으로 설명에 들어가기에 앞서 각 연산자의 역할을 간략히 공유드리면 아래와 같습니다.

소스(Source): MySQL 테이블에서 데이터를 읽고 RowData 포맷의 메시지를 생성합니다.
쓰기(Writer): 상위 연산자가 건네준 메시지의 이벤트 타입에 맞춰 아이스버그 테이블에 데이터를 적재합니다.
커밋터(Committer): 플링크 잡이 체크포인트 수행 시 아이스버그 테이블에 커밋을 수행합니다.

그림 4. 아이스버그로 적재하는 플링크 잡 예시

RowData 포맷

디비지움을 기반으로 하는 CDC 연동은 대부분 카프카와 카프카 커넥트를 사용하며, 디비지움의 Change event 포맷으로 메시지를 전송합니다. Change event 포맷은 스키마 정보, 쿼리 수행 전 후의 데이터, 원천 데이터베이스 정보 및 라이브러리 정보와 같이 다양한 정보들을 담고 있는 JSON 타입의 포맷입니다. 아래의 예시 2의 내용이 Change event 포맷 메시지 예시인데요, 예시를 통해 알 수 있듯 다양한 정보를 담고 있기 때문에 무거워진 메시지는 메시지 처리율에 부정적인 영향을 줍니다.

만약 메시지를 가볍게 하고 싶다면 스키마 레지스트리(Schema Registry)를 사용하여 스키마 정보를 생략할 수 있지만 스키마 레지스트리 목적의 서버를 구축하고 운영해야 한다는 단점이 남습니다. 이전 글인 아파치 플링크와 CDC의 만남. 플링크 CDC 맛보기에서도 카프카와 카프카 커넥트를 통해 타겟 MySQL 테이블에 데이터를 적재하고 있어서 Change event 포맷을 사용하고 있다고 소개드린 바 있습니다. 플링크에서 타겟 MySQL 테이블로 바로 데이터를 적재하지 못하는 이유는, 플링크 CDC는 소스 커넥터 위주로 데이터 수집 기능을 제공하기 때문입니다.

{
  "schema": {
    "type": "struct",
    "fields": [
      {
        "type": "struct",
        "fields": [
          {
            "type": "int32",
            "optional": false,
            "field": "id"
          },
          {
            "type": "string",
            "optional": true,
            "default": "Unknown",
            "field": "first_name"
          },
          ...
        ],
        "optional": true,
        "name": "mysql_binlog_source.source_database.source_table.Value",
        "field": "before"
      },
      {
        "type": "struct",
        "fields": [
          {
            "type": "int32",
            "optional": false,
            "field": "id"
          },
          {
            "type": "string",
            "optional": true,
            "default": "Unknown",
            "field": "first_name"
          },
          ...
        ],
        "optional": true,
        "name": "mysql_binlog_source.source_database.source_table.Value",
        "field": "after"
      },
      ...
    ],
    "optional": false,
    "name": "mysql_binlog_source.source_database.source_table.Envelope"
  },
  "payload": {
    "before": null,
    "after": {
      "id": 23,
      "first_name": "seungmin",
      ...
    },
    "source": {
      "connector": "mysql",
      "name": "mysql_binlog_source",
      "db": "source_database",
      "table": "source_table",
      ...
    },
    "op": "r",
    "ts_ms": 1726664542015,
    "transaction": null
  }
}

<예시 2. Change event 포맷 메시지 예시>

하지만 아이스버그는 플링크에서 바로 적재가 가능하도록 API를 제공합니다. 즉, 플링크 CDC의 소스 커넥터를 사용하여, MySQL 데이터를 가져온 후 아이스버그의 플링크 적재 기능을 통해 아이스버그 테이블로 데이터를 적재할 수 있습니다. 따라서 카프카를 사용하지 않기에 Change event 포맷을 사용하지 않아도 되며, 대신 플링크의 RowData 포맷을 사용합니다. 이처럼 RowData 포맷을 사용하여 읽어온 테이블 데이터와 바이너리 로그는 플링크에서 3종류의 타입으로 분류됩니다. 분류되는 메시지 타입은 아래의 예시 3에서 확인할 수 있듯 +I(INSERT), +U(UPDATE_AFTER) 그리고 -D(DELETE)가 존재합니다.

+I(23,seungmin,lee,seungmin@example.com,010-4321-9876,Employee,30000.00,1,1)
+U(24,Unknown,Unknown,null,None,None,0.00,-1,0)
-D(25,gildong,hong,gildong@example.com,010-1234-5678,Employee,50000.00,1,1)

<예시 3. RowData 포맷 메시지 예시>

이렇게 간단한 포맷으로 CDC가 가능한 이유는 플링크 잡에 필요한 모든 테이블들의 정보가 이미 존재하기 때문입니다. 해당 내용은 본 파트에 이어서 바로 기술된 테이블 동적 생성 파트에서 설명드리겠습니다. 또한 하나의 플링크 잡이 하나의 아이스버그 테이블을 적재하기에 Change event 포맷처럼 데이터베이스나 테이블에 대한 정보를 메시지에 넣을 필요가 없습니다.

가벼워진 메시지 포맷은 메시지 처리율 향상과 연결되며, 사내 카프카에 대한 의존성이 사라지기 때문에 지켜야 했던 사내 카프카 가이드라인의 메시지 처리율 상한도 없어졌습니다. 결과적으로 데이터베이스 부하만 적절히 조율하면 메시지 처리율을 원하는 만큼 늘려, CDC 연동의 증분 스냅샷 단계를 더 빠르게 끝낼 수 있습니다.

성능과 수행 시간에 대해 조금 더 공유드리면 플링크 CDC로 메시지를 가져와 카프카로 전송할 경우, 플링크 잡의 병렬성 1 당 평균 5k msg/s 정도의 처리율을 보입니다. 병렬성을 늘려 처리율을 늘릴 수 있지만 사내 공용 카프카 가이드라인과 유관 조직과 협의된 테이블의 크기에 따른 메시지 처리율 상한이 각각 존재하여, 사실상 플링크의 분산 처리 기능을 CDC 연동의 증분 스냅샷 단계에서 완전히 사용할 수 없었습니다.

하지만 아이스버그 테이블로 바로 적재하는 구조에서는 카프카 의존성이 사라지기 때문에, 사내 공용 카프카 가이드라인을 고려하지 않아도 됩니다. 또한 가벼워진 메시지 포맷으로 플링크 잡의 병렬성 1 당 평균 15k msg/s 정도의 처리율이 나옵니다. 유일하게 고려해야 하는 부분이 데이터베이스의 부하인데, 이를 사내 DBA 담당자분과 부하의 관점에서 성능 상한 테스트를 진행한 결과, 최대 45 정도의 병렬성까지는 사용이 가능할 것으로 확인되었습니다. 혹시 모를 상황을 위해 병렬성 40을 기준으로 메시지 처리율을 계산해 보면, 예상되는 메시지 전송률의 상한은 600k msg/s 정도이며 이는 사내 공용 카프카 가이드라인에 존재하는 상한을 훨씬 상회하는 처리율입니다. 최근 아이스버그 테이블로 연동한 2개의 플링크 잡들의 초당 메시지 처리율 지표를 공유드리면, 아래 그림 5에서 확인 가능한 것 처럼 각각 초당 최대 360k msg/s, 280k msg/s의 처리율을 보였는데요, 이때 두 플링크 잡의 병렬성은 동일하게 20으로 설정되었습니다.

그림 5. 아이스버그 테이블로 적재시 증분 스냅샷 단계의 처리율

테이블 동적 생성

플링크에서 아이스버그 테이블로 데이터를 적재하는 플링크 잡을 실행할 때는 동적으로 두 개의 테이블을 생성해야 합니다. 첫 번째는 플링크 다이내믹 테이블이며 두 번째는 아이스버그 테이블입니다. 이번 파트에서는 두 테이블을 생성하는 데 필요한 테이블 스키마를 어떻게 동적으로 생성하여 사용하는지 소개드리겠습니다. 이후 아이스버그 테이블 테이블을 플링크를 통해 생성하는 이유와 각 시스템 별로 정의된 테이블들의 칼럼 타입들을 맵핑한 방법에 대해서도 하나씩 기술하겠습니다.

플링크의 다이내믹 테이블은 플링크에서 아이스버그로 테이블 적재 시 반드시 사용해야 하는 기능입니다. 플링크에서 아이스버그 테이블로 데이터를 적재하려면 플링크의 Table API를 사용해야 하며, Table API의 핵심이 플링크 다이내믹 테이블입니다. 플링크 다이내믹 테이블은 MySQL 테이블의 각 칼럼 값들을 어떤 타입으로 가져올지 결정하는 역할을 수행합니다. 플링크 다이내믹 테이블의 칼럼 타입에 맞추어 MySQL 테이블의 각 칼럼 데이터들을 읽어와 변환하기 때문입니다.

예를 들어 MySQL 테이블의 정수형 타입 칼럼을 플링크 다이내믹 테이블의 칼럼이 문자열에 대응시키면, 정수형 값을 문자열 값으로 변환합니다. 플링크 다이내믹 테이블을 생성하는 일반적인 방법은 아래의 코드 4 예시처럼 테이블 생성에 필요한 스키마 정보를 코드에 넣거나, 설정 파일에 기입 후 플링크 잡에서 동적으로 불러와 사용하는 것입니다. 하지만 테이블 스키마 정보를 코드나 설정 파일에 넣는 것은 보안 관점에서 좋지 않습니다. 또한 스키마가 예기치 않게 변할 수 있기에, 주기적으로 변할 가능성이 있는 정보를 파일로 관리하는 건 운영 관점에서도 좋지 않습니다.

아이스버그 테이블도 아래의 코드 4의 예시에 나와 있듯이 필요한 스키마 정보를 코드나 파일에 기입한 후 플링크 잡에서 생성할 수 있습니다. 다만 아이스버그 테이블 생성은 다른 방법으로도 가능한데, 스파크나 트리노와 같은 별도의 엔진을 통해 생성이 가능합니다. 일반적으로는 미리 테이블을 생성하는 방식을 많이 사용합니다. 하지만 다른 엔진을 사용하여 테이블을 생성하는 것은 플링크 외에 다른 엔진에 대한 의존성을 만들고 고려해야 할 부분이 늘어나기 때문에 결과적으로 전체 연동 과정을 복잡하게 만들게 됩니다.

// 플링크 다이내믹 테이블 스키마 코드
val flinkDynamicTableSchema = DataTypes.ROW(  
  DataTypes.FIELD("id", DataTypes.BIGINT),  
  DataTypes.FIELD("first_name", DataTypes.STRING),  
  DataTypes.FIELD("last_name", DataTypes.STRING),  
  DataTypes.FIELD("email", DataTypes.STRING),  
  DataTypes.FIELD("phone_number", DataTypes.STRING),  
  DataTypes.FIELD("job_title", DataTypes.STRING),  
  DataTypes.FIELD("salary", DataTypes.STRING),  
  DataTypes.FIELD("department_id", DataTypes.INT),  
  DataTypes.FIELD("is_active", DataTypes.INT)
)


// 아이스버그 테이블 스키마 코드
val icebergTableSchema = new Schema(  
  Types.NestedField.required(1, "id", Types.LongType.get),  
  Types.NestedField.optional(2, "first_name", Types.StringType.get),  
  Types.NestedField.optional(3, "last_name", Types.StringType.get),  
  Types.NestedField.optional(4, "email", Types.StringType.get),  
  Types.NestedField.optional(5, "phone_number", Types.StringType.get),  
  Types.NestedField.optional(6, "job_title", Types.StringType.get),  
  Types.NestedField.optional(7, "salary", Types.StringType.get),  
  Types.NestedField.optional(8, "department_id", Types.IntegerType.get),  
  Types.NestedField.optional(9, "is_active", Types.IntegerType.get)
)

<코드 4. 스키마 정보를 코드에 넣어 사용하는 예시>

칼럼 타입과 관련된 이슈도 존재하는데 MySQL의 칼럼 타입, 플링크의 다이내믹 테이블의 칼럼 타입 그리고 아이스버그 테이블의 칼럼 타입들은 개별적인 시스템에서 정의된 타입입니다. 즉, MySQL의 칼럼 타입과 플링크 다이내믹 테이블의 칼럼 타입이 호환되지 않거나, 플링크 다이내믹 테이블의 칼럼 타입과 아이스버그 테이블의 칼럼 타입이 호환되지 않을 수 있습니다.

저희는 이 문제들을 해소하기 위해 예시 4의 경우처럼 플링크 잡에서 MySQL에 DESCRIBE TABLE 구문을 먼저 수행하고, 수행 결과를 사용하여 동적으로 플링크 다이내믹 테이블과 아이스버그 테이블 생성에 필요한 스키마를 생성하여 사용하도록 구현하였습니다.

// MySQL 테이블 스키마
+---------------+--------------+------+-----+---------+-------+
| Field         | Type         | Null | Key | Default | Extra |
+---------------+--------------+------+-----+---------+-------+
| id            | bigint       | NO   | PRI | NULL    |       |
| first_name    | varchar(50)  | NO   |     |         |       |
| last_name     | varchar(50)  | NO   |     |         |       |
| email         | varchar(100) | YES  |     | NULL    |       |
| phone_number  | varchar(20)  | NO   |     | NULL    |       |
| job_title     | varchar(50)  | YES  |     | NULL    |       |
| salary        | float        | NO   |     | 0.00    |       |
| department_id | int          | NO   |     | -1      |       |
| is_active     | tinyint(1)   | NO   |     | 1       |       |
+---------------+--------------+------+-----+---------+-------+


// MySQL 테이블 스키마를 기반으로 동적으로 생성되는 플링크 다이내믹 테이블 스키마
ROW<
	`id` BIGINT, 
	`first_name` STRING, 
	`last_name` STRING, 
	`email` STRING, 
	`phone_number` STRING, 
	`job_title` STRING, 
	`salary` STRING, 
	`department_id` INT, 
	`is_active` INT
	>	
	
// MySQL 테이블 스키마를 기반으로 동적으로 생성되는 아이스버그 테이블 스키마
table {
	1: id: required long (id) // 동등 칼럼으로 id 칼럼을 설정
	2: first_name: required string
	3: last_name: required string
	4: email: optional string
	5: phone_number: required string
	6: job_title: optional string
	7: salary: required string
	8: department_id: required int
	9: is_active: required int
}

<예시 4. 스키마 동적 생성 예시>

칼럼 타입 맵핑의 경우 저희 팀의 주된 목표는 지표를 추출하는 것이기 때문에, 완전히 동일한 타입을 가져갈 이유도 없을뿐더러 실제로 가능하지도 않습니다. 대신 라이브러리 코드 확인 및 테스트를 통해 각 칼럼 타입들 간의 변환이 호환되고, 지표 추출에도 문제가 없는 선으로 맵핑 룰을 정하여 사용하고 있습니다. 아래의 코드 5가 라이브러리 상의 변환에 관련한 코드 예시이며, 타입별로 정의된 convert 함수들을 보고 변환이 허용되는지 파악하였습니다.

// 플링크 다이나믹 테이블의 칼럼 타입이 Int 타입인 경우
public Object convert(Object dbzObj, Schema schema) {  
    if (dbzObj instanceof Integer) {  
        return dbzObj;  
    } else {  
        return dbzObj instanceof Long ? ((Long)dbzObj).intValue() : Integer.parseInt(dbzObj.toString());  
    }  
}


// 플링크 다이나믹 테이블의 칼럼 타입이 Long 타입인 경우
public Object convert(Object dbzObj, Schema schema) {  
    if (dbzObj instanceof Integer) {  
        return ((Integer)dbzObj).longValue();  
    } else {  
        return dbzObj instanceof Long ? dbzObj : Long.parseLong(dbzObj.toString());  
    }

<코드 5. 타입 변환 코드 예시>

실제 저희 팀이 사용 중인 칼럼 타입 맵핑 룰을 공유드리면 아래의 코드 6 예시와 같습니다. 하나씩 설명을 드리면, 먼저 정수형 타입, 부동소수점 타입(float), 그리고 일부 시간 관련 타입은 동일한 타입을 설정합니다. 예시로 시간 관련 타입 중 MySQL의 datetime 타입은 플링크 다이내믹 테이블과 아이스버그 테이블에서 둘 다 지원되지 않기에 Timestamp 타입으로 맵핑합니다. 또한, MySQL의 bit 타입은 먼저 자릿수를 확인합니다. 자릿 수가 1이면 참 또는 거짓을 나타내는 데이터로 취급하고 String 타입으로 읽어와 적재하며, 이외에는 VARBINARY 타입으로 읽어와 아이스버그의 String 타입으로 적재합니다. tinyint 타입은 int 타입으로 맵핑하는데, 일반적으로 0과 1로 참 또는 거짓을 나타내는 데 사용되나 종종 2 이상의 값을 저장하여 사용하는 경우가 있습니다. 이 경우 String 타입으로 변환하면 2 이상의 값이 전부 true로 변환되어 지표 추출에 문제가 발생하기 때문입니다. 이 외의 타입들은 전부 String 타입으로 변환하여 적재합니다.

import org.apache.flink.table.types.DataType
import org.apache.iceberg.types.Types


// desc table 결과 기준, 플링크 다이나믹 테이블 스키마 타입 맵핑 룰
case "int" => DataTypes.INT()  
case "bigint" => DataTypes.BIGINT()  
case "tinyint" => DataTypes.INT() 
case "bit" if n > 1 => DataTypes.VARBINARY(n)  
case "date" => DataTypes.DATE()  
case "datetime" => DataTypes.TIMESTAMP(n)  
case "timestamp" => DataTypes.TIMESTAMP(n)  
case "float" => DataTypes.FLOAT()
case _ => DataTypes.STRING()


// desc table 결과 기준, 아이스버그 테이블 스키마 타입 맵핑 룰
case "int" => Types.IntegerType.get  
case "bigint" => Types.LongType.get  
case "tinyint" => Types.IntegerType.get  
case "datetime" => Types.TimestampType.withZone()  
case "timestamp" => Types.TimestampType.withZone()  
case "date" => Types.DateType.get  
case "float" => Types.FloatType.get
case _ => Types.StringType.get

<코드 6. 칼럼 타입 맵핑 예시>

소스 연산자

위에서 테이블들이 동적으로 생성되면 이제 소스 연산자가 MySQL에서 데이터를 읽어오기 시작합니다. 소스 연산자는 증분 스냅샷 단계에서는 테이블 데이터를, 빈로그 스트림 단계에서는 바이너리 로그를 읽어와 플링크 다이내믹 테이블 스키마에 맞춰 칼럼 값들을 변환하고, 이후 RowData 타입의 메시지를 생성합니다.

소스 연산자의 동작 과정을 간략히 설명드리면 아래와 같습니다.

설정된 접속 정보에 맞추어 데이터베이스에 연결합니다.
MySQL 테이블의 데이터와 바이너리 로그들을 읽어 옵니다.
플링크 다이내믹 테이블의 칼럼 타입에 맞춰 읽어온 값들을 변환하고, RowData 타입의 메시지를 생성합니다.
생성된 메시지를 하위 연산자로 전송합니다.

읽어온 메시지를 RowData 포맷 메시지로 만들려면 플링크 CDC의 RowDataDebeziumDeserializeSchema 함수를 사용해야 합니다. 만약 기존 JSON 타입의 Change event 포맷으로 변환을 원하신다면 JsonDebeziumDeserializationSchema를 사용하시면 됩니다. RowDataDebeziumDeserializeSchema를 사용하려면 코드 4에서 공유드린 DataType 타입의 플링크 다이내믹 테이블 스키마를, 플링크의 TypeInformation[RowData] 타입으로 변환해야 합니다. 해당 변환은 DataType, logicalType, TypeInformation[RowData] 순서로 진행되며 staticfromDataTypeToLegacyInfo 함수를 통해 한번에 변환할 수 있지만, 해당 함수는 곧 사라질(Deprecated) 예정이기에 앞서 설명드린 순서로 변환하는 것을 권장드립니다. 타입 변환에는 플링크 Table API에서 제공하는 TypeConvertions과 InternalTypeInfo를 사용하면 됩니다. 아래의 코드 7 부분이 타입 변환 구현 및 사용 예시입니다.

MySqlSource.builder[A]()  
  .hostname(...)  
  .port(...)  
  ...
  .deserializer(buildRowDataDebeziumDeserializeSchema(flinkDynamicTableSchema))
  .build()
  
private def buildRowDataDebeziumDeserializeSchema(flinkDynamicTableSchema: DataType): RowDataDebeziumDeserializeSchema = {  
  val logicalType = TypeConversions.fromDataToLogicalType(flinkDynamicTableSchema)  
  val typeInfo = InternalTypeInfo.of(logicalType).asInstanceOf[TypeInformation[RowData]]  
  
  RowDataDebeziumDeserializeSchema.newBuilder  
    .setPhysicalRowType(flinkDynamicTableSchema.getLogicalType.asInstanceOf[RowType])  
    .setChangelogMode(DebeziumChangelogMode.UPSERT)  
    .setResultTypeInfo(typeInfo)  
    .build()  
}

<코드 7. RowDataDebeziumDeserializeSchema 사용 예시>

쓰기 연산자

쓰기 연산자는 상위 연산자에서 보낸 RowData 포맷의 메시지를 사용하여 아이스버그 테이블에 데이터를 적재합니다. 이때, 아이스버그에서 제공하는 플링크 적재 API를 사용할 수 있습니다. 아래의 코드 8이 아이스버그 테이블로 적재하는 함수의 예시이며, 저희 팀에서는 UPSERT 모드를 사용하기에 관련 설정을 추가하였습니다. 또한 UPSERT 모드를 사용할 경우, 동일 레코드 판단의 기준이 되는 동등 칼럼을 항상 설정해야 합니다. 예시처럼 설정이 완료되면 이제 소스 연산자에서 생성된 메시지가 쓰기 연산자를 통해 아이스버그 테이블에 적재됩니다.

import org.apache.iceberg.flink.sink.FlinkSink


FlinkSink.forRowData(rowDataFormatDataStream)  
  .table(...)  
  .tableLoader(...)  
  .upsert(true)  
  .equalityFieldColumns(pk)  
  .append()

<코드 8. 플링크 적재 함수 예시>

일전에 아이스버그는 데이터와 메타데이터들을 파일로 관리한다 설명드렸는데, 이제 파일에 메시지들이 어떻게 쓰이지 설명드리겠습니다. 아래의 예시 코드 9가 아이스버그 테이블에 데이터를 적재 시 사용되는 함수인데, 메시지 타입에 따라 그 세부적인 동작 방식이 다릅니다. 데이터가 추가되거나 변경되는 +I(INSERT)와 +U(UPDATE_AFTER) 타입의 메시지는 동일한 로직이 수행됩니다. UPSERT 모드의 경우 메시지가 삭제 파일로도 들어가며, 이는 테이블 조회 시 과거 데이터를 지우고 항상 최신의 데이터만을 보여주기 위함입니다. 테이블 조회 시 데이터가 어떤 기준으로 삭제되어 보이는지는 후술 된 스캔 플래닝(Scan planning) 파트에서 설명드릴 예정입니다. 이외에 -D(DELETE) 타입의 메시지는 항상 삭제 파일로만 들어갑니다.

public void write(RowData row) throws IOException {  
	RowDataDeltaWriter writer = route(row);  
	switch (row.getRowKind()) {  
		case INSERT:  
		case UPDATE_AFTER:  
			if (upsert) {  
				writer.deleteKey(keyProjection.wrap(row));  
			}  
			writer.write(row);  
			break;  
		case UPDATE_BEFORE:  
			if (upsert) {  
			  break;
			}  
			writer.delete(row);  
			break;  
		case DELETE:  
			if (upsert) {  
				writer.deleteKey(keyProjection.wrap(row));  
			} else {  
				writer.delete(row);  
			}  
			break;  
		default:  
			throw new UnsupportedOperationException("Unknown row kind: " + row.getRowKind());  
	}  
}

<코드 9. 아이스버그-플링크의 적재 코드>

삭제 파일에 대해 설명드리면, 삭제 파일에는 포지션 삭제 파일과 동등 삭제 파일 두 종류가 존재한다고 설명드렸습니다. 데이터 삭제 시 관련 정보가 어떤 종류의 삭제 파일에 쓰이는지가 다르며 이는 엔진 별로 구현 방식에 따라 다릅니다. 플링크에서 UPSERT 모드인 경우 삭제 메시지가 삭제 파일로 들어가는 로직은 그림 6과 같은데, 아래 흐름에 따라 삭제 파일이 결정되어 들어갑니다.

삭제 메시지의 동등 칼럼 값을 확인합니다.
동등 칼럼 값이 동일한 메시지가 쓰기 연산자의 메모리에 존재하는지 확인합니다. 동일 스냅샷 시점 및 동일 쓰기 연산자에서 한 번 이상 들어왔다면 메모리에 존재합니다.
위 조건이 만족되면 메시지가 저장된 데이터 파일 및 포지션 정보가 존재하기에, 포지션 삭제 파일로 들어갑니다.
그 외에는 동등 삭제 파일로 들어갑니다.

그림 6. 플링크에서 아이스버그 테이블로 삭제 메시지 적재 흐름 예시

커밋터 연산자

커밋터 연산자는 데이터를 가져와 적재하는 흐름과는 별개의 역할로, 플링크 잡의 체크포인트 주기에 맞추어 아이스버그 테이블에 커밋을 수행합니다. 또한, 앞의 두 연산자와 달리 사용자가 직접 구현할 내용은 없습니다. 코드 10을 보면, 사용자가 소스 연산자부터 쓰기 연산자까지 파이프라인을 구성하면, 라이브러리가 자동으로 사용자가 만든 파이프라인의 마지막에 커밋터 연산자를 추가합니다.

private  DataStreamSink chainIcebergOperators() {

    ...
	// 사용자가 정의한 파이프라인이 distributeStream으로 들어갑니다
	SingleOutputStreamOperator writerStream = appendWriter(distributeStream, flinkRowType, equalityFieldIds);  


	// 파이프라인 마지막에 커밋터 연산자를 붙여줍니다.
	SingleOutputStreamOperator committerStream = appendCommitter(writerStream);

	...
}

<코드 10. 커밋터 연산자를 붙이는 로직>

커밋의 결과로는 스냅샷이 생성되는데, 만약 들어온 메시지가 하나도 없다면 empty commit이 수행되며 아무 일도 발생하지 않습니다. 관련 설정으로 flink.max-continuous-empty-commits이 기본 값 10으로 존재합니다. 만약 empty commit이 연속해서 10회 발생하면, 변화분이 없어도 스냅샷을 생성합니다. 아래 그림 7이 empty commit에 따른 스냅샷 생성 예시입니다. 플링크 잡의 체크포인트 주기가 10분이라 아이스버그 테이블로의 커밋도 10분마다 수행되며, 연속해서 10번 empty commit이 수행될 경우 새로운 스냅샷이 생성됩니다. 연속 10번은 시간 기준으로 1시간 40분(10분 * 10회)이며, 그림 7의 메타데이터 파일 및 매니페스트 리스트 생성 시간을 보면 실제로 11시 14분, 12시 54분 및 13시 34분으로 1시간 40분마다 새로운 파일들이 생성되고 있는 것을 확인할 수 있습니다.

그림 7. empty commit 연속 수행에 따른 스냅샷 생성

시간에 따른 메타데이터 심층 탐구

이번 파트에서는 아이스버그 테이블의 메타데이터, 매니페스트 리스트 그리고 매니페스트 파일들을 실제 예시와 함께 살펴보겠습니다. 실제 동작 과정을 떠올리시기 쉽게 전달하기 위해, 아래 순서에 맞춰 각 시점에 생성된 파일들 및 각 파일들이 가지고 있는 정보들과 이 정보들이 어떻게 활용되는지 설명드리겠습니다.

아이스버그 테이블 생성 시점(첫 메타데이터 파일 생성 시점)
첫 번째 스냅샷
두 번째 스냅샷

테이블 생성 시점

아이스버그 테이블이 생성되면 그림 8과 같은 형상이 되며, 예시 5와 같은 첫 번째 메타데이터 파일이 생성됩니다.

그림 8. 아이스버그 테이블 생성시 생성되는 파일

파일에는 생성된 테이블의 유니크 아이디부터 파일 저장 위치, 스키마 그리고 파티션 정보들이 저장되어 있습니다. 스키마와 파티션 정보들은 리스트 형태로 저장되어 있는데, 이는 과거 이력들을 전부 저장하기 위함입니다. 예시 5는 테이블 생성 직후의 시점이기 때문에, 각각 하나의 정보만 있습니다. 또한 첫 번째로 할당된 스키마 아이디(schema-id)와 스펙 아이디(spec-id)이기에 해당 값에도 0이 할당됩니다.

또한 테이블 설정 파트에서 설명드린 테이블 관련 설정들도 확인할 수 있습니다. 구체적으로 하이브 관련 설정, 커밋 재시도 관련 설정, 그리고 UPSERT 모드로 설정된 것을 확인할 수 있습니다. 다만, 현재 스냅샷 아이디 current-snapshot-id는 -1로 설정되어 있는데, 실제로 데이터가 적재되고 커밋이 수행되어 스냅샷이 생성된 건 아니기 때문입니다. 동일한 이유로 스냅샷 관련 정보나 메타데이터 로그 정보에도 텅 빈 리스트가 들어 있습니다.

{
  "format-version" : 2,
  "table-uuid" : "884766bb-9cab-4298-829e-a096689d9ddc",
  "location" : "hdfs://hadoop-cluster/.../namespace/source_table",
  "last-sequence-number" : 0,
  "last-updated-ms" : 1723879432267,
  "last-column-id" : 9,
  "current-schema-id" : 0,
  "schemas" : [ {
    "type" : "struct",
    "schema-id" : 0,
    "identifier-field-ids" : [ 1 ],
    "fields" : [ {
      "id" : 1,
      "name" : "id",
      "required" : true,
      "type" : "int"
    }, 
    ... // 중략
    {
      "id" : 9,
      "name" : "is_active",
      "required" : true,
      "type" : "int"
    } ]
  } ],
  "default-spec-id" : 0,
  "partition-specs" : [ {
    "spec-id" : 0,
    "fields" : [ {
      "name" : "id_bucket",
      "transform" : "bucket[5]",
      "source-id" : 1,
      "field-id" : 1000
    } ]
  } ],
  "last-partition-id" : 1000,
  "default-sort-order-id" : 0,
  "sort-orders" : [ {
    "order-id" : 0,
    "fields" : [ ]
  } ],
  "properties" : {
    "engine.hive.enabled" : "true",
    "commit.retry.total-timeout-ms" : "300000",
    "write.format.default" : "parquet",
    "write.parquet.compression-codec" : "zstd",
    "write.upsert.enabled" : "true",
    "write.metadata.delete-after-commit.enabled" : "true",
    "commit.retry.num-retries" : "60"
  },
  "current-snapshot-id" : -1,
  "refs" : { },
  "snapshots" : [ ],
  "statistics" : [ ],
  "partition-statistics" : [ ],
  "snapshot-log" : [ ],
  "metadata-log" : [ ]
}

<예시 5. 최초 생성되는 메타데이터 파일 예시>

첫 번째 스냅샷 - 메타데이터 파일

첫 번째 스냅샷이 생성되면 그림 9와 같이 새로운 메타데이터 파일, 매니페스트 리스트, 매니페스트 파일 및 데이터 파일들이 생성됩니다. 그리고 현 메타데이터 포인터는 새로 생성된 메타데이터 파일을 가리키도록 변경됩니다.

그림 9. 첫 번째 스냅샷에서 생성된 파일들

아래 예시 6은 새로 생성된 메타데이터 파일의 예시입니다. 스냅샷 아이디가 추가되었으며, 첫 번째 생성된 스냅샷이기에 스냅샷들의 상대적 나이를 의미하는 시퀀스 넘버에는 1이 할당되었습니다. 이외에도 스냅샷 관련 정보에 새로 생성된 스냅샷 관련 요약 정보 snapshots.summary가 있으며 요약 정보에는 추가 및 삭제된 레코드와 파일 개수, 전체 파일 개수 등이 있습니다.

{
  "format-version" : 2,
  "table-uuid" : "884766bb-9cab-4298-829e-a096689d9ddc",
  "location" : "hdfs://hadoop-cluster/.../namespace/source_table",
  "last-sequence-number" : 1,
  "last-updated-ms" : 1723879779668,
  ... // 동일하여 생략 
  "current-snapshot-id" : 6943424146698635855,
  "refs" : {
    "main" : {
      "snapshot-id" : 6943424146698635855,
      "type" : "branch"
    }
  },
  "snapshots" : [ {
    "sequence-number" : 1,
    "snapshot-id" : 6943424146698635855,
    "timestamp-ms" : 1723879779668,
    "summary" : {
      "operation" : "overwrite",
      "flink.operator-id" : "e883208d19e3c34f8aaf2a3168a63337",
      "flink.job-id" : "a40a5d923f750731620feafd9bf8930d",
      "flink.max-committed-checkpoint-id" : "1",
      "added-data-files" : "5",
      "added-equality-delete-files" : "5",
      "added-position-delete-files" : "5",
      "added-delete-files" : "10",
      "added-records" : "6971",
      "added-files-size" : "49492",
      "added-position-deletes" : "346",
      "added-equality-deletes" : "6625",
      "changed-partition-count" : "5",
      "total-records" : "6971",
      "total-files-size" : "49492",
      "total-data-files" : "5",
      "total-delete-files" : "10",
      "total-position-deletes" : "346",
      "total-equality-deletes" : "6625"
    },
    "manifest-list" : "hdfs://hadoop-cluster/../namespace/source_table/metadata/snap-6943424146698635855-1-9dcf9752-e937-44bf-a020-a42c3b5b3d63.avro",
    "schema-id" : 0
  } ],
  "statistics" : [ ],
  "partition-statistics" : [ ],
  "snapshot-log" : [ {
    "timestamp-ms" : 1723879779668,
    "snapshot-id" : 6943424146698635855
  } ],
  "metadata-log" : [ {
    "timestamp-ms" : 1723879432267,
    "metadata-file" : "hdfs://hadoop-cluster/../namespace/source_table/metadata/00000-9b14051e-3ae5-4e17-84cd-6bfe52578f7e.metadata.json"
  } ]
}

<예시 6. 첫 번째 스냅샷에서 생성된 메타데이터 파일 예시>

또한 요약 정보에는 커밋을 수행하고 스냅샷을 생성한 엔진에 대한 정보도 존재합니다. 저희는 플링크를 사용하기에 플링크 관련 정보들이 포함되어 있습니다. 실제 플링크 잡의 정보와 일치하는지 확인을 위해, REST API 요청을 통해 플링크 잡의 아이디 조회 시 아래 그림 10과 같이 조회되며 아이디는 a40a5d923f750731620feafd9bf8930d입니다. 해당 값은 예시 6의 flink.job-id와 비교하면 동일한 것을 확인할 수 있습니다. 또한, flink.operator-id에는 커밋터 연산자 아이디도 존재하는데, 동일하게 REST API 요청으로 값을 확인해 보면 e883208d19e3c34f8aaf2a3168a63337로 두 값이 모두 동일합니다.
그림 10. 플링크 잡 및 연산자 관련 정보 조회 예시

플링크가 아닌 다른 엔진을 사용하여 스냅샷을 생성한 경우의 정보도 요약 정보를 통해 알 수 있는데, 예시 7은 스파크를 통해 테이블에 특정 작업을 수행하여 스냅샷을 생성하고 생성된 메타데이터 파일의 일부입니다. 이때, 이전까지의 정보에는 플링크 정보가 남아 있었다면, 해당 예시의 요약 정보에는 스파크 애플리케이션 아이디 정보가 spark.app_id에 남아 있습니다.

{
  ..
  "snapshots" : [ 
  ...
  {
    "sequence-number" : 78,
    ...
    "summary" : {
      "operation" : "overwrite",
      "flink.operator-id" : "fbb4ef531e002f8fb3a2052db255adf5",
      "flink.job-id" : "fe781e5fbddd17d43e2290453dda4aa7",
      "flink.max-committed-checkpoint-id" : "1014",
      "added-data-files" : "5",
      "added-equality-delete-files" : "5",
      "added-position-delete-files" : "3",
      "added-delete-files" : "8",
      "added-records" : "271",
      "added-files-size" : "92717",
      "added-position-deletes" : "5",
      "added-equality-deletes" : "287",
      "changed-partition-count" : "5",
      "total-records" : "3420319",
      "total-files-size" : "330131861",
      "total-data-files" : "4601",
      "total-delete-files" : "5575",
      "total-position-deletes" : "1599",
      "total-equality-deletes" : "3422410"
    },
    "manifest-list" : "hdfs://hadoop-cluster/../namespace/source_table/metadata/snap-4930217877644357147-1-5f476623-f2ec-42b0-92d8-bef7dd5cd72d.avro",
    "schema-id" : 0
  }, 
  {
    "sequence-number" : 79,
    ...
    "summary" : {
      "operation" : "overwrite",
      "spark.app.id" : "application_1727068550645_39886", // spark 관련 정보
      "added-data-files" : "989",
      "deleted-data-files" : "4601",
      "removed-equality-delete-files" : "4629",
      "removed-position-delete-files" : "938",
      "removed-delete-files" : "5567",
      "added-records" : "2966565",
      "deleted-records" : "3420319",
      "added-files-size" : "119275096",
      "removed-files-size" : "330120927",
      "removed-position-deletes" : "1594",
      "removed-equality-deletes" : "3422123",
      "changed-partition-count" : "5",
      "total-records" : "2966565",
      "total-files-size" : "119286030",
      "total-data-files" : "989",
      "total-delete-files" : "8",
      "total-position-deletes" : "5",
      "total-equality-deletes" : "287"
    },
    "manifest-list" : "hdfs://hadoop-cluster/../namespace/source_table/metadata/snap-8098152503420040164-1-db2d1c42-cd99-40cf-aea9-d322ace8bd62.avro",
    "schema-id" : 0
  } ]
  ...
}

<예시 7. 스파크로 스냅샷이 생성된 경우의 요약 정보 예시>

첫 번째 스냅샷 - 매니페스트 리스트

이제 생성된 스냅샷에 대응되는 매니페스트 리스트를 확인해 보겠습니다. 예시 8이 생성된 매니페스트 리스트 예시이며, 두 개의 매니페스트 파일에 대한 정보가 열거되어 있습니다. 만약 두 번째 스냅샷의 매니페스트 리스트라면 총 네개의 매니페스트 파일에 대한 정보가 열거되어 있으며, 후술될 스냅샷 만료 기능을 수행하기 전까지는 모든 매니페스트 파일이 누적됩니다. 또한 각 매니페스트 파일 정보에는 콘텐츠(content)가 존재하는데, 해당 값을 통해 매니페스트 파일이 데이터 파일들에 대응되는지 삭제 파일들에 대응되는지 알 수 있습니다. 0은 데이터 파일을 1은 삭제 파일을 의미하기에, 첫 번째 매니페스트 파일 9dcf9752-e937-44bf-a020-a42c3b5b3d63-m0.avro은 데이터 파일에 대응되고 두 번째 매니페스트 파일 9dcf9752-e937-44bf-a020-a42c3b5b3d63-m1.avro은 삭제 파일들에 대응됩니다.

이 외에도 매니페스트 파일들의 시퀀스 넘버, 파티션 정보, 추가 및 삭제된 파일이나 레코드 수 등의 통계 정보들이 저장되어 있습니다. 이 정보들 중 시퀀스 넘버는 테이블 조회 시 데이터 파일에 삭제 파일의 반영 여부를 결정할 때 사용되며, 반영 여부를 결정하는 세부 내용은 후술할 스캔 플래닝(Scan planning) 파트에서 자세히 설명드리겠습니다. 파티션 정보에는 파티션 칼럼의 상한(upper_bound)과 하한(lower_bound)이 저장되어 있습니다. 버킷(Bucket) 파티션의 모듈로(Modulo) 값으로 5를 사용했기에 상한과 하한은 각각 4와 0이며, 실제 저장된 값들도 base64로 인코딩 된 값으로 각각 0과 4를 의미합니다.

[{
  "manifest_path" : "hdfs://hadoop-cluster/.../namespace/source_table/metadata/9dcf9752-e937-44bf-a020-a42c3b5b3d63-m0.avro",
  "manifest_length" : 7668,
  "partition_spec_id" : 0,
  "content" : 0, // 데이터 파일을 의미
  "sequence_number" : 1,
  "min_sequence_number" : 1,
  "added_snapshot_id" : 6943424146698635855,
  "added_files_count" : 5,
  "existing_files_count" : 0,
  "deleted_files_count" : 0,
  "added_rows_count" : 6971,
  "existing_rows_count" : 0,
  "deleted_rows_count" : 0,
  "partitions" : [ {
    "contains_null" : false,
    "contains_nan" : false,
    "lower_bound" : "AAAAAA==", // base64로 인코딩. "00 00 00 00"으로 0을 의미 
    "upper_bound" : "BAAAAA=="  // base64로 인코딩. "04 00 00 00"으로 4를 의미
  } ]
},
{
  "manifest_path" : "hdfs://hadoop-cluster/.../namespace/source_table/metadata/9dcf9752-e937-44bf-a020-a42c3b5b3d63-m1.avro",
  "manifest_length" : 7616,
  "partition_spec_id" : 0,
  "content" : 1, // 삭제 파일을 의미
  "sequence_number" : 1,
  "min_sequence_number" : 1,
  "added_snapshot_id" : 6943424146698635855,
  "added_files_count" : 10,
  "existing_files_count" : 0,
  "deleted_files_count" : 0,
  "added_rows_count" : 6971,
  "existing_rows_count" : 0,
  "deleted_rows_count" : 0,
  "partitions" : [ {
    "contains_null" : false,
    "contains_nan" : false,
    "lower_bound" : "AAAAAA==",
    "upper_bound" : "BAAAAA=="
  } ]
}]

<예시 8. 첫 번째 스냅샷에서 생성된 매니페스트 리스트 예시>

첫 번째 스냅샷 - 매니페스트 파일

마지막으로 매니페스트 리스트에 존재하는 매니페스트 파일들을 확인해 보겠습니다. 먼저 각 매니페스트 파일은 데이터 파일들과 삭제 파일들에 대응된다 설명 드렸는데, 그림 11과 그림 12처럼 9dcf9752-e937-44bf-a020-a42c3b5b3d63-m0.avro 파일은 데이터 파일 00000-0-6215b83a-023c-409e-8c6d-d9460bdf23a7-00009.parquet을 포함한 데이터 파일들에 대응되며, 9dcf9752-e937-44bf-a020-a42c3b5b3d63-m1.avro 파일은 삭제 파일 00000-0-6215b83a-023c-409e-8c6d-d9460bdf23a7-00010.parquet과 00000-0-6215b83a-023c-409e-8c6d-d9460bdf23a7-00015.parquet을 포함한 삭제 파일들에 대응됩니다. 예시 그림들은 이미지가 너무 길어질 것을 고려하여 하나의 파티션에 대해서만 나타내었으며, 실제는 설명드린 것처럼 모든 데이터 파일들과 삭제 파일들에 각각 대응됩니다.

그림 11. 첫 번째 스냅샷에서 생성된 데이터 파일들 대응

그림 12. 첫 번째 스냅샷에서 생성된 삭제 파일들 대응

데이터 파일들에 대응되는 매니페스트 파일 예시인 예시 9를 확인하면 앞서 언급드린 것처럼, 해당 스냅샷에서 생성된 모든 데이터 파일에 대한 정보가 열거되어 있습니다. 예시 9에는 버킷(Bucket) 파티션에 모듈로(Modulo) 값으로 5를 사용했기에, 총 5개의 데이터 파일이 존재합니다. 이제 각 데이터 파일들에 존재하는 정보들을 확인하면, 제일 먼저 상태(Status)가 있으며 해당 매니페스트 파일이 기존 파일이 변경(0: EXISTING)된 것인지, 새롭게 생성된 건지(1: ADDED) 또는 삭제 (2: DELETED)된 것인지 나타냅니다. 예시는 전부 새로 생성 파일들이기에 값으로 1을 가집니다. 또한 각 데이터 파일들이 생성된 스냅샷 아이디와 시퀀스 넘버가 들어 있으며, 시퀀스 넘버 null의 의미는 대응되는 매니페스트 파일의 시퀀스 넘버를 상속한다는 의미입니다.
data_file에 있는 세부 정보들을 더 살펴보면 각 칼럼들에 대한 통계 정보들이 존재하며, 데이터 파일들이 어떤 버킷 파티션에 속하는지(예: id_bucket=0)도 저장되어 있습니다. 해당 정보들은 시퀀스 넘버와 유사한 역할을 하는데, 데이터 파일에 삭제 파일 반영 여부 결정할 때 사용됩니다. 예를 들어, 파티션 정보는 데이터 파일과 삭제 파일이 동일한 파티션인지 확인할 때 사용되며, 통계 정보는 동등 삭제 파일에 존재하는 동등 칼럼의 구간이 데이터 파일에 존재하는 동등 칼럼의 구간과 겹치는지 확인할 때 사용됩니다. 이렇게 제공되는 정보들을 사용하여 파일들을 합치는 자세한 과정은 스캔 플래닝 파트에서 중점적으로 설명드리겠습니다.

[{
  "status" : 1,
  "snapshot_id" : 6943424146698635855,
  "sequence_number" : null,
  "file_sequence_number" : null,
  "data_file" : {
    "content" : 0,
    "file_path" : "hdfs://hadoop-cluster/.../namespace/source_table/data/id_bucket=0/00000-0-6215b83a-023c-409e-8c6d-d9460bdf23a7-00009.parquet",
    "file_format" : "PARQUET",
    "partition" : {
      "id_bucket" : 0
    },
    "record_count" : 1632,
    "file_size_in_bytes" : 6223,
    "column_sizes" : [...],
    "value_counts" : [...],
    "null_value_counts" : [...],
    "nan_value_counts" : [ ],
    "lower_bounds" :  [...],
    "upper_bounds" :  [...],
    "key_metadata" : null,
    "split_offsets" : [ 4 ],
    "equality_ids" : null,
    "sort_order_id" : 0
  }
},
{
  "status" : 1,
  "snapshot_id" : 6943424146698635855,
  "sequence_number" : null,
  "file_sequence_number" : null,
  "data_file" : {
    "content" : 0,
    "file_path" : "hdfs://hadoop-cluster/.../namespace/source_table/data/id_bucket=1/00000-0-6215b83a-023c-409e-8c6d-d9460bdf23a7-00005.parquet",
    "file_format" : "PARQUET",
    "partition" : {
      "id_bucket" : 1
    }
    ... // 중략
},
{
  "status" : 1,
  "snapshot_id" : 6943424146698635855,
  "sequence_number" : null,
  "file_sequence_number" : null,
  "data_file" : {
    "content" : 0,
    "file_path" : "hdfs://hadoop-cluster/.../namespace/source_table/data/id_bucket=2/00000-0-6215b83a-023c-409e-8c6d-d9460bdf23a7-00001.parquet",
    "file_format" : "PARQUET",
    "partition" : {
      "id_bucket" : 2
    }
    ... // 중략
},
{
  "status" : 1,
  "snapshot_id" : 6943424146698635855,
  "sequence_number" : null,
  "file_sequence_number" : null,
  "data_file" : {
    "content" : 0,
    "file_path" : "hdfs://hadoop-cluster/.../namespace/source_table/data/id_bucket=3/00000-0-6215b83a-023c-409e-8c6d-d9460bdf23a7-00007.parquet",
    "file_format" : "PARQUET",
    "partition" : {
      "id_bucket" : 3
    }
    ... // 중략
  }
},
{
  "status" : 1,
  "snapshot_id" : 6943424146698635855,
  "sequence_number" : null,
  "file_sequence_number" : null,
  "data_file" : {
    "content" : 0,
    "file_path" : "hdfs://hadoop-cluster/.../namespace/source_table/data/id_bucket=4/00000-0-6215b83a-023c-409e-8c6d-d9460bdf23a7-00003.parquet",
    "file_format" : "PARQUET",
    "partition" : {
      "id_bucket" : 4
    }
    ... // 중략
  }
}]

<예시 9. 데이터 파일들에 대응되는 매니페스트 파일 예시>

삭제 파일들에 대응되는 매니페스트 파일은 예시 10이며, 총 10개의 삭제 파일들에 대한 정보가 있습니다. 5개의 버킷 파티션에 각각 2 종류의 삭제 파일이 생성되었기 때문에, 도합 10개의 삭제 파일이 존재하는 경우입니다. 다만 이 부분은 엔진에서 삭제 파일에 데이터를 쓰는 로직이 어떻게 구현되어 있는지에 따라 다르기 때문에 항상 10개가 아닌, 최대 10개가 생성된다고 이해하시면 됩니다. 그 외에 가지고 있는 정보는 데이터 파일에 대응되는 매니페스트와 거의 동일한데 상태 정보, 스냅샷 아이디, 시퀀스 넘버, 파일 경로, 파티션 정보 및 각 칼럼들에 대한 통계 정보 등이 있습니다.
데이터 파일에 대응되는 매니페스트 파일과 다른 부분을 찾아보면, 삭제 파일들의 content에 1 또는 2의 값이 존재하는 부분입니다. 이때, 1은 포지션 삭제 파일을 의미하고 2는 동등 삭제 파일을 의미합니다. 또한 동등 삭제 파일에는 동등 칼럼의 번호가 equality_ids에 기입되어 있는데, 예시에는 1과 2의 값이 존재하고 있으며, 첫 번째와 두 번째 칼럼이 동등 칼럼으로 사용된다는 것을 의미입니다.

[{
  "status": 1,
  "snapshot_id": 6943424146698635855,
  "sequence_number": null,
  "file_sequence_number": null,
  "data_file": {
    "content": 2, // 동등 삭제 파일을 의미
    "file_path": "hdfs://hadoop-cluster/.../namespace/source_table/data/id_bucket=0/00000-0-6215b83a-023c-409e-8c6d-d9460bdf23a7-00010.parquet",
    "file_format": "PARQUET",
    "partition": {
      "id_bucket": 0
    },
    "record_count": 1555,
    "file_size_in_bytes": 1983,
    "column_sizes": [...],
    "value_counts": [...],
    "null_value_counts": [...],
    "nan_value_counts": [],
    "lower_bounds": [...],
    "upper_bounds": [...],
    "key_metadata": null,
    "split_offsets": [ 4 ],
    "equality_ids": [ 1, 2 ],
    "sort_order_id": 0
  }
},
{
  "status": 1,
  "snapshot_id": 6943424146698635855,
  "sequence_number": null,
  "file_sequence_number": null,
  "data_file": {
    "content": 1, // 우치 삭제 파일을 의미
    "file_path": "hdfs://hadoop-cluster/.../namespace/source_table/data/id_bucket=0/00000-0-6215b83a-023c-409e-8c6d-d9460bdf23a7-00015.parquet",
    "file_format": "PARQUET",
    "partition": {
      "id_bucket": 0
    },
    "record_count": 77,
    "file_size_in_bytes": 1997,
    "column_sizes": [
      {
        "key": 2147483546, // 포지션 삭제 파일의 file_path 칼럼에 대응되는 예약키
        "value": 219
      },
      {
        "key": 2147483545, // 포지션 삭제 파일의 pos 칼럼 대응되는 예약키
        "value": 154
      }
    ],
    "value_counts": null,
    "null_value_counts": null,
    "nan_value_counts": null,
    "lower_bounds": [...],
    "upper_bounds": [...],
    "key_metadata": null,
    "split_offsets": [ 4 ],
    "equality_ids": null,
    "sort_order_id": null
  }
},
{
  "status": 1,
  "snapshot_id": 6943424146698635855,
  "sequence_number": null,
  "file_sequence_number": null,
  "data_file": {
    "content": 2,
    "file_path": "hdfs://hadoop-cluster/.../namespace/source_table/data/id_bucket=1/00000-0-6215b83a-023c-409e-8c6d-d9460bdf23a7-00006.parquet",
    "file_format": "PARQUET",
    "partition": {
      "id_bucket": 1
    },
    ... // 중략
  }
},
{
  "status": 1,
  "snapshot_id": 6943424146698635855,
  "sequence_number": null,
  "file_sequence_number": null,
  "data_file": {
    "content": 1,
    "file_path": "hdfs://hadoop-cluster/.../namespace/source_table/data/id_bucket=1/00000-0-6215b83a-023c-409e-8c6d-d9460bdf23a7-00011.parquet",
    "file_format": "PARQUET",
    "partition": {
      "id_bucket": 1
    },
    ... // 중략
  }
},
{
  "status": 1,
  "snapshot_id": 6943424146698635855,
  "sequence_number": null,
  "file_sequence_number": null,
  "data_file": {
    "content": 2,
    "file_path": "hdfs://hadoop-cluster/.../namespace/source_table/data/id_bucket=1/00000-0-6215b83a-023c-409e-8c6d-d9460bdf23a7-00006.parquet",
    "file_format": "PARQUET",
    "partition": {
      "id_bucket": 2
    },
	... // 중략
  }
},
{
  "status": 1,
  "snapshot_id": 6943424146698635855,
  "sequence_number": null,
  "file_sequence_number": null,
  "data_file": {
    "content": 1,
    "file_path": "hdfs://hadoop-cluster/.../namespace/source_table/data/id_bucket=1/00000-0-6215b83a-023c-409e-8c6d-d9460bdf23a7-00011.parquet",
    "file_format": "PARQUET",
    "partition": {
      "id_bucket": 2
    },
	... // 중략
  }
},
{
  "status": 1,
  "snapshot_id": 6943424146698635855,
  "sequence_number": null,
  "file_sequence_number": null,
  "data_file": {
    "content": 2,
    "file_path": "hdfs://hadoop-cluster/.../namespace/source_table/data/id_bucket=2/00000-0-6215b83a-023c-409e-8c6d-d9460bdf23a7-00002.parquet",
    "file_format": "PARQUET",
    "partition": {
      "id_bucket": 3
    },
	... // 중략
  }
},
{
  "status": 1,
  "snapshot_id": 6943424146698635855,
  "sequence_number": null,
  "file_sequence_number": null,
  "data_file": {
    "content": 1,
    "file_path": "hdfs://hadoop-cluster/.../namespace/source_table/data/id_bucket=3/00000-0-6215b83a-023c-409e-8c6d-d9460bdf23a7-00013.parquet",
    "file_format": "PARQUET",
    "partition": {
      "id_bucket": 3
    },
	... // 중략
  }
},
{
  "status": 1,
  "snapshot_id": 6943424146698635855,
  "sequence_number": null,
  "file_sequence_number": null,
  "data_file": {
    "content": 2,
    "file_path": "hdfs://hadoop-cluster/.../namespace/source_table/data/id_bucket=4/00000-0-6215b83a-023c-409e-8c6d-d9460bdf23a7-00004.parquet",
    "file_format": "PARQUET",
    "partition": {
      "id_bucket": 4
    },
	... // 중략
  }
},
{
  "status": 1,
  "snapshot_id": 6943424146698635855,
  "sequence_number": null,
  "file_sequence_number": null,
  "data_file": {
    "content": 1,
    "file_path": "hdfs://hadoop-cluster/.../namespace/source_table/data/id_bucket=4/00000-0-6215b83a-023c-409e-8c6d-d9460bdf23a7-00014.parquet",
    "file_format": "PARQUET",
    "partition": {
      "id_bucket": 4
    },
	... // 중략
  }
}]

<예시 10. 삭제 파일들에 대응되는 매니페스트 파일 예시>

두 번째 스냅샷

그림 13. 두 번째 스냅샷에서 생성된 파일들

두 번째 스냅샷 생성 후 추가된 파일들은 첫 번째 스냅샷 생성시 생성된 부분들과 동일합니다. 다른 점이라면, 현재 메타데이터 포인터가 새로 생성된 메타데이터 파일을 가리키며, 새로 생성된 메타데이터 파일에는 새로운 스냅샷이 추가되었다는 점입니다. 또한 예시 11에서 확인할 수 있듯 시퀀스 넘버가 2인 스냅샷 정보가 추가 되었으며 메타데이터와 스냅샷 로그에도 이제 2개의 정보가 각각 들어가 있는 것을 볼 수 있습니다. 이처럼 커밋이 수행되어 새로운 스냅샷이 생성되면 기존 스냅샷들의 정보들도 최신 메타데이터 파일에 전부 누적이 됩니다. 이러한 특성을 기반으로 아이스버그는 과거 상태 조회 기능인 타임 트레블 기능을 지원합니다. 다만 아이스버그 테이블 사용 목적과 환경에 따라 과거 스냅샷 정보들을 유지할 필요가 없을 수 있기에, 아이스버그는 스냅샷 만료라는 최적화 기능을 지원합니다. 스냅샷 만료 기능에 관한 보다 자세한 내용은 후술된 스냅샷 만료와 고아 파일 제거 파트에서 설명드리겠습니다.

{
  "format-version" : 2,
  "table-uuid" : "884766bb-9cab-4298-829e-a096689d9ddc",
  "location" : "hdfs://hadoop-cluster/.../namespace/source_table",
  "last-sequence-number" : 2,
  ... // 중략
  "snapshots" : [ {
    "sequence-number" : 1,
    "snapshot-id" : 6943424146698635855,
    "timestamp-ms" : 1723879779668,
    "summary" : {
      "operation" : "overwrite",
      "flink.operator-id" : "e883208d19e3c34f8aaf2a3168a63337",
      "flink.job-id" : "a40a5d923f750731620feafd9bf8930d",
      "flink.max-committed-checkpoint-id" : "1",
	  ... // 중약
    },
    "manifest-list" : "hdfs://hadoop-cluster/.../namespace/source_table/metadata/snap-6943424146698635855-1-9dcf9752-e937-44bf-a020-a42c3b5b3d63.avro",
    "schema-id" : 0
  }, {
    "sequence-number" : 2, // 새로운 스냅샷
    "snapshot-id" : 3185425291985982690,
    "parent-snapshot-id" : 6943424146698635855,
    "timestamp-ms" : 1723880377728,
    "summary" : {
      "operation" : "overwrite",
      "flink.operator-id" : "e883208d19e3c34f8aaf2a3168a63337",
      "flink.job-id" : "a40a5d923f750731620feafd9bf8930d",
      "flink.max-committed-checkpoint-id" : "2",
      ...
    },
    "manifest-list" : "hdfs://hadoop-cluster/.../namespace/source_table/metadata/snap-3185425291985982690-1-941930ea-ffff-41aa-991c-28fbd38c232f.avro",
    "schema-id" : 0
  } ],
  "statistics" : [ ],
  "partition-statistics" : [ ],
  "snapshot-log" : [ {
    "timestamp-ms" : 1723879779668,
    "snapshot-id" : 6943424146698635855
  }, {
    "timestamp-ms" : 1723880377728,
    "snapshot-id" : 3185425291985982690
  } ],
  "metadata-log" : [ {
    "timestamp-ms" : 1723879432267,
    "metadata-file" : "hdfs://hadoop-cluster/.../namespace/source_table/metadata/00000-9b14051e-3ae5-4e17-84cd-6bfe52578f7e.metadata.json"
  }, {
    "timestamp-ms" : 1723879779668,
    "metadata-file" : "hdfs://hadoop-cluster/.../namespace/source_table/metadata/00001-66289348-dc78-4ffc-93e8-28a2f933b88e.metadata.json"
  } ]
}

<예시 11. 두 번째 스냅샷 후 생성된 메타데이터 파일 예시>

아이스버그 테이블 조회 방식 및 최적화

플링크에서 아이스버그로 데이터를 적재할 때, 데이터 파일과 두 종류의 삭제 파일이 함께 저장되며, 테이블 조회 시 이들 파일들을 합쳐서 보여준다는 설명을 드린 바 있습니다. 이처럼 파일들을 합칠 때 매니페스트 리스트와 파일에 있는 정보들을 참고하게 되는데요, 이번 파트에서는 아이스버그 테이블을 조회할 시 데이터 파일의 어떤 삭제 파일들이, 어떤 정보들을 참고하여 어떻게 반영되는지 설명드릴 예정입니다. 또한, 성능과 사용 환경에 따라 주기적으로 수행해야 할 아이스버그 최적화 기능에 대해서도 소개드리도록 하겠습니다.

스캔 플래닝(Scan planning)

스캔 플래닝은 아이스버그 테이블 조회 시, 매니페스트 파일에 있는 정보들을 사용하여 필요한 데이터 파일과 삭제 파일들을 선정하고 합쳐서 보여주는 최적화 과정입니다. 사용되는 정보는 아래와 같으며 각 순서대로 설명드리겠습니다.

파티션 정보
시퀀스 넘버
통계 정보

데이터 파일에 삭제 파일 반영 시 가장 먼저 확인하는 정보는 파티션 정보입니다. 앞선 파트에서 데이터 파일과 삭제 파일들에 대응되는 매니페스트 파일에는 파티션 관련 정보가 위치한다는 설명을 드렸었는데요, 이 정보를 기반으로 동일 파티션의 데이터 파일과 삭제 파일들을 취합합니다. 이 과정 덕분에 파티션을 적절히 설정하는 것만으로도 데이터 파일에 반영할 삭제 파일들의 수가 줄어들어 테이블 조회 성능을 향상할 수 있게 되는 것입니다. 만일 파티션 정보가 없는 삭제 파일인 경우, 시스템 내부적으로는 globalDeletes로 명명되며 모든 데이터 파일과 비교를 수행하기에 성능에 부정적인 영향을 끼칩니다.

다음으로 확인하는 정보는 시퀀스 넘버입니다. 스냅샷 생성 시 1씩 증가되어 할당되며, 동일 스냅샷에서 생성된 모든 파일들이 공유하는 스냅샷들의 상대적 나이라고 설명을 드렸었습니다. 삭제 파일 종류에 따라 시퀀스 넘버 비교 조건이 상이한데, 간단히 정리하면 아래에 나열된 조건과 같습니다. 이때 해당 조건을 만족하는 삭제 파일들만 걸러냅니다.

동등 삭제 파일은 시퀀스 넘버가 더 작은 데이터 파일에만 반영됩니다. 즉, 과거 스냅샷에만 영향을 줍니다.
포지션 삭제 파일은 시퀀스 넘버가 같거나 작은 데이터 파일에 반영됩니다. 즉, 동일 스냅샷까지 영향을 줍니다.

예시 1을 확인해 보면 동등 삭제 파일의 id 칼럼에 데이터 파일에 있는 id 칼럼의 모든 값이 존재하는데, 동등 삭제 파일은 위에서 설명드린 것 처럼 상대적으로 더 이전에 생성된 데이터 파일들에만 반영이 되기 때문입니다. 즉, 동등 삭제 파일은 과거 스냅샷에만 영향을 주어 동등 칼럼 값이 동일한 과거 레코드들을 지우고, 테이블 조회 시 최신 데이터 파일에 있는 최신 레코드들을 보여줍니다. 포지션 삭제 파일은 동일 스냅샷 시점에 들어온 레코드에 대해 삭제 메시지가 들어오면, 포지션 삭제 파일로 들어가게 된다고 설명을 드렸었습니다. 예시 1의 id=24인 레코드가 동일 스냅샷 시점에 데이터가 들어오고 이후 삭제 메시지도 들어온 경우에 해당합니다. 이 경우 테이블 조회 시 id=24인 레코드가 조회되면 안 되기에, 포지션 삭제 파일은 동일 스냅샷까지 영향을 주도록 동작합니다.

마지막으로 동등 삭제 파일은 동등 칼럼의 통계 정보를 사용하여 한 단계 더 작업을 수행합니다. 사용되는 통계 정보는 동등 칼럼의 최소 및 최대 값 정보로, 정보를 사용하여 각 파일에 있는 동등 칼럼 값의 구간을 계산합니다. 이후 동등 칼럼 구간이 데이터 파일과 겹치는 삭제 파일만 최종적으로 반영합니다. 그림 14는 시퀀스 넘버는 1이고 파티션이 0번인 데이터 파일에 대한 예시이며, 데이터 파일에 반영이 필요한 동등 삭제 파일을 선정하는 전체 과정을 도식화하였습니다. 이러한 과정을 거쳐 아이스버그는 반영이 필요한 최소한의 삭제 파일들만 데이터 파일에 반영하고 그 결과를 사용자에게 보여줍니다.

그림 14. 동등 삭제 파일이 데이터 파일에 합쳐지는 기준 및 순서

컴팩션(Compaction)

컴팩션은 데이터 적재 과정에서 생성되는 데이터 파일과 삭제 파일들을 합쳐 새로운 데이터 파일을 생성하는 최적화 기능입니다. 앞서 설명드린 것처럼 플링크에서 아이스버그 테이블로 데이터를 적재할 경우, 플링크는 데이터 파일과 두 종류의 삭제 파일에 데이터를 저장합니다. 또한, 체크포인트마다 새로운 파일들이 생성되는데, CDC가 변화 데이터를 계속해서 수집하기 때문에, 시간이 지날수록 생성되는 파일의 수는 증가합니다. 그 결과 장기적으로는 테이블 조회 시간이 늘어날 수밖에 없는 한계를 지니고 있습니다.

물론 플링크 설정 파트에서 언급드렸던 바와 같이, 체크포인트 주기를 늘려 파일이 생성되는 주기를 줄일 수는 있습니다. 하지만 이는 테이블 조회 시간이 늘어나는 비율만 감소시킬 뿐, 파일이 많아지면서 조회 시간이 증가하는 근본적인 원인을 해결하지는 못합니다. 따라서 테이블 조회 시간을 줄이는 근본적인 방법은 컴팩션을 사용하는 방법뿐입니다.

그림 15. 컴팩션을 통해 새로운 데이터 파일이 생성되는 예시

아래의 그래프 1은 컴팩션을 수행하지 않으면 테이블 전체 데이터 조회 시간이 얼마나 증가하는지에 대한 추세를 보여주는 그래프입니다. 테스트에 사용된 테이블은 30억 개의 레코드를 가졌으며, 하루 평균 400만 개의 변화분이 발생합니다. 플링크의 체크포인트 주기는 10분으로, 10분마다 파일들이 생성됩니다. 테이블 조회에는 스파크를 사용했으며, 1,000개의 익스큐터와 각 익스큐터에 48G 메모리를 할당하였습니다. 그래프 1에서 볼 수 있듯 연동 직후(0일) 전체 테이블 조회에는 단순히 2.8분이 걸렸지만, 일주일이 지난 후에는 조회 시간이 60분까지 늘어난 것을 볼 수 있습니다. 마지막으로 테스트 7일 차에 컴팩션을 수행한 후 조회 시간을 측정했을 때, 컴팩션에는 39분, 전체 데이터 조회에는 1.7분이 걸렸습니다. 이 결과는 컴팩션이 아이스버그 테이블 운영 시, 테이블의 조희 시간을 짧게 유지하기 위해 사실상 필수적인 기능인 것을 시사하고 있습니다.

그래프 1. 컴팩션을 수행하지 않을 경우 시간에 따른 소싱 시간의 변화

다음으로 컴팩션시 한 가지 주의 깊게 보셔야 할 경우를 공유드리겠습니다. 이미 컴팩션된 데이터 파일은 특정 조건에서 후속 컴팩션에서 제외될 수 있습니다. 관련 설정으로 min-file-size-bytes와 max-file-size-bytes가 존재합니다. 이 설정들로 인해 컴팩션된 데이터 파일의 크기가 사용자가 설정한 target-file-size-bytes 설정 값의 0.75배에서 1.8배 내에 존재하면, 해당 데이터 파일들은 후속 컴팩션 수행 시 컴팩션에서 제외됩니다. 저희 팀에서 발생한 상황을 예시로 설명드리면, target-file-size-bytes를 250MB로 설정한 경우 187.5 ~ 450MB 사이로 컴팩션된 데이터 파일들은 이후 컴팩션 작업에서 제외되었습니다. 이로 인해 187.5 ~ 450MB 사이의 크기로 컴팩션된 데이터 파일 이후 시점에 생성된 삭제 파일들은, 앞선 데이터 파일들이 존재하기에 항상 참조 상태가 됩니다. 따라서 후술 될 스냅샷 만료와 고아 파일 제거 기능을 수행해도 참조 상태이기에 지워지지 않아, 그림 16 처럼 작은 크기의 삭제 파일들이 4만 개가 넘게 계속 늘어나는 현상이 발생하였습니다. 이에 저희 팀은 컴팩션에 rewrite-all 설정을 true로 설정하여, 항상 모든 데이터 파일들이 컴팩션 후보에 들어가도록 하여 해당 문제를 해결하였습니다.

그림 16. rewrite-all: true 설정 추가 전후 동등 삭제 파일과 포지션 삭제 파일 개수 변화

마지막으로 독자분들께 컴팩션 수행 시간이 오래 걸릴지 판단하기 위해 저희가 확인하는 지표를 한 가지 공유드리겠습니다. 컴팩션은 테이블 조회 속도를 향상하기 위해 미리 파일을 합치는 과정이기 때문에, 기본적으로 테이블 소싱과 유사한 작업을 수행합니다. 따라서 컴팩션 작업에는 많은 시간이 소요됩니다.

현시점에서 컴팩션 수행 시간에 영향을 미치는 모든 요인과 상관관계를 알 수는 없지만, 저희는 다양한 테이블에 컴팩션을 수행해 보며 테이블에 들어오는 변화분의 타입에 주목하고 있습니다. 경험적으로는 업데이트 쿼리의 비율이 높은 테이블이, 동일한 수의 변화 분이 들어와도 컴팩션 수행 시 더 많은 시간이 소요된다는 점을 강조드릴 수 있겠습니다. 다만 테이블 레코드 수와 변화분 수가 비슷하면서, 쿼리 타입의 비율이 크게 다른 테이블을 찾는데 어려움이 있어 이 경우를 테스트해보지는 못했습니다.

대신 레코드 수는 다르지만 변화분의 수가 비슷하고 쿼리 타입에 비율이 다른 테이블들에 대해 테스트를 진행했습니다. 테스트 테이블은 각각 30억 개와 9천만 개의 레코드를 가지며, 두 테이블은 하루 평균 400만 개의 변화분이 발생합니다. 두 테이블에 대해 스파크를 통해 컴팩션 수행에 걸리는 시간을 측정했으며, 500개의 익스큐터를 설정한 후, 각 익스큐터 당 16G 메모리를 할당하였습니다. 아래의 테이블 1에 정리된 내용에서 두 테이블의 변화분 비율과 컴팩션 수행 시간을 확인할 수 있습니다.

	30억 레코드 테이블	9천만 레코드 테이블
INSERT 쿼리 개수 및 비율	1.7M (42%)	0.09M (2.2%)
UPDATE 쿼리 개수 및 비율	2.1M (53%)	3.88M (96.3%)
DELETE 쿼리 개수 및 비율	0.2M (5%)	0.06M (1.5%)
하루치 변화분 평균 컴팩션 시간	평균 5분	평균 9.7분

<테이블 1. 테이블별 변화분의 쿼리 타입과 컴팩션 시간 비교>

테이블 1에서 볼 수 있듯이, 레코드 수는 30배 이상 차이나지만 컴팩션에는 레코드가 9천만 개인 테이블이 더 많은 수행 시간이 소요되었습니다. 이 결과에 대해, 저희는 30억 레코드 테이블은 변화분의 약 53%가 업데이트 쿼리이고, 9천만 레코드를 가진 테이블은 약 96%가 업데이트 쿼리인 부분에 초점을 맞추어 분석하였습니다. 저희가 분석한 내용이 완벽하거나 정답은 아닐 수 있겠지만, 아이스버그의 동작 과정을 고려하여 분석한 내용을 공유 드려보면 아래와 같습니다.

먼저 인서트 쿼리와 업데이트 쿼리는 둘 다 데이터 파일과 삭제 파일에 데이터를 저장합니다. 다만 차이가 나는 부분은, 인서트 쿼리는 새로 할당된 동등 칼럼의 값을 갖는 반면, 업데이트 쿼리는 기존 레코드를 갱신하기 때문에 이미 존재하는 동등 칼럼 값을 가집니다. 따라서 업데이트 쿼리의 비율이 높으면, 포지션 삭제 파일에 더 많은 데이터가 저장됩니다. 하지만, 포지션 삭제 파일은 스캔 플래닝 파트에서 설명드린 것처럼 동등 칼럼 구간을 이용한 최적화를 사용하지 못합니다. 결과적으로 더 많은 데이터 파일들과 비교되는 포지션 삭제 파일에 더 많은 데이터가 들어가기에, 업데이트 쿼리의 비율이 높을수록 수행시간에 부정적인 영향을 미친다고 저희는 해석하고 있습니다.

스냅샷 만료와 고아 파일 제거(Expire snapshots & Delete orphan files)

다음으로 고려해야 할 최적화 기능은 스냅샷 만료와 고아 파일 제거입니다. 아이스버그의 메타데이터 파일에는 모든 스냅샷들의 정보가 남는다고 설명드렸습니다. 스냅샷 만료는 사용자가 지정한 시점까지의 스냅샷만을 유지하며, 그 보다 오래된 스냅샷을 삭제하는 기능입니다. 즉, 메타데이터 파일에 있는 특정 시점보다 오래된 스냅샷들의 정보를 지운다고 보시면 됩니다.

하지만 스냅샷 만료 자체는 참조만 제거하며 실제로 관련 파일들을 지우지는 않습니다. 실제 참조가 사라진 파일들을 지우는 기능은 고아 파일 제거 기능입니다. 해당 기능은 참조가 끊긴 모든 파일들을 지우는 것은 아니며, 스냅샷 만료와 동일하게 사용자가 지정한 시점보다 오래된 파일들 중 참조가 없는 파일들을 모두 삭제합니다. 만약 두 기능 기능에 동일한 시점을 지정하면, 해당 시점까지의 스냅샷을 유지하며 그보다 오래된 스냅샷 및 파일들을 삭제하게 됩니다.

그림 17. 스냅샷 만료와 고아 파일 제거 예시

다만, 타임 트레블 기능의 조회 가능 시점에 영향을 주기 때문에, 타임 트레블 기능을 사용해야 한다면 사용 목적에 맞는 적절한 시점을 설정하셔야 합니다. 스냅샷 만료를 수행하게 되면 스냅샷 만료에 사용된 시점보다 과거의 테이블 상태는 조회할 수 없기 때문입니다. 따라서 타임 트레블 기능을 통해 최대 6개월 전까지의 테이블에 대해 조회할 것으로 예상된다면, 스냅샷 만료 및 고아 파일 제거의 시점을 6개월 또는 그보다 더 이전 시점으로 설정해야 합니다.

저희 팀에서는 타임 트레블 기능을 사용하지 않기에 파일들에 대한 유지 기간 기준을 다르게 책정하였습니다. 보안 이슈가 있는 민감 데이터는 해싱이나 마스킹을 수행해도, 주기적으로 데이터를 지워달라는 요청이 자주 있었기 때문에, 과거 상태 조회가 불가하도록 짧은 주기로 스냅샷 만료와 고아 파일 제거를 수행했습니다. 일반적인 데이터의 경우 보안에 대한 이슈는 없었지만, 객체 저장소로 하둡 파일 시스템을 사용 중이고 하둡의 블록 사이즈(Block size) 보다 작은 파일들이 계속 생성되며 하둡의 IO에 부정적인 영향을 끼치고 있었습니다. 따라서 팀의 상황에 맞게 일반 데이터에 대해서도 짧은 주기로 스냅샷 만료와 고아 파일 제거를 수행하도록 설정하였습니다.

본 글에서 설명드린 최적화 기능 외에도 포지션 삭제 파일들이나 매니페스트들을 최적화 해주는 rewrite_position_delete_files, rewrite_manifests등의 최적화 기능도 존재하기에, 사용 환경에 맞춰 적절한 최적화 기능들을 선정하여 수행하시는 것을 권장드립니다.

샤딩 테이블의 단일 아이스버그 테이블로 통합 운영

이번 파트에서는 샤딩(Sharding) 된 테이블들을 하나의 아이스버그 테이블로 통합하여 운영이 가능할지 테스트한 과정을 공유드리겠습니다. 테스트에서는 샤딩된 테이블들을 물리적으로 하나의 아이스버그 테이블로 통합할 수 있는지, 지표 추출 시 성능 개선과 운영상의 이점이 있는지, 그리고 안정적 운영이 가능한지 여부를 검토하였습니다. 먼저 하나의 테이블로 통합 시 발생하는 이점을 정리하면 아래와 같습니다.

지표 추출 시 복수의 테이블 소싱 작업을 단일 테이블 소싱으로 간소화
소싱된 여러 테이블들을 합치는 유니언(Union) 과정의 생략
컴팩션, 스냅샷 만료 및 고아 파일 제거를 단일 테이블에 대한 수행으로 간소화

구현 방향

해당 기능 구현을 위해 고려한 방향은 각 아이스버그 테이블에 데이터를 저장하고 조회하는 방식이 물리적 및 논리적으로 유사한 구조를 유지하는 것입니다. 이는 하나의 아이스버그 테이블로 연동할 경우 발생할 수 있는 추가적인 부하를 줄이기 위함입니다.

이를 위해 샤딩 테이블들을 각 파티션으로 분리하여 저장합니다. 파티션으로 분리하는 이유는 스캔 플래닝(Scan planning) 파트에서 설명한 바와 같이, 테이블 조회 시 데이터 파일과 삭제 파일들을 합치는 작업이 파티션별로 수행되기 때문입니다. 따라서 샤딩 테이블이 각각의 파티션에만 적재되도록 하면, 샤딩 테이블 별 데이터 및 변화분들이 각각의 파티션에 물리적으로 분리되어 저장되고, 결과적으로 각 파티션은 하나의 샤딩 테이블의 데이터와 변화분만을 처리하게 됩니다. 아래 그림 18은 이러한 파티션 구조의 예시입니다.

그림 18. 파티션 계층 구조 예시

실제로 파티션을 어떻게 설정하였는지 설명드리면 아래와 같습니다. 먼저 identity 파티션을 통해 샤딩 테이블들을 각각의 파티션으로 분리합니다. 이후 각 파티션 내에서는 다시 bucket 파티션을 수행하여, 하나의 테이블을 아이스버그 테이블로 연동할 때와 동일한 구조를 가지도록 합니다. 이러한 이유로, identity 파티션과 bucket 파티션을 순차적으로 설정하였으며, 이에 따른 세부 작업 절차를 정리하면 아래와 같습니다.

플링크에서 아이스버그 테이블을 생성할 때 샤딩 번호를 저장하는 shard_column 칼럼을 추가해 생성
shard_column 칼럼을 동등 칼럼에 추가
shard_column 칼럼에 identity 파티션을, 기본 키(예: id)에 bucket 파티션을 순차적으로 설정
e.g. …/identity={shard_column}/id_bucket=0/…, …/identity={shard_column}/id_bucket=1/…
플링크에서 아이스버그 테이블로 데이터 적재 시, 전송되는 메시지에 샤딩 번호를 동적으로 추가

그림 18을 기준으로 설명드리면 identity=? 형태로 존재하는 번호들은 샤딩 테이블 별로 할당된 샤딩 번호를 의미합니다. identity 파티션 내에서는 bucket 파티션을 통해 ../identity=0/id_bucket=4와 같은 구조로 데이터가 저장됩니다. 정리해 보면, 각 샤딩 테이블의 데이터는 identity 파티션으로 구분되어 테이블 조회 시 샤딩 테이블들의 데이터가 섞여 발생할 수 있는 추가적인 데이터 합침 등의 연산을 방지합니다. 또한, 단일 아이스버그 테이블로 연동할 때와 동일한 bucket 파티션을 하위에 추가하여 효율적인 테이블 조회 성능을 확보합니다. 각각의 샤딩 테이블이 플링크 잡을 통해 아이스버그 테이블로 적재되는 형상은 그림 19과 그림 20를 참고해 주시기 바랍니다.

그림 19. 0번 테이블의 identity=0 파티션으로 데이터 인입 예시

그림 20. 31번 테이블의 identity=31 파티션으로 데이터 인입 예시

테스트

총 27억 개의 레코드를 가진 32개의 테이블을 하나의 아이스버그 테이블에 적재하였습니다. 이후 스파크를 통해 컴팩션과 테이블 전체 데이터 소싱을 수행하고, 소싱에 걸린 시간을 측정하였습니다. 테스트 결과, 평균 270초 가 소요되었으며, 비교를 위해 9천만 레코드를 가진 아이스버그 테이블 소싱했을 때는 평균 20초가 소요되었습니다. 20초가 소요되는 작업을 32개 수행하는 것과 하나의 테이블로 통합했을 때 소요된 270초를 비교해 보면, 성능면에서 통합 방식이 유의미하다는 것을 알 수 있었습니다.

한계점

그러나 이 기능은 안정적 운영이 불가할 것으로 판단되어 실서비스 환경에는 적용하지 못했습니다. 아이스버그 테이블은 주기적으로 커밋이 수행되는데, 복수의 플링크 잡이 하나의 아이스버그 테이블에 커밋을 수행하는 형상이 되기 때문입니다.

이러한 형상이 안정성이 떨어지는 원인이 되는 이유를 설명드리겠습니다. 커밋은 베이스 메타데이터 파일을 기준으로 수행되며, 커밋이 완료되면 베이스 메타데이터 파일을 업데이트합니다. 여기서 베이스 메타데이터 파일은 현재 메타데이터 포인터가 가리키는 파일입니다. 만약 동시에 여러 커밋이 하나의 테이블에 수행되면, 커밋을 제일 먼저 성공한 플링크 잡을 제외한 나머지 플링크 잡은, 각각이 바라보고 있던 베이스 메타데이터 파일이 변경되어 동작 과정에서 아래의 에러 메시지 2가 발생합니다.

org.apache.iceberg.exceptions.CommitFailedException: Cannot commit: Base metadata location  'hdfs://../../namespace/table/metadata/XXX-A.metadata.json' is not same as the current table metadata location 'hdfs://../../namespace/table/metadata/XXX-B.metadata.json' for namespace.source_table

<에러 메시지 2. 커밋 충돌로 발생 가능한 에러 메시지>

아래의 그림 21을 예시로 들어 계속해서 설명을 드리겠습니다.

먼저 0번 플링크 잡과 31번 플링크 잡이 베이스 메타데이터 파일 XXX-A.metadata.json을 기준으로 커밋을 수행합니다. 0번 플링크 잡이 먼저 커밋을 완료하여, 현재 메타데이터 포인터가 가리키는 메타데이터 파일은 새로 생성된 메타데이터 파일인 XXX-B.metadata.json을 가리킵니다. 이 경우 31번 플링크 잡은 커밋을 완료하기 전 베이스 메타데이터 파일을 다시 확인하고, 참고 중인 베이스 메타데이터 파일이 달라졌기에 에러가 발생합니다.

그림 21. 커밋 충돌 발생 상황

물론 해당 에러가 발생해도 플링크 잡이 바로 중지되지는 않습니다. 이러한 에러와 관련한 아이스버그 설정으로 COMMIT_NUM_RETRIES가 이미 존재하며, 커밋에 실패하더라도 최대 재시도 횟수 내에서만 커밋에 성공하면 플링크 잡이 중지되지 않습니다. 따라서, 해당 설정 값을 늘려 안정성을 어느 정도 확보가 가능합니다. 내부적으로 진행한 테스트에서도 재시도 횟수를 60으로 변경하여 일주일 정도 지켜보았을 때, 32개의 플링크 잡들 중 일부 플링크 잡에서 커밋 실패가 종종 발생하였지만, 플링크 잡에 예외가 발생하며 중지되는 경우는 없었습니다.

하지만 아이스버그로 연동이 필요한 테이블 중에는 100개가 넘는 테이블로 샤딩이 된 테이블이 존재하고, 이 정도 규모에서는 재시도 횟수 설정을 늘리는 것 만으로는 안정성을 확보하기가 힘들다고 판단하였습니다. 특히 실서비스 환경은 성능만큼이나 안정성 또한 중요하기에, 자칫 플링크 잡이 중지될 수 있는 기능을 적용할 수 없었기 때문입니다. 대안으로 샤딩 개수가 적은 테이블에 적용하는 것을 한 번 더 논의했으나, 그 과정에서 모든 샤딩 테이블들에 대해 동일한 정책을 가져가는 방향으로 의견이 수렴하며 최종적으로 이 기능은 적용하지 않게 되었습니다.

마무리하며

긴 글의 끝까지 함께 해주신 독자 분들에게 진심으로 감사 인사를 드리며, 이번 글을 개인적 소회와 함께 마무리하고자 합니다.

하나씩 회고를 진행하면, 먼저 플링크를 통해 아이스버그에 데이터를 적재하는 과정이 쉽지 않았다고 생각됩니다. 팀에서도 데이터 레이크하우스 기술을 처음 도입한 사례였고, 플링크를 통해 아이스버그에 CDC를 적용하는 경우가 국내에서는 아직 드문 것 같습니다. 해외에는 일부 사례가 있으나, 그 사례들도 당시 저희 팀의 요구 사항과는 조금씩 차이가 있었습니다. 예를 들면, 플링크의 데이터스트림 API가 아닌 플링크 SQL을 사용하거나, 하둡 파일 시스템이 아닌 S3를 객체 저장소로 활용하는 경우가 많았습니다.

또한, 초기에는 아이스버그의 동작 과정에 대해 전혀 모르는 상태였기 때문에 데이터가 어떻게 적재되고 파일에는 어떠한 정보들이 있는지 파악할 필요가 있었습니다. 이를 위해 시스템과 라이브러리의 코드를 심도 있게 분석하고, 테스트를 수행해 보며 생성된 파일들을 하나하나 뜯어보아야 했습니다. 이러한 과정들이 쉽지는 않았으나, 그 과정을 잘 정리한 덕분에 독자 분들에게 보다 심도 있는 내용을 공유드릴 수 있게 되어 의미 있고 보람찬 경험이었다고 생각합니다.

마지막으로 이번 글이 플링크 및 아이스버그를 통한 CDC 수행을 고려하시는 독자 분들에게는 하나의 이정표로, 단순 호기심을 가지시고 읽으신 독자분들에게는 그 호기심을 만족시켜 주는 글로써 남기를 바랍니다. 관련 작업 같이 지원하고 진행해 준 동료분들 archer.kang, dawn.choi, huan.15, levie.yumtam, max.iam, stephen.c과 wayne.pk에게 감사 인사를 드리며 글을 마치겠습니다.

감사합니다.

참고 문서

Written by Louis.sml
Edited by June.6

Apache Iceberg와 Flink CDC 심층 탐구