Apache Iceberg 사용 사례 (LINE Data Platform에서 Apache Iceberg 도입)

https://www.youtube.com/watch?v=7y9gNwqLNtU

위 동영상 정리함

 hive -> iceberg로 변경한 사례

 

Line 데이터 플랫폼의 기존 방식

데이터 플랫폼의 특징

• 데이터의 크기/종류/활용 규모가 크다
• 머신 수: 5000 대+
• 데이터 크기: 290 PB+
• 하이브 테이블 수: 4만개 (모두 external table)
• log ingestion rate: 초당 17.5M+
• 하루에 구동되는 job 수: 15만+
• 플랫폼 사용자 수: 700+

 

Query Processing

SQL 기반, spark, hive, trino, flink를 분산 sql 쿼리 엔진으로 사용

데이터파일이 어디에 있는지, 어떻게 읽고 쓸수있는지에 대한 정보 필요

쿼리에 대한 테이블을 어떻게 관리할지는 테이블 포맷에 의해 정의됨

 

* Table Format이란:

어떤 데이터 셋을 하나의 테이블로 표현할 수 있도록 파일 관리 방식 규정

쿼리 실행 시에 주로 아래의 두 가지의 역할을 수행한다.

1. 어느 파일이 테이블 데이터를 구성하는지 알려줌
2. 그 파일을 어떻게 읽고 쓸 수 있는지 알려줌

→ 파티셔닝/스키마 정의 등 메타 데이터의 관리도 필요함

 

메타 데이터 관리: Hive Table Format (de-facto standard)

데이터 파일을 파일 시스템의 디렉토리에 의해 관리하며, 테이블 관리에 사용되는 파티션/스키마 등의 메타데이터 변경 및 조회는 metastore 서브를 이용해 이루어진다.

hive metastore의 백엔드에서는 메타데이터가 metastore db라고 하는 rdbms에 저장된다. (Thrift API 사용)

 

 

Log 관련 End-to-End 파이프라인

• Source -> Kafka -> Flink -> RAW Table -> Tez on YARN -> ORC table -> User

columnar 파일 포맷 사용을 위해 Raw Table을 ORC Table로 변환하는 작업 수행

 

 

기존 방식의 문제점

Hive metastore와 DB의 Capacity

모든 테이블의 메타스토어가 Hive Metastore에 저장되어 있어 Metastore DB 에 대한 부하 문제가 있음.

DB의 QPS 는 5천개로, CPU 사용률이 50~60% 이상이었고, Abnormal한 사용량(90% 이상)도 종종 관찰됨.

대량의 파티션 조회로 Memory Pressure, OOM으로 인한 Metastore 다운 문제가 있음.

조회의 퍼포먼스가 제한되지만, Scale-out 하기가 어려움.

 

정리

1. bottleneck

2. inefficient data access

3. less opportunites for optimization

 

Log 파이프라인 End-to-End의 high latency

1. RAW Table Truncate로 인한 지연

small files가 대량으로 존재하게 되면 Namenode의 메모리 사용량이 증가한다.

이러한 문제를 해결하기 위해 truncate를 지원하는 Flink의 BucketingSink를 사용하여 가능한 한 파일을 생성하지 않도록 주기적으로 Flush 함

이 Flush 주기때문에 최대 1시간의 Latency 발생

 

2. ORC 테이블 변환

테이블 수가 많아 변환에 시간이 걸림(약 50분)

 

파이프라인의 Robustness

1. 의존하고 있는 componets가 너무 많음

• Flink, Watcher, HDFS, Hive Metastore, Hiveserver2, YARN

모든 components가 정상적으로 작동하지 않으면 파이프라인이 중지되기 때문에

장애 발생 확률이 높아지고 장애 발생 시 원인 파악 난이도도 높아짐.

 

2. Table 운영 문제

Table Partition의 조작은 자동화 되어 있지만 유저로부터 특수한 요청이 있을 때에는 수동으로 관리해야할 때가 있다.

이로 인해 두 종류의 Hive External Table 이 필요해진다.

 

3. 고부하 문제

partition/file마다 너무 많음 metadata가 있어 아래와 같은 문제를 야기한다.

• hive metastore에 heavy partition scan
• namenode에 heavy directory scan

 

Apache Iceberg

table 생성 하면 아래와 같은 레이아웃이 생성됨

• ./data 
• ./metadata

Key Concept: Snapshot

iceberg는 테이블 파일 상태를 스냅샷으로 관리함으로써 데이터를 추적하며, 스냅샷을 통해 테이블을 구성하는 모든 파일의 참조 정보 저장 파일 Path, 데이터의 통계정보 를 얻을 수 있다..

(wirte & commit 이 일어나면 새로운 스냅샷을 생성함)

 

hive와 결정적으로 다른 부분으로 iceberg는 테이블 상태가 파일(snapshot, manifast list file, manifest file)에 의해 추적이 되기 때문에 아래와 같은 이점을 얻을 수 있다.

• hive metastore 에 대한 의존을 없앨 수 있음
• hdfs 등 scale out 하기 쉬운 파일시스템을 메타데이터의 보관 장소로 이용할 수 있음

Iceberg가 파일을 추적하는 방식 (출처: https://www.dremio.com/resources/guides/apache-iceberg-an-architectural-look-under-the-covers/)

 

Iceberg는 다음과 같은 과정을 통해 테이블의 파일을 추적한다.

1. 현재 Snapshot을 참조하여 지정된 Manifest list 파일을 찾는다. (이미지에서 s0, s1)
2. Manifest List 파일은 1) Manifest File의 경로, 2) 그 Manifest File로 관리되는 데이터의 Partition Value의 상한치/하한치를 가져온다.
   * 이때 쿼리에 파티션 필터가 있는 경우 Range를 사용하여 실제로 읽는 Manifest File을 필터링할 수 있다.
3. Manifest File을 읽어서 Data path file을 취득한다.
   Menifest File에서는 파일별/컬럼별 Status를 저장하고 있어 쿼리 시에 실제로 읽는 파일을 더 줄일 수 있다.

Hive와 Iceberg의 차이점과 이점은 아래와 같이 정리될 수 있다.

	hive	apache iceberg	iceberg의 이점
Metadata stored at	hive metastore	files	scalability
Finer partitioning granularity	limited	unblokced	efficiency
stats support	per partition	per file	performance

 

 

Iceberg 도입을 통한 개선 내용

Hive Metastore 의존 삭제를 통한 부하 문제 해결

iceberg는 metastore을 db로 관리하지 않고 파일을 이용해서 데이터를 추적한다.

그렇기때문에 HDFS와 같은 Scale-out이 쉬운 파일 시스템을 메타데이터 저장소로 사용함으로서 Capablity 를 확보할 수 있었음.

 

로그 파이프라인 개선

• Source -> Kafka -> Flink -> Iceberg Table -> User
                                                      |
                                            spark on YARN

Flink에서는 Iceberg file을 다이렉트로 write하고, (parquet/ORC 포맷)

Spark 를 이용해 5분에 한 번씩 Flush하도록 변경함

Spark에서는 아래의 작업을 수행한다.

• merge Small File
• Expire Snapshots 
• Rewrite manifests
• Remove orphans
• delete expired records

이로 인해 얻는 이점은 아래와 같다.

1. low latency: 2시간 -> 5분으로 단축
2. simple/scalable한 아키텍처로 변경
   - yarn, hiveserver2의 디펜던시가 사라짐
   - Hidden partition 기능으로 파티션 관리가 용이해짐 (external table에서 사용해야 하는 partition operation은 불필요해짐)
   - Namenode/Hive metastore에 대한 부담이 경감됨
3. Schema Evolution
   Iceberg에서는 컬럼을 column name, position으로 구분하지 않음
   각 필드마다 유니크한 ID가 부여되어 있음
   이로 인해 필드 추가/ Rename/ 삭제/ 삽입/ 이동 등 모든 변경이 Field ID 에 기반하여 쉽게 스키마를 변경할 수 있다.