[광고 플랫폼 Lake House 만들기] 7. 자동화 & BI 파이프라인

우리가 지금까지 만들어온 파이프라인은 위와 같다.

이를 운영하기 위해..

이전 블로그 포스팅에서 이야기한 maintenance 묶음(compact + rewrite-position-deletes+expire)을 자동화 해보자..!

 

Airflow 자동화

우리는 Airflow를 활용할 것인데, 왜 airflow인지 먼저 생각해보도록 하자. 

배치 주기로 돌려야 하는 작업들이 여러가지고, 작업간 의존성이 있는데 이게 시각적으로 잘 표현되게 해준다.
추가로, task가 실패했을때 운영 가시성이 좋다. (어디서 잘못되었는지 한눈에 보임)
cron에서는 의존성/재시도/관찰과 같은 부분이 떨어짐.

 

지금 우리가 자동화해야 할 작업들에 어떤 것들이 있고, 주기/의존성에 대해 알아보자.

작업	주기	의존성
raw -> processed MERGE	매 1시간	streaming 적재 후
processed -> summary MERGE	매 1시간	silver MERGE 완료 후
Compaction	매일	새벽 한가한 시간대
Expire Snapshots	매일	Compaction 직후
Orphan File Cleanup	주 1회	일요일

 

위와 같은 작업들을 Airflow로 자동화함으로써 우리는 스케줄링 + 의존성 + 재시도/알림 + 관찰 가능성을 얻을 수 있다.

 

DAG 설계

Dag를 설계 함에 있어 아래 4가지 의사결정을 따르도록 하자.

• 레이어별 DAG 분리 (schedule/maintence 분리)
• INSERT와 MERGE 별도 task (두 분기를 Graph view에서 분리)
• BashOperator + docker exec  (기존 컨테이너에 위임)
• Merge-on-Read 테이블 (write amplification 회피) -> 주기적 compaction + delete rewrite

실제 run_demo_cycle 스크립트(트리거)를 실행한다면 그 순서는 아래와 같을 것이다.

1. spark-iceberg / kafka / airflow 상태 + AWS 자격 검증
2. 샘플 CSV 7일치 생성
3. kafka 토픽 3개 생성 + producer 로 발행
4. 3개 streaming spark-submit -> S3 raw zone 적재
5. ad_lakehouse_sliver manual trigger + 완료 대기
6. ad_lakehouse_gold manual trigger + 완료 대기
7. spark - sql로 row 수 + 최근 snapshot 출력

 

이중 ad_lakehouse_silver DAG 내부의 Task 흐름을 살펴보자.

1. check_raw_zones_have_data : raw 있는지 확인 후, 없으면 멈춤.
2. insert_new_events : 신규 이벤트 적재
3. update_late_arrivals : 지각 도착 분 머지
4. compact_processed_events : small files 정리
5. rewrite_position_deletes : 위에서 Merge로 생긴 delete 파일 삭제
6. expire_snapshots : 스냅샷 만료
7. remove_orphan_files_weekly : 고아 파일 청소

위와 같이  스케쥴/비용등을 고려하는것도 중요하지만,

장애 시 대응을 위해서 airflow log에 뭘 남길지도 잘 생각해야 한다.

 

예를 들어, check_raw_zones 에서는 처리 대상 파티션/날짜 범위, 작업 테이블의 스냅샷 ID 등을 기록해둔다면

어느 시점에서 데이터가 깨졌는지 그리고 어떤 스냅샷으로 재현해야 하는지 알 수 있을 것이다.

또한 각 단계별 결과 수치를 로그로 끌어낸다면 문제 지점을 파악할 수 있다.

만약 컴팩션 전후로 데이터 파일 수가 같다면 컴팩션이 정상적으로 이루어지지 않았음을 의미할 것이다. 

추가적으로 실행해보며 아래 내용들을 검토해보며 확인해보자.

• Airflow Graph View
• Insert/Update 단계 snapshot
• http://localhost:4040 실제 Spark 잡 동작 확인
• Athena Console  같은 테이블 SELECT 즉시 동작 (Glue 카탈로그 공유)

BI

데이터 시각화 방법과 같은 경우는 Athnea -> QuickSight 로 당초 계획을 했었으나,

비용문제로 인해 Superset같은 오픈소스로 대체할 계획이다.

 

Athena에서 먼저 검증 후, 데이터셋을 연결해서 시각화한다. 

이 부분에 관해서는 다음 포스팅에서 개인 프로젝트 시각화 한것을 대상으로 더 다루도록 하겠다.
 
 
 추가적으로.. 우리가 다룬 오픈 테이블 포맷의 최근 향방에 대해 알아보고 이번 글을 마무리 하겠다.

Open Table Format의 미래

 XTable과 UniForm과 같은 경우에는 포맷끼리 번역해주는 도구로,

delta, hudi, iceberg와 같은 여러 포맷에서의 메타데이터를 번역하는 추상화 도구이다.

 

Paimon과 같으느 경우는 Apache에서 진행중인 프로젝트로, 실시간 환경에 적합한 오픈 테이블 포맷이다.

Flink 기반 스트리밍 우선 포맷이라 CDC와 같은 환경에 특화되어 있다.

 

DuckLake

DuckDB팀이 2025년 5월 발표한 것으로, 메타데이터를 파일이 아닌 RDBMS 트랜잭션으로 관리한다.
이는 Orphan File을 DB 트랜잭션으로 해소하고, DB가 알아서 관리하기에 컴팩션이 필요하지 않다. 

또한, select * from metadata가 가능하기에 메타데이터 디버깅에 용이하다는 장점이 있다.
 
데이터 포맷은 현재 Iceberg 중심으로 표준이 정해지는 추세인데,

카탈로그 레이어는 아직 표준이 없기에 이 부분이 다음 종속 지점이라고 보고 있다. (Catalog Wars)

 

현재 주요 카탈로그를 비교해보면 아래와 같다.

카 탈로그	만든 곳	특징
Unity Catalog	Databricks	Delta + Iceberg 통합 관리. Databricks 생태계 통합 전략
Apache Polaris	Snowflake -> Apache 기증	Iceberg REST Catalog 표준. Unity 대안 포지셔닝
Nessie	Dremio	Git 처럼 데이터 브랜치/머지 지원
Gravitino	DataStrato	여러 카탈로그를 통합하는 메타 카탈로그
AWS Glue	AWS	서버리스, AWS 종속

 
하나의 회사에 속하는 기술을 써볼려고하기보다는 왜 이런 기술들이 나오게 되었는지 흐름을 이해하기 위해 노력하자....
 
5-Layer 생태계 

1. 쿼리 엔진 (SPARK, TRINO, FLINK, DUCKDB, DREMOIO)
2. 카탈로그 (UNITY, POLARIS, NESSIE, GLUE, GRAVITINO)
3. 테이블 포맷 (ICEBERG, DELTA , DUCKLAKE)
4. 파일 포맷 (APACHE PARQUET)
5. 스토리지 (S3, GCS, ADLS)

위 내용들 중 Trino는 코드 구현부도 구경을 해보면 재밌는게 많다..

Java로 쓰여져 있지만 지원 함수들중 java primitive을 통해 빠르게 동작한다.
이외에도 운영 환경에서 Trino가 지원하지 않는 함수를 내가 운영하는 환경에서 써야하는 순가이 올 수 있기에

user define 함수에 대해서도 사용할 줄 알아야 한다..
 
Iceberg/Paimon은 관심있게 지켜보고 공부하고, 쿼리엔진에 대해도 공부해볼 계획이다..!

---

전체 코드 리뷰  및 리팩토링 포인트

코드 개선 가능 point

sturct type과 같은 경우도 별도로 저장해두는게 더 좋을듯.

spark operator를 사용하면 운영할때 더 까다로운 부분이 있음. -> 도커 에서 airflow가 spark summit 실행가능하도록 해줬음.

ddl도 코드로 관리할 수 있도록 만드는게 더 좋다.

컨테이너 띄우는 내용을 스크립트로 하나 만들어두는걸 추천. 

 

주의 사항 및 추가적으로 고민해볼만한 부분

compact/delete에 대한 기준 및 코드로 어떻게 정리할것인지 고민해보자. 별도의 데이터 클래스로 만든다거나...

나중에 스케일이 커졌을때를 대비해서 각각의 컨테이너를 어떻게 분리할것인가? 

인프라 구성시 너무 강결합되지 않도록 주의할것. (하나가 죽었을때 우루루 죽으면 안되니까)

compaction시 minfiles설정하는 부분도 하드코딩이 아니라 분리해두는게 좋음. (값 자체를 별도의 변수로 관리해야함)

더 나아가서 management 전략별로 변수들을 할당할 수 있도록 설정. 

 

이모저모

DDL 을 어디서 실행할 것인가??

DDL을 spark에서 실행하는 경우와 athena에서 실행하는 경우의 차이점. Iceberg는 athena에서 했을때 잘 안먹는 경우가 있어서, management나 테이블 만드는 작업이 의도대로 안돌아가는 경우가 있음. trino에서도 매니지먼트 작업을 지원하긴 하지만, 안되는 경우가 또 있음.. 철학상 어울리지도 않음 (compaction 지원 x). 아직 구현이 잘 되어있는 상황이 아닌것 같음. -> 운영시에는 DDL을 스파크에서 보통 함. 아테나도 분산 쿼리 엔진이긴 한데 페스토 쿼리해서 한거고, trino랑 비슷한점도 잇음.- iceberg는 spark 환경을 가정하고 만들어진거라서 spark랑 합이 괜찮음.  Trino는 쿼리가 빨리 돌아가게 만드는게 중요(stateless 지향)
최근 Trino <-> Iceberg관련 이슈 찾아보아도 재미있을듯?

 

Trino 학습에 관하여... 
아테나에서 쿼리를 날려본것만으로는 Trino에 대해 경험을 어필하기 어렵다.
운영을 하다 보면 Trino 내부 구현에 대해 생각보다 잘 알아야 하는 부분이 있다.
요즘 정말 메인 쿼리 엔진이기에, definitive guide 아래 책을 보며 학습하는걸 추천. + Trino project 구경.
Trino 기여하는 커뮤니티도 폐쇄적이라 평소에 눈팅을 자주하는걸 추천함.
athena를 극한까지 쓰다보면 aws에서 설정을 바꿔달라는 요청을 보내야하는 시점이 있는데,
이때 Trino내부 구현에 대해 고민해볼 기회가 있을것이다.
Query plan을 봤을때 trino와 athena가 서로 약간 다르게 나오는 케이스들이 있긴 함.

인프라측면에서 DE가 신경 쓸만한 부분에 뭐가 있을까 ???

클라우드이든 온프레임이든, 분산환경처리에서 노드 스펙을 어떻게 가져갈것인지.
비용이 2배씩느는데 어떻게 잘 활용하며 죽지 않는 시스템을 만들것인가.
spark/kafka/flink등.. 어디에서도 통용됨.
클라우드 - 비용 측면 / 온프렘 - 새롭게 받아들이는걸 싫어함.

 

HA에 대해 강건한 인프라를 어떻게 만들 것인가.

보안/정확성은 어느 도메인에서든 중요함. 현재 운영중인 플랫폼/인프라가 절대 무너지면안됨.
만약 온프렘이 아니더라도 분산 엔진을 써야하는 경우에 k8s를 쓰는게 유행. 
운영간 노드풀 관리를 어떻게 할것인가? 오토스케일링 그룹/카펜터는 어떻게 할것인가?
네카라 - 온프렘. 쿠팡 -aws (+ 온프렘). 
k8s직접 구축 보다는 aws라면 보통 eks를 통해서 관리함.