배움의 끄적임 👩‍💻

Trino iceberg connector의 권한을 제어하기 위해서 아래 3가지를 비교했었다.

OPA와 Apache Ranger는 외부 정책 엔진으로 Trino 외 다른 서비스(k8s, Spark, Kafka 등)에서도 사용 가능하며 Trino에 국한된게 아니다.

File-based AC + 자동 갱신 Watcher은 Trino 엔진에서 처리하는 Trino에 국한된 방식이다.

1. OPA (Open Policy Agent)

방식: 외부 OPA 서버 + Rego 정책 엔진 연동 (권한 판단 로직만 별도 관리)

장점: Rego 언어의 유연성으로 정책 설정을 최대한 건드리지 않아도 됨.

단순 A는 B를 할 수 있다/없다가 아니라 특정 시간대에만 접속 가능, 속성 기반, 컬럼 중에서 민감 정보가 포함되어 있는지 등 다양하게 체크 가능

문제점: 436 버전에서는 트리노 엔진 내부의 OPA Access Control 플러그인 지원 X

결론: Trino 446 버전으로 업그레이드

2. File-based AC + 자동 갱신 Watcher

방식: 사용자가 SQL로 권한 부여 시 Python 스크립트가 hive metastore를 폴링하여 rules.json 자동 갱신

문제점: iceberg connector는 기본적으로 allow all(규칙에 없으면 일단 허용)이라서 제어할 모든 계정, 권한을 적어야하고 누락 시 접근 허용이 된다. 그래서 사용자 계정, 스키마, 테이블 추가 시 rules.json 파일이 계속 변경되어야하고 rule.json 파일 변경 여부를 Trino 엔진이 설정된 주기마다 연동해야해서 지연이 발생한다.

결론: iceberg connector의 Default Policy의 방향성으로 인해 기각

3. Apache Ranger

방식: 데이터 보안 전용 프레임워크

문제점: 운영해야 할 컴포넌트(Admin, DB, Solr, Usersync 등)가 너무 많아 관리 포인트 늘어나고 리소스도 늘어남.

결론: Trino의 접근제어만을 위해서 도입하기엔 비효율적

Trino → OPA → Ownership Service → Metastore DB 순서로 Data flow가 흐른다.

1. 사용자 쿼리: 사용자가 Trino에 SELECT * FROM mydb.mytable 같은 쿼리를 날린다.

2. Trino의 권한 체크: Trino는 이 사용자가 해당 테이블에 접근 권한이 있는지 확인하기 위해 OPA에 "이 유저가 이 테이블 봐도 돼?"라고 HTTP 호출하여 물어본다.

3. OPA의 판단 유보: OPA는 정책(Policy)을 계산하다가, "어? 주인이 누군지 알아야 권한을 줄지 말지 결정하겠는데?"라고 판단한다.

4. Ownership Service 호출: OPA가  Ownership Service의 URL(예: http://ownership-service:8080/?schema=mydb&table=mytable)로 요청을 보낸다.

5. Metastore DB 조회 및 응답: Ownership Service는 Metastore DB에서 소유자 정보를 꺼내서 OPA에게 JSON({"owner": "admin"})으로 던져준다.

6. OPA의 최종 결정: OPA는 "방금 받은 소유자가 'admin'네? 그럼 이 쿼리 승인(Allow)!" 혹은 "소유자가 다른 사람이네? 거절(Deny)!"이라는 결과를 Trino에 보낸다.

7. Trino의 실행/차단: Trino는 OPA의 결정에 따라 쿼리를 실행하거나 Access Denied 에러를 보낸다.

트러블슈팅

1. Trino 448 버전 버그

Trino 버전을 448로 진행했었는데 Trino iceberg DB의 한글 컬럼을 조회하지 못하는 버그가 있어서 446 버전으로 다시 업그레이드를 진행했었다.

2. OPA health check port 구분

OPA는 정책 판단 쿼리가 몰리면서 부하가 있었다. 그래서 health check port를 따로 구분하여 진단용 port와 서비스용 port를 구분하여 잠시 OPA API가 느려지는 것에서 오는 k8s pod 재시작을 막았다.

Trino에서 connector는 iceberg와 hive가 존재한다.
hive와 iceberg를 모두 사용하는 이유는 데이터를 저장하고 관리하는 방식이 다르기 때문이다.
 
Hive 커넥터의 경우 폴더 단위로 데이터를 관리한다. 예를 들어서 day=20240101 폴더 안에 있는 모든 파일이 해당 파티션의 데이터에 해당한다. 그래서 UPDATE/DELETE가 어렵기에 한번 쌓이면 안 변하는 로그나 백업 테이블을 주로 append한다. hive는 메타스토어에 가서 직접 읽는다. 구조가 단순해서 나중에 다른 솔루션 (예: Spark, Python, 혹은 다른 DB)으로 데이터를 옮길 때 훨씬 호환성이 좋다.
 
iceberg connector의 경우, 스냅샷 단위로 관리하기에 UPDATE/DELETE가 자유롭고 Time Travel 기능처럼 과거 데이터를 조회하는게 가능하다. iceberg는 iceberg 전용 메타데이터 파일을 찾아서 파일에 적힌대로 데이터를 조합해서 읽는다. 따라서 빠르게 조회하거나 데이터를 수정해야하는 경우에 사용하지만 메타데이터가 꼬이면 복구가 복잡하다.
 
Trino는 데이터를 3계층으로 관리한다.
 

• Catalog (카탈로그): 네 설정 파일 이름 (hive, iceberg)
• Schema (스키마): 흔히 말하는 데이터베이스(DB) 이름
• Table (테이블): 실제 데이터가 담긴 표

 
iceberg -> hive (❌)
우선 먼저 iceberg connector로 만들어진 테이블은 hive connector로 접속 불가능하다.
아래처럼 오류 메세지가 나온다. hive connector는 iceberg의 복잡한 스냅샷 구조를 해석할 능력이 없다.

SQL Error [133001]: Query failed (#20260401_074953_00144_wv8fq): Cannot query Iceberg table 'dw.tablename'

 
 
hive -> iceberg (⭕) 
configmap에 아래 설정을 추가해주면 iceberg 커넥터가 내가 모르는 정보가 있으면 hive 카탈로그를 참고한다고 연결하는 상태가 돼서 iceberg 카탈로그 경로로 쿼리를 날려도 실제로는 hive 카탈로그에 정의된 메타데이터를 끌어와서 보여준다. iceberg 커넥터로 조회는 되지만, 단순 select만 가능하다. hive catalog를 iceberg 쪽에 브릿지해둔 경우이다. catalog를 바꾸지않아도 편하게 백업 및 개발 데이터를 볼 수 있도록 하기 위해서 설정해둔 경우가 많다.

iceberg.hive-catalog-name=hive

 
 
왜 Hive connector는 Update나 Delete가 제한적일까??
파일시스템으로써 Hive는 S3 같은 오브젝트 스토리지에 Parquet나 ORC 같은 파일을 차곡차곡 쌓아두는 방식이다. 그래서 수정하거나 삭제하려면 전체 파일을 새로 쓰거나 덮어버려야한다. Hive ACID 테이블은 가능하지만 설정이 까다롭고 성능이 나빠서 거의 사용하지 않는다. Hive connector를 이용해서 만든 테이블의 S3 폴더를 확인하면 Parquet 파일만 확인할 수 있다. Hive도 DDL를 만들때 파티션을 설정할 수 있지만, insert 시 파티션의 폴더값을 사용자가 설정해줘야하는 불편함이 있다.
 
iceberg는 데이터를 수정하면 기존 파일을 건드리지 않고, 어떤 데이터가 수정/삭제되었는지 기록하는 작은 파일(Delete Delta)을 따로 만든다. 그래서 나중에 조회할때 이 파일을 합쳐서 보여주기에 S3 파일을 통째로 다시 쓸 필요없이 Update와 Delete를 할 수 있다. 그래서 iceberg connector를 이용해서 만든 테이블의 S3 폴더를 확인하면 data 폴더(실제 데이터 저장)가 파티션별로 저장되는 것과 metastore 폴더(테이블 상태, 스키마, 파티션 정보, 스냅샷 저장)가 나눠져있는걸 볼 수 있고 metastore 폴더 하위에서 snapshot을 확인 가능하다. 가장 최신의 스냅샷 경로를 HMS에 저장해두기에 iceberg도 HMS는 필요하다.

Hive connector로 테이블을 생성하면 하이브 메타스토어에 Owner(소유자) 정보가 기록되고, Owner가 아니라면 Admin 계정이여도 테이블 삭제가 불가능하다. 그래서 소유권을 강제로 변경해줘야 Drop이 가능하다.

-- 1. 소유권을 admin으로 강제 변경
ALTER TABLE hive.dw.table_name SET AUTHORIZATION admin;

-- 2. 이제 삭제 가능
DROP TABLE hive.dw.table_name;

Hive connector로 만들었으면 테이블과 스키마를 삭제했다고 Storage가 삭제되지 않는다. 그러나 iceberg로 만들었다면 Storage 데이터도 자동으로 삭제된다.
 

Airflow 자세히 확인하기

k8s 내에 있는 airflow pod 확인

주요 Executor 방식 비교

2. 장단점 상세 분석 (엔지니어 관점)

CeleryExecutor (현재 방식)

• 장점:
  • 응답 속도가 빠름: 워커가 이미 떠 있어서 작업을 던지면 즉시 실행돼.
  • 부하 예측 가능: 워커 개수를 고정해두니 인프라 전체 자원 소모가 안정적이야.
• 단점:
  • 낭비되는 자원: 밤에 분석 작업이 없어도 워커 Pod들은 계속 떠서 월급(CPU/Mem)을 받아가.
  • 관리 포인트 증가: Redis랑 Flower까지 네가 관리해야 할 Pod이 늘어나지.

KubernetesExecutor

• 장점:
  • 극강의 효율: 일이 있을 때만 Pod이 뜨니까 클러스터 자원을 아주 알뜰하게 써.
  • 환경 자유도: 작업 A는 Python 3.8, 작업 B는 Python 3.10 이미지로 각각 다르게 띄울 수 있어. (이미지 태그를 다르게 설정 가능)
• 단점:
  • 기동 지연 (Overhead): 작업마다 Pod을 새로 만드니, 실제 코드가 돌기 전까지 5~10초 정도 준비 시간이 걸려.
  • K8s 부하: 작업이 수천 개면 K8s API 서버가 Pod 생성하느라 정신없어질 수 있어.

3. 어떤 환경에서 뭐가 더 좋을까?

CeleryExecutor가 유리한 경우

• "빨리빨리"가 중요한 환경: 1~2분 안에 끝나는 짧은 작업이 엄청나게 많이 몰아칠 때 좋아.
• 자원이 넉넉한 환경: 회사 내부 클러스터(L.Cloud 등) 자원이 여유로워서 워커 몇 개 상시 띄워두는 게 안 아까울 때.
• 이미 익숙한 운영: 지금처럼 Redis, Flower 구조가 이미 잡혀 있고 트러블슈팅이 익숙할 때.

KubernetesExecutor가 유리한 경우

• "자원 아끼기"가 최우선인 경우: 작업이 간헐적으로(예: 새벽에만 왕창) 돌 때.
• 작업마다 사양이 다를 때: 어떤 작업은 메모리 32GB가 필요하고, 어떤 건 1GB면 될 때 (Pod마다 Resource 설정을 다르게 줄 수 있음).
• 인프라 관리 최소화: Redis 같은 외부 브로커 관리가 귀찮고 K8s 기능만으로 끝내고 싶을 때.

1. Airflow: "똑똑한 작업 반장" (Workflow Orchestrator)

Airflow는 수많은 데이터 작업이 언제, 어떤 순서로 실행될지 관리하는 도구

• 역할: "A 계열사 데이터를 가져온 뒤 -> B 계열사 데이터랑 합치고 -> 마지막으로 대시보드용 테이블을 만들어라" 같은 순서를 짠(DAG) 뒤, 정해진 시간에 실행

2. Trino: "엄청나게 빠른 요리사" (Distributed SQL Query Engine)

Trino는 여러 곳에 흩어진 데이터를 SQL 한 번으로 아주 빠르게 조회해 주는 엔진

• 역할: 데이터는 S3나 HDFS에 가만히 있고, Trino는 그 데이터를 메모리로 끌어올려서 순식간에 계산해. Redshift와 비슷하지만, 데이터를 직접 저장하지 않고 '계산'만 전담하는 게 차이점

3. 둘의 관계: "주문자와 요리사"

분석가들이 서비스를 사용할 때, 이 둘은 실과 바늘처럼 움직여.

1. Airflow(주문자): "자, 이제 분석할 시간이야! 이 쿼리 좀 실행해 줘."라고 Trino(요리사)에게 주문서를 던져.
2. Trino(요리사): S3에서 데이터를 슥슥 긁어모아 요리(계산)한 뒤, 결과를 다시 테이블에 저장하거나 Airflow에 알려줘.
3. K8s(식당 건물): 너는 이 둘이 싸우지 않고 자원을 잘 나눠 쓰도록 식당 건물을 관리(Config, Resource Limit)하는 주인장인 셈이지.

러닝 스파크 chapter 3장까지 읽은 내용을 정리해 보고자 한다.

HDFS와 맵리듀스 프레임워크를 사용하여 빅데이터를 처리하는 것에는 단점이 존재한다. 가장 큰 단점은 배치 작업을 수행하면서 MR 태스크의 중간 단계들을 로컬 디스크에 써야했단 점이다. 이것은 잦은 디스크 I/O 수행을 일으키며 데이터 처리 속도가 매우 느려진다.

그래서 아파치 스파크가 탄생하였다. 아파치 스파크는 대규모 분산 데이터 연산 처리를 위해서 설계된 통합형 엔진이다.

스파크는 하둡 맵리듀스의 단점을 보완하며 업그레이드 되었고 4가지 특징이 있다.

첫번째, 속도

스파크는 질의 연산을 방향성 비순환 그래프(DAG)로 구성하고 클러스터의 워크 노드 위에서 병렬 수행된다. 스파크 SQL 엔진 의 주요 기능 중 하나인 텅스텐(Tungsten)은 전체 쿼리를 하나의 함수로 합쳐 가상 함수 호출이나 중간 데이터를 위한 CPU 사용을 없앤다. 이 방법을 포괄 코드 생성이라고 한다. 이러한 방법은 CPU 사용량을 줄이고 효율을 높인다. 하둡 맵리듀스와 다르게 모든 중간 결과는 메모리에 저장되며 디스크 I/O 사용을 줄인다.

두번째, 사용편리성

데이터 프레임, 데이터세트, RDD API로 인해 사용자가 간결하고 단순한 코드 작성이 가능하다.

세번째, 모듈성

4가지 컴포넌트(스파크SQL, 스파크 정형화 스트리밍, MLlib, GraphX)가 존재하며 하나의 엔진 안에서 사용 가능하며 별도의 엔진을 돌릴 필요도 없고 별도의 API를 배울 필요도 없다.

네번째, 확장성

스파크는 빠른 병렬 연산에 초점이 맞춰져 있기에 저장을 담당하는 여러 데이터 소스와의 확장성이 뛰어나다. 하둡은 저장과 연산을 모두 포함하고 있다.

코드가 어떻게 스파크 애플리케이션에서 동작하는지 살펴보자.

애플리케이션

- API를 써서 스파크 위에서 돌아가는 사용자 프로그램. 드라이버와 클러스터 실행기로 이루어진다.

SparkSession

- 스파크 코어 기능들과 상호 작용할 수 있는 진입점을 제공하는 객체이며, 사용자가 SparkSession 객체를 생성해야 한다.

Job

- 스파크 액션에 대한 응답으로 생성되는 여러 태스크로 이루어진 병렬 연산

Stage

- Job은 스테이지라 불리는 서로 의존성을 가지는 다수의 태스크 모음으로 나뉜다.

Task- 스파크 이그제큐터로 보내지는 작업 실행의 가장 기본적인 단위

스파크 연산은 Transformation과 Action으로 나뉘며, Action이 호출되지 않으면 스파크는 아예 실행되지 않는다.

Transformation은 원본 데이터를 수정하지 않고 하나의 스파크 데이터 프레임을 새로 만들어서 결과값을 리턴한다.

모든 Transformation은 결과값이 즉시 계산되지 않는다. 이를 lineage라 하며, 기록된 리니지는 스파크가 트렌스포메이션들끼리 합치거나 재배열하거나 해서 효율적으로 실행할 수 있도록 최적화 한다. 트랜스포메이션을 리니지에 기록만 해두고 실행하지 않는 것을 지연 평가라고 한다. 코드에서 액션을 만났을 때 모든 트랜스포메이션의 지연 연산이 발동되며 새로운 데이터 프레임을 생성하여 작업한다.

=> 지연 평가는 쿼리 최적화를 가능하게 하고 리니지와 데이터 불변성은 장애에 대한 데이터 내구성을 제공한다.

좁은/넓은 트랜스포메이션

하나의 파티션 내에서 처리 할 수 있으면 좁은 트랜스포메이션으로 filter()나 contains()가 포함된다. 반면에 파티션끼리 데이터 교환이 발생하고 디스크 작업 등이 발생하면 넓은 트랜스포메이션으로 groupby()나 orderby()가 포함된다.

하둡 생태계 데이터 수집/저장/처리/적재
데이터 수집 - Kafka - 실시간 분산환경에서 메시지를 송수신하는 메시지 전달 솔루션
분산 데이터 저장 - HDFS - 데이터를 클러스터 환경에 분산 저장하는 솔루션으로 Namenode(리더노드)와 Datanode(컴퓨팅노드)로 관리
분산 클러스터 관리 - YARN - 분산 클러스터의 리소스 관리 솔루션으로 Resourse Manager가 Node Manager를 관리
분산 데이터 배치처리 - Hadoop MapReduce - Map과 Reduce의 2상로 데이터를 처리하는 하둡 기반의 배치 작업 플랫폼
인메모리 데이터 처리 - Apache Spark - 인메모리 상에서의 데이터 처리 플랫폼으로 배치처리, 실시간 스트리밍, SQL 질의와 Graph 처리, 머신러닝 같은 하위 프로젝트를 사용
데이터웨어하우스 연동 - Hive - 하둡 기반의 데이터 웨어하우스 시스템

분산 클러스터에서 데이터 검색 방법: ElasticSearch
Restful 프로토콜을 사용하여 정형, 비정형, 데이터 위치정보, 메트릭 등 원하는 방법으로 다양한 유형의 데이터 검색 수행

하둡 생태계를 사용하는 이유
1. 하둡의 HDFS는 동일한 저가 리눅스를 분산 클러스터로 사용하여 scale out을 사용 -> 저장소의 비용이 저렴함
2. 정형 데이터 외에도 반정형, 비정형 데이터 모두 다룸
3. YARN으로 메모리, CPU, 디스크 리소스를 효율적으로 저렴하게 관리 가능 -> 운영할때 리소스 관리 포인트에서 좀 자유로움
4. 딥러닝, 인공지능에서 필요한 INPUT 데이터 제공 가능 -> 데이터 모델링을 통한 데이터의 인사이트 추출 가능
5. Schema on Read -> 데이터를 read 하는 시점에 스키마를 지정하여 데이터 추출 -> 데이터가 저장되는 스키마의 위치가 모두 달라서 시간 절약 가능

데이터 전처리
정의: 원시 데이터를 비즈니스 요구사항의 분석 및 처리에 적합한 형식으로 데이터를 가공하고 처리하는 작업<br>
기법: 데이터 정제, 데이터 통합, 데이터 변환, 데이터 정리

ETL 프로세스(Extract/Transformation/Loading)
데이터 수집하고 비즈니스 규칙에 따라 데이터를 변환한 후 대상 데이터를 저장소로 적재하는데 사용되는 데이터 파이프라인

ELT 프로세스(Extract&Load/Transformation)
일단 데이터를 수집항 적재하고 이후 비즈니스 요구사항에 맞춰 데이터 변환 -> 비즈니스 요구사항이 정해지지 않았어도 데이터 적재 가능

하둡파일시스템
- 구성: 마스터 역할의 NameNode와 데이터를 실제로 저장하는 다수의 슬레이브 Datanode
- 구조: 마스터-슬레이브 구조(하나의 마스터에 여러개의 슬레이브 구조)
- 동작원리
  1. 사용자가 데이터를 HDFS에 저장
  2. HDFS Client는 메타데이터(파일명, 폴더구조 등)를 NameNode에 저장
  3. NameNode는 Datanode에게 데이터 관련 명령 지시,  Datanode는 자신의 상태 정보를 NameNode에게 송신
  4. Datanode안의 여러개의 블록으로 나눠져 실데이터가 복제 저장됨
  5. 사용자가 데이터를 읽어올 경우, HDFS Client는 NameNode로부터 파일 블록에 대한 메타 데이터 정보를 수신
  6. HDFS Client가 원하는 데이터가 있는 Datanode의 파일 블록을 찾아서 파일을 읽어옴.
  7. NameNode의 메타 데이터 변경 시, Secondary Namenode의 Edits 파일에 로그 저장
 
Yarn의 구조
|데몬|기능|
|------|---|
|Resource Manager|Yarn의 마스터 역할 수행(총괄 관리자)|
|Scheduler|Resource Manager 내의 데몬으로 Node Manager와 통신하며, Application Master에 자원 할당|
|Application Manager|Resource Manager 내의 데몬으로 Application Master 실행 및 상태 관리|
|Node Manager|각 노드의 상태 정보를 모니터링 하여 Resource Manager에게 제공|
|Application Master|각각의 Application을 관리하는 마스터의 역할|
|Container|Application Master가 관리하는 태스크를 수행, Node Manager에 의해 생성|

하둡파일시스템의 장점
- 스케일 아웃 분산 시스템(병렬로 스케일 아웃됨으로 데이터 양에 따라서 서버 필요)
- 장애복구(마스터 서버가 슬레이브 서버의 장애를 실시간으로 감지하여 서비스 우회 OR 다른 노드에 복제된 데이터 사용)
- 대용량데이터(한번 저장된 데이터는 only read = 데이터 업데이트 X)