👉 공식 문서

Airflow는 여러 컨테이너(Webserver, Scheduler, DB, Redis 등)가 함께 동작하기 때문에, VM 리소스가 너무 작으면 설치는 되더라도 실행이 불안정할 수 있습니다.

✅ 2) Docker 설치

VM에 Docker 및 Docker Compose 플러그인을 설치합니다.
아래 스크립트를 파일로 저장 후 실행하면 한 번에 설치할 수 있습니다.

docker_install.sh

파일 생성 후 실행 → bash docker_install.sh

#!/bin/bash

set -e

echo "=== Docker 및 Docker Compose 설치 스크립트 ==="

# 1. 기존 Docker 제거
echo "[1/6] 기존 Docker 제거..."
sudo apt-get remove -y docker docker-engine docker.io containerd runc || true

# 2. 패키지 업데이트 및 의존성 설치
echo "[2/6] 패키지 업데이트 및 의존성 설치..."
sudo apt-get update
sudo apt-get install -y ca-certificates curl gnupg lsb-release

# 3. Docker 공식 GPG 키 추가
echo "[3/6] Docker GPG 키 추가..."
sudo mkdir -p /etc/apt/keyrings
curl -fsSL https://download.docker.com/linux/ubuntu/gpg | \
  sudo gpg --dearmor -o /etc/apt/keyrings/docker.gpg

# 4. Docker 리포지토리 등록
echo "[4/6] Docker 리포지토리 추가..."
echo \
  "deb [arch=$(dpkg --print-architecture) signed-by=/etc/apt/keyrings/docker.gpg] \
  https://download.docker.com/linux/ubuntu $(lsb_release -cs) stable" | \
  sudo tee /etc/apt/sources.list.d/docker.list > /dev/null

# 5. Docker Engine 설치
echo "[5/6] Docker 설치..."
sudo apt-get update
sudo apt-get install -y docker-ce docker-ce-cli containerd.io docker-buildx-plugin docker-compose-plugin

# 6. 설치 확인
echo "[6/6] 설치 확인..."
docker --version
docker compose version

echo "✅ Docker 및 Docker Compose 설치 완료!"

✅ 3) VM 방화벽 설정

Airflow Web UI는 기본적으로 8080 포트에서 실행됩니다.
따라서 VM 외부에서 웹 UI에 접속하려면 클라우드의 방화벽 설정 에서 8080 포트를 허용해야 합니다.

• AWS → VPC 보안그룹(Security Group) 설정
• GCP → VPC 방화벽 규칙 설정 (참고 영상)

✅ 4) Airflow 설치 (Docker Compose 기반)

Airflow는 공식 문서에서 제공하는 docker-compose.yaml 파일로 빠르게 설치할 수 있습니다.

👉 공식 문서

1) 작업 디렉토리 생성 및 이동

Airflow 관련 파일을 관리할 디렉토리를 만들고 이동합니다.

mkdir Apache_Airflow
cd Apache_Airflow

2) docker-compose.yaml 다운로드

아래 명령어로 Airflow 공식 docker-compose.yaml 파일을 다운로드합니다.

curl -LfO 'https://airflow.apache.org/docs/apache-airflow/3.1.5/docker-compose.yaml'

3) 예제 DAG 로딩 비활성화

docker-compose.yaml 파일에서 아래 설정을 찾아 값을 변경합니다.

• AIRFLOW__CORE__LOAD_EXAMPLES: 'true' → 'false'

4) 필요한 폴더 및 .env 파일 생성

Airflow가 기본적으로 사용하는 디렉토리 및 파일들을 만들어줍니다. (디렉토리 4개 + 파일 1개)

mkdir -p ./dags ./logs ./plugins ./config
echo -e "AIRFLOW_UID=50000" > .env

5) Airflow 초기화 수행

초기 DB 셋업 등을 진행합니다. 명령 실행 후 터미널로 돌아올 때까지 기다립니다.

sudo docker compose up airflow-init

6) Airflow 실행

전체 컨테이너를 실행합니다.

sudo docker compose up -d

7) 모든 컨테이너가 정상 상태(healthy) 될 때까지 대기

Airflow는 여러 컨테이너가 동시에 떠야 정상적으로 동작합니다. 아래 명령어로 컨테이너 상태를 모니터링합니다.

sudo watch -n 1 docker ps

총 7개 컨테이너가 모두 healthy 상태가 될 때까지 기다립니다.

8) Airflow UI 접속

'healthy' 확인 후 VM의 공인 IP를 확인한 후 아래 URL로 접속합니다.

http://[Compute Engine 공인 IP]:8080

ID : airflow
PW : airflow

✅ 5) Airflow 작업 디렉토리 권한 변경

Docker Compose로 Airflow를 설치하면 dags/, logs/, plugins/, config/ 등의 폴더를 컨테이너와 volume(bind mount) 으로 연결하게 됩니다.
이 과정에서 컨테이너가 파일을 생성하거나 수정하면, 파일 소유자가 내 계정이 아닌 다른 사용자(root/airflow) 로 잡히는 경우가 있어 VM에서 직접 편집할 때 권한 오류(Permission denied)가 발생할 수 있습니다.

이를 해결하려면, 아래 명령어를 통해 Airflow 작업 디렉토리의 소유권을 현재 로그인한 사용자 계정으로 변경해주면 됩니다.

sudo chown -R "${USER:-$(id -un)}" .

왜 Airflow 스케줄링을 이해해야 할까?

데이터 파이프라인을 운영하다 보면 "매일 새벽에 ETL을 돌리고 싶다", "평일 오후에만 리포트를 생성하고 싶다", "3일마다 한 번씩 배치를 돌리고 싶다" 같은 요구가 생깁니다. Apache Airflow에서는 DAG의 schedule 파라미터로 이런 실행 주기를 정의합니다. 스케줄 방식을 잘못 설정하면 과거 구간이 한꺼번에 돌아가는 catchup 문제가 생기거나, 의도와 다른 시간에 DAG가 실행될 수 있어요. 이 글에서는 Airflow가 제공하는 4가지 스케줄링 방식(Presets, Cron, Timedelta/Delta, Timetable)을 개념부터 코드 예시, 실무 선택 기준까지 정리합니다.

Airflow 스케줄링 방식 개요

Airflow에서 DAG 실행 시점을 정하는 방식은 크게 네 가지로 나눌 수 있습니다.

1. Presets (사전 정의 문자열) — @daily, @hourly 등 고정된 의미의 문자열
2. Cron Expression — Unix cron 형식의 5필드 표현식
3. Timedelta / DeltaDataIntervalTimetable — 고정 간격(예: 3일마다, 6시간마다)
4. Timetable (Custom / EventsTimetable) — 불규칙한 날짜·이벤트 기반 스케줄

1. Presets (사전 정의 문자열)

1) 개요

주요 기능:

• Airflow가 미리 정의한 문자열로, "매일 자정", "매시간", "매주 일요일" 등 흔한 주기를 한 번에 표현
• 내부적으로 cron 표현식으로 변환되어 처리됨
• 코드가 짧고 의도가 바로 읽혀서 초보자에게 적합

실무 관점:

• 일일 리포트, 시간 단위 로그 수집, 주간/월간 정산처럼 규칙적인 주기가 있을 때 가장 먼저 고려
• "매일 새벽 0시에 돌리면 된다" 수준이면 @daily 한 줄로 해결
• 나중에 "매일 오전 9시로 바꿔 달라"는 요구가 생기면 cron 표현식(0 9 * * *)으로 전환하는 경우가 많음

언제 필요한가?

• 단순하고 일반적인 주기만 필요할 때
• 팀 내에서 "매일/매시간/매주"라는 표현으로 소통할 때

from airflow import DAG
from airflow.operators.empty import EmptyOperator
from pendulum import datetime

with DAG(
    dag_id="preset_example",
    schedule="@daily",  # 매일 자정 실행
    start_date=datetime(2026, 1, 1, tz="Asia/Seoul"),
    catchup=False,
):
    task = EmptyOperator(task_id="daily_task")

2. Cron Expression

1) 개요

주요 기능:

• Unix cron과 동일한 5필드 형식(분, 시, 일, 월, 요일)으로 실행 시점을 세밀하게 지정
• "평일 오후 4시", "매월 1일·15일 자정", "30분마다" 등 복합 조건 표현 가능
• Airflow 내부에서는 CronDataIntervalTimetable로 변환되어 동작

실무 관점:

• 업무 시간대에만 돌리거나(예: 평일 9–18시), 특정 요일·특정 일자에만 실행해야 할 때 cron이 표준 선택
• 예: 0 16 * * MON-FRI → 평일 오후 4시, 0 0 1,15 * * → 매월 1일·15일 자정
• 타임존은 DAG의 start_date 등에서 사용하는 datetime에 tz를 지정해 일관되게 맞추는 것이 중요

언제 필요한가?

• "특정 요일", "특정 시각", "특정 일자" 조합이 필요할 때
• Preset으로는 표현이 안 되는 구체적인 시간 조건이 있을 때

┌─────── 분 (0-59)
│ ┌───── 시 (0-23)
│ │ ┌─── 일 (1-31)
│ │ │ ┌─ 월 (1-12)
│ │ │ │ ┌ 요일 (0-6, 0=일요일)
│ │ │ │ │
* * * * *

from airflow import DAG
from airflow.operators.empty import EmptyOperator
from pendulum import datetime

with DAG(
    dag_id="cron_example",
    schedule="0 16 * * MON-FRI",  # 평일 오후 4시
    start_date=datetime(2026, 1, 1, tz="Asia/Seoul"),
    catchup=False,
):
    task = EmptyOperator(task_id="weekday_afternoon_task")

자주 쓰는 패턴:

• 0 9 * * * → 매일 오전 9시
• */30 * * * * → 30분마다
• 0 0 1,15 * * → 매월 1일, 15일 자정

3. Timedelta / DeltaDataIntervalTimetable

1) timedelta 직접 사용 (간단한 방식)

주요 기능:

• "N일마다", "N시간마다", "N분마다"처럼 고정 간격 실행을 표현
• schedule=timedelta(days=3)처럼 넣으면 Airflow가 내부적으로 DeltaDataIntervalTimetable로 변환

실무 관점:

• "3일마다 한 번", "6시간마다 한 번" 같은 주기는 cron보다 timedelta가 의도가 분명함
• 예: timedelta(days=3), timedelta(hours=6), timedelta(minutes=30)
• cron은 "언제"를 지정하고, timedelta는 "간격"을 지정한다는 차이를 두고 선택하면 됨

언제 필요한가?

• 실행 시점이 "매일 0시"가 아니라 "N시간/ N일 간격"일 때
• 주기만 중요하고 정각 맞춤이 필요 없을 때

from airflow import DAG
from airflow.operators.empty import EmptyOperator
from pendulum import datetime
from datetime import timedelta

with DAG(
    dag_id="timedelta_simple_example",
    schedule=timedelta(days=3),  # 3일마다 실행
    start_date=datetime(2026, 1, 1, tz="Asia/Seoul"),
    catchup=False,
):
    task = EmptyOperator(task_id="every_3_days_task")

2) DeltaDataIntervalTimetable 사용 (명시적 방식)

주요 기능:

• Timetable 객체를 직접 지정해 "고정 간격" 스케줄을 명시적으로 표현
• 커스텀 Timetable과 조합하거나, 코드에서 "Timetable 기반 스케줄"임을 드러내고 싶을 때 사용

실무 관점:

• schedule=timedelta(...)와 동작은 동일함. 다만 "Timetable을 쓰고 있다"는 것이 코드에 드러남
• 커스텀 Timetable을 만들거나, 여러 Timetable을 조합하는 고급 패턴으로 넘어갈 때 같은 계열로 이해하면 됨

언제 필요한가?

• 고정 간격을 Timetable API 수준에서 명시하고 싶을 때
• 커스텀 Timetable과의 일관성을 위해 같은 방식으로 쓰고 싶을 때

from airflow import DAG
from airflow.operators.empty import EmptyOperator
from airflow.timetables.simple import DeltaDataIntervalTimetable
from pendulum import datetime
from datetime import timedelta

every_3_days = DeltaDataIntervalTimetable(timedelta(days=3))

with DAG(
    dag_id="delta_timetable_example",
    schedule=every_3_days,
    start_date=datetime(2026, 1, 1, tz="Asia/Seoul"),
    catchup=False,
):
    task = EmptyOperator(task_id="every_3_days_task")

💡 schedule=timedelta(...)를 넣으면 Airflow 내부에서 DeltaDataIntervalTimetable로 자동 변환됩니다. 결과는 동일합니다.

4. Timetable (Custom / EventsTimetable)

1) EventsTimetable

주요 기능:

• 특정 datetime 목록만큼만 DAG를 실행하도록 함
• 공휴일, 이벤트일, 월말 등 불규칙한 날짜에만 돌리고 싶을 때 사용
• Airflow 2.2+에서 도입된 Timetable API의 대표 활용 예

실무 관점:

• "이 날짜들에만 실행"이 필요할 때(예: 월별 마감일, 특별 이벤트 일자)
• event_dates 리스트를 코드나 설정에서 생성해 넘기면 됨
• catchup=True와 조합하면 지정한 과거 날짜들도 한 번씩 실행 가능

언제 필요한가?

• 실행 일자가 규칙적인 주기가 아닐 때
• 공휴일 제외, 특정 영업일만 등 불규칙한 날짜 집합이 필요할 때

from airflow.sdk import dag, task
from pendulum import datetime
from airflow.timetables.events import EventsTimetable

special_dates = EventsTimetable(
    event_dates=[
        datetime(2026, 1, 1, tz="Asia/Seoul"),
        datetime(2026, 1, 15, tz="Asia/Seoul"),
        datetime(2026, 1, 26, tz="Asia/Seoul"),
        datetime(2026, 1, 30, tz="Asia/Seoul"),
    ]
)

@dag(
    schedule=special_dates,
    start_date=datetime(2026, 1, 1, tz="Asia/Seoul"),
    end_date=datetime(2026, 1, 31, tz="Asia/Seoul"),
    catchup=True,
)
def events_timetable_example():
    @task
    def run_on_special_date():
        print("특별한 날에만 실행!")

    run_on_special_date()

events_timetable_example()

5. 스케줄링 방식 요약

6. Timetable Import 정리

# Delta 관련
from datetime import timedelta
from airflow.timetables.simple import DeltaDataIntervalTimetable

# Events 관련
from airflow.timetables.events import EventsTimetable

# Cron 관련 (필요시)
from airflow.timetables.simple import CronDataIntervalTimetable

7. 실무 활용 가이드

신규 DAG 설계 시 선택 순서

1. "매일/매시간/매주" 수준이면 → Presets (@daily, @hourly 등)
2. "평일 오후 4시", "매월 1일·15일"처럼 구체 시각/요일/일자가 필요하면 → Cron
3. "N시간/ N일마다" 간격이 중요하면 → Timedelta (또는 필요 시 DeltaDataIntervalTimetable)
4. 날짜가 불규칙하면 → EventsTimetable 또는 커스텀 Timetable

주의사항

• start_date와 catchup: 과거 구간을 채울지 여부를 catchup로 제어. 기본값이 True이므로 의도치 않은 대량 실행을 막으려면 catchup=False를 자주 사용함
• 타임존: pendulum.datetime(..., tz="Asia/Seoul") 등으로 DAG와 태스크에서 타임존을 통일할 것
• Timetable은 Airflow 2.2+ 기능이며, schedule이 내부적으로 어떻게 Timetable로 매핑되는지 이해하면 디버깅과 확장에 유리함

8. 마무리

• Airflow 스케줄은 Presets → Cron → Timedelta/Delta → EventsTimetable(커스텀) 순으로 "단순 규칙 → 세밀한 시간 → 간격 → 불규칙 일자"를 다룹니다.
• 실무에서는 "매일 새벽"이면 Preset, "평일 특정 시각"이면 Cron, "N일마다"면 Timedelta, "이 날들만"이면 EventsTimetable로 정리하면 선택이 쉽습니다.
• catchup과 타임존 설정을 함께 점검하면 예상치 못한 실행을 줄일 수 있습니다.

참고 자료:

• Apache Airflow Documentation - DAGs
• Airflow Timetables (공식 문서 내 Timetable 설명)

왜 Databricks 아키텍처를 이해해야 할까?

Databricks를 처음 사용하는 개발자나 관리자에게 가장 혼란스러운 부분 중 하나는 "데이터가 어디에 저장되는가?", "컴퓨팅 리소스는 어디서 실행되는가?", "보안은 어떻게 구성되는가?"와 같은 아키텍처 관련 질문입니다.

Databricks는 전통적인 단일 계정 구조가 아니라, Databricks 계정과 고객 계정으로 분리된 하이브리드 아키텍처를 사용합니다. 이 구조를 이해하지 못하면, 네트워크 설정, 보안 정책, 비용 관리 등에서 예상치 못한 문제에 직면할 수 있습니다.

이 글에서는 Databricks의 데이터 인텔리전스 플랫폼 아키텍처를 계층별로 설명하고, 각 구성 요소의 역할과 상호작용을 이해할 수 있도록 정리합니다.

Databricks 아키텍처의 핵심 개념

Databricks 플랫폼은 크게 두 가지 계정 영역으로 나뉘어 운영됩니다:

1. Databricks 계정 (Databricks Account)

   • Databricks가 직접 관리하는 영역
   • Control Plane과 Serverless Compute가 배포됨
   • 고객의 클라우드 계정과 분리되어 운영

2. 고객 계정 (Customer Account)

   • 고객의 클라우드 환경(AWS, Azure, GCP) 내에 존재
   • 데이터 저장소(Cloud Storage)와 Classic Compute가 실행됨
   • 고객이 직접 관리하는 네트워크 및 보안 설정 적용

이러한 분리 구조는 다음과 같은 이점을 제공합니다:

• 보안 격리: Control Plane과 데이터 저장소를 분리하여 보안 강화
• 유연한 배포: Serverless와 Classic Compute를 선택적으로 사용 가능
• 비용 최적화: 사용 패턴에 따라 적절한 컴퓨팅 모델 선택 가능

1. 전체 플랫폼 구조 (High-Level Architecture)

1) 아키텍처 다이어그램 개념

┌─────────────────────────────────────────────────────────┐
│              Databricks 계정 (Databricks Account)   │
│  ┌──────────────────────────────────────────────────┐   │
│  │         Control Plane (컨트롤 플레인)          │   │
│  │  - Web App                                   │   │
│  │  - Unity Catalog                             │   │
│  │  - Workflow Management                       │   │
│  │  - Intelligence Engine                       │   │
│  └──────────────────────────────────────────────────┘   │
│  ┌──────────────────────────────────────────────────┐   │
│  │    Serverless Compute Plane (서버리스 컴퓨팅)  │   │
│  │  - Serverless SQL Warehouse                  │   │
│  │  - Model Serving                             │   │
│  │  - Vector Search                             │   │
│  │  - Online Tables                             │   │
│  └──────────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────────┘
                          │
                          │ 네트워크 연결
                          │
┌─────────────────────────────────────────────────────────┐
│            고객 계정 (Customer Cloud Account)        │
│  ┌──────────────────────────────────────────────────┐   │
│  │         Classic Compute Plane                │   │
│  │  - Workspace Clusters                        │   │
│  │  - Classic SQL Warehouse                     │   │
│  └──────────────────────────────────────────────────┘   │
│  ┌──────────────────────────────────────────────────┐   │
│  │              Cloud Storage                   │   │
│  │  - S3 / ADLS / GCS                           │   │
│  │  - Delta Tables                              │   │
│  └──────────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────────┘

2) 계정 간 데이터 흐름

1. 사용자 요청: 사용자가 Web App을 통해 쿼리나 작업을 요청
2. Control Plane 처리: Unity Catalog가 권한을 확인하고 작업을 스케줄링
3. Compute 실행: Classic 또는 Serverless Compute에서 실제 데이터 처리
4. Storage 접근: Compute가 고객 계정의 Cloud Storage에서 데이터 읽기/쓰기
5. 결과 반환: 처리 결과를 사용자에게 반환

2. 주요 구성 요소 상세 설명

1) Control Plane (컨트롤 플레인)

Control Plane은 Databricks의 두뇌 역할을 하며, 플랫폼의 모든 관리 및 오케스트레이션 기능을 제공합니다.

Web App

역할: 사용자가 Databricks 플랫폼에 접속하는 주요 인터페이스

주요 기능:

• 노트북 작성 및 실행
• SQL 쿼리 실행
• 작업(Jobs) 관리 및 모니터링
• Unity Catalog를 통한 데이터 탐색
• 클러스터 및 워크스페이스 관리

실무 관점:

• Web App은 Databricks 계정에 배포되어 있으므로, 인터넷 연결만 있으면 어디서든 접근 가능
• SSO(Single Sign-On)를 통해 기업 인증 시스템과 통합 가능

Unity Catalog

역할: 데이터 거버넌스의 핵심으로, 모든 데이터 객체에 대한 중앙 집중식 메타데이터 및 권한 관리

주요 기능:

• 접근 제어(Access Control): 테이블, 뷰, 함수 등에 대한 세밀한 권한 관리
• 메타데이터 관리: 데이터 객체의 스키마, 통계, 리니지 정보 저장
• 데이터 리니지: 데이터의 출처와 변환 과정 추적
• 데이터 검색: 메타데이터 기반 데이터 검색 및 탐색

실무 관점:

• Unity Catalog는 Databricks 계정에 위치하지만, 고객 계정의 스토리지에 대한 메타데이터를 관리
• 여러 워크스페이스에서 동일한 메타스토어를 공유하여 데이터 일관성 유지

Workflow Management

역할: 데이터 파이프라인과 작업 워크플로우를 오케스트레이션

주요 기능:

• Jobs: 스케줄링된 작업 실행 및 관리
• Delta Live Tables (DLT): 선언적 데이터 파이프라인 구축
• 의존성 관리: 작업 간 의존성 및 실행 순서 관리
• 모니터링: 작업 실행 상태 및 성능 모니터링

실무 관점:

• Workflow Management는 작업을 스케줄링하고 모니터링하지만, 실제 실행은 Compute Plane에서 수행
• 작업 실패 시 자동 재시도 및 알림 기능 제공

Intelligence Engine

역할: 머신러닝 모델을 사용하여 플랫폼을 최적화하고 관리

주요 기능:

• 자동 최적화: 쿼리 성능 및 리소스 사용량 최적화
• 예측 분석: 리소스 사용 패턴 예측 및 자동 스케일링
• 비용 최적화: 비용 효율적인 리소스 할당 제안
• 문제 감지: 성능 저하나 오류 패턴 자동 감지

실무 관점:

• Intelligence Engine은 백그라운드에서 작동하여 사용자가 명시적으로 설정하지 않아도 자동으로 최적화 수행
• 시간이 지날수록 더 정확한 최적화 제안 제공

2) Compute Plane (컴퓨팅 플레인)

Compute Plane은 실제 데이터 처리가 일어나는 계층입니다. Databricks는 두 가지 컴퓨팅 모델을 제공합니다.

Classic Compute (클래식 컴퓨팅)

특징:

• 고객의 클라우드 계정 내 가상 네트워크(VNet/VPC)에서 실행
• 고객이 직접 네트워크 및 보안 설정 관리
• 완전한 제어권과 커스터마이징 가능

주요 구성 요소:

• Workspace Clusters: 노트북 실행 및 작업 실행용 클러스터
• Classic SQL Warehouse: SQL 쿼리 실행용 전용 웨어하우스

장점:

• 네트워크 격리 및 보안 정책을 완전히 제어 가능
• 기존 클라우드 인프라와의 통합 용이
• 특정 규정 준수 요구사항 충족 가능

단점:

• 클러스터 시작 시간이 상대적으로 김 (수 분 소요)
• 유지보수 및 패치 관리를 고객이 담당해야 함
• 초기 설정 및 구성이 복잡할 수 있음

실무 활용:

• 엄격한 보안 요구사항이 있는 조직
• 기존 클라우드 네트워크와의 통합이 필요한 경우
• 장기 실행 작업이나 대용량 데이터 처리

Serverless Compute (서버리스 컴퓨팅)

특징:

• Databricks 계정 내에서 실행
• Databricks가 인프라 관리 및 유지보수 담당
• 빠른 시작 시간과 자동 스케일링

주요 구성 요소:

• Serverless SQL Warehouse: 서버리스 환경에서 실행되는 SQL 웨어하우스
• Model Serving: 실시간 ML 모델 서빙
• Vector Search: 벡터 검색 서비스
• Online Tables: 실시간 데이터 동기화

장점:

• 빠른 시작: 클러스터 시작 시간이 수 초 내로 단축
• 유지보수 부담 감소: Databricks가 패치 및 업데이트 관리
• 자동 스케일링: 워크로드에 따라 자동으로 리소스 조정
• 비용 효율성: 사용한 만큼만 비용 지불

단점:

• 네트워크 제어가 제한적 (Private Link로 일부 해결 가능)
• 특정 커스터마이징 제한
• 다중 테넌트 환경 (격리는 보장되지만)

실무 활용:

• 빠른 쿼리 응답이 필요한 BI 및 분석 작업
• 간헐적인 워크로드
• 유지보수 부담을 줄이고 싶은 경우
• 실시간 AI 애플리케이션

3. 서비스 모델별 아키텍처 특징

1) Databricks SQL: Classic vs. Serverless

Classic SQL Warehouse

아키텍처:

고객 클라우드 계정
  └─ VNet/VPC
      └─ 로드 밸런서
          └─ 컴퓨팅 클러스터 (고객 관리)
              └─ Cloud Storage 접근

특징:

• 고객의 클라우드 계정 내에서 실행
• 로드 밸런서 뒤에서 실행되는 컴퓨팅 클러스터 사용
• 네트워크 및 보안 설정을 고객이 완전히 제어
• VNet/VPC 피어링을 통한 온프레미스 시스템과의 통합 가능

실무 관점:

• 기존 클라우드 네트워크와의 통합이 필요한 경우
• 특정 IP 대역이나 방화벽 규칙이 필요한 경우
• 장기 실행 쿼리나 대용량 데이터 처리

Serverless SQL Warehouse

아키텍처:

Databricks 계정
  └─ 다중 테넌트 인프라
      └─ VM 격리 (테넌트별)
          └─ 네트워크 격리 (테넌트별)
              └─ Cloud Storage 접근 (Private Link)

특징:

• 컴퓨팅 리소스가 Databricks 클라우드 계정에서 실행
• 다중 테넌트(Multi-tenant) 구조
• 각 테넌트 간 VM 및 네트워크 수준에서 엄격히 격리
• Private Link를 통한 안전한 스토리지 접근

실무 관점:

• 빠른 쿼리 시작이 필요한 경우
• 유지보수 부담을 줄이고 싶은 경우
• Azure 환경에서는 Private Link로 보안 연결 가능

보안 고려사항:

• 다중 테넌트 환경이지만 격리는 보장됨
• Azure에서는 Private Link를 통해 스토리지와의 통신을 비공개로 유지 가능
• AWS와 GCP에서도 유사한 프라이빗 연결 옵션 제공

2) Serverless 확장 서비스

Model Serving & Vector Search

역할: 실시간 AI 애플리케이션을 위한 서버리스 인프라 제공

Model Serving:

• ML 모델을 REST API 엔드포인트로 노출
• 자동 스케일링 및 고가용성 보장
• A/B 테스팅 및 모델 버전 관리 지원

Vector Search:

• 벡터 임베딩을 인덱싱하여 유사도 검색 제공
• RAG(Retrieval-Augmented Generation) 애플리케이션 지원
• 실시간 업데이트 및 검색 가능

실무 활용:

• 챗봇 및 생성형 AI 애플리케이션
• 추천 시스템
• 이미지 및 텍스트 유사도 검색

Online Tables

역할: Delta Table과 실시간으로 동기화되는 온라인 테이블 제공

특징:

• Delta Table의 변경사항을 실시간으로 동기화
• 낮은 지연 시간의 읽기 접근 제공
• 서버리스 인프라에서 자동 관리

실무 활용:

• 실시간 추천 시스템
• 실시간 대시보드
• 실시간 의사결정 애플리케이션

4. 보안 및 사용자 관리

1) Identity Provider (IDP) 통합

역할: 고객의 중앙 사용자 관리 시스템과 Databricks를 통합

지원하는 IDP:

• Azure AD / Microsoft Entra ID: Azure 환경에서 주로 사용
• Okta: 엔터프라이즈 SSO 솔루션
• Google Workspace: GCP 환경에서 주로 사용
• SAML 2.0 호환 IDP: 기타 SAML 2.0을 지원하는 모든 IDP

주요 기능:

• SSO (Single Sign-On): 기업 인증 시스템을 통한 자동 로그인
• SCIM 프로비저닝: 사용자 및 그룹 자동 동기화
• 역할 기반 접근 제어: IDP 그룹을 Databricks 그룹으로 매핑

실무 관점:

• 사용자 생명주기 관리를 IDP에서 중앙 집중식으로 관리
• 퇴사자나 역할 변경 시 자동으로 Databricks 접근 권한 업데이트
• 여러 워크스페이스에서 동일한 사용자 및 그룹 구조 공유 가능

2) Network Security

Private Link (Azure)

역할: 서버리스 컴퓨팅 플레인에서 고객의 스토리지 계정으로의 보안 연결 제공

작동 방식:

1. Databricks가 고객의 VNet에 Private Endpoint 생성
2. 서버리스 컴퓨팅이 Private Endpoint를 통해 스토리지 접근
3. 모든 트래픽이 Microsoft 백본 네트워크를 통해 전송
4. 공용 인터넷을 거치지 않아 보안 강화

장점:

• 공용 인터넷을 거치지 않는 안전한 연결
• 네트워크 격리 및 방화벽 규칙 적용 가능
• 데이터 유출 위험 감소

실무 활용:

• 엄격한 보안 요구사항이 있는 조직
• 규정 준수 요구사항 충족
• 민감한 데이터 처리

VNet/VPC 피어링 (Classic Compute)

역할: Classic Compute가 고객의 기존 네트워크와 직접 통신

작동 방식:

1. Databricks VNet/VPC와 고객 VNet/VPC 간 피어링 설정
2. Classic Compute가 피어링된 네트워크를 통해 리소스 접근
3. 온프레미스 시스템과의 VPN/ExpressRoute 연결 가능

실무 활용:

• 온프레미스 데이터베이스 접근
• 기존 클라우드 리소스와의 통합
• 하이브리드 클라우드 아키텍처

5. 실무 활용 가이드

1) 시나리오 1: 엄격한 보안 요구사항이 있는 금융 기관

요구사항:

• 모든 데이터가 고객 계정 내에만 존재
• 네트워크 격리 및 방화벽 규칙 적용
• 온프레미스 시스템과의 통합 필요

아키텍처 선택:

• Classic Compute 사용
• VNet/VPC 피어링을 통한 온프레미스 연결
• Private Link는 사용하지 않음 (모든 리소스가 고객 계정 내)

구성:

고객 계정 (Azure)
  ├─ VNet (피어링됨)
  │   ├─ Classic SQL Warehouse
  │   └─ Workspace Clusters
  ├─ Storage Account (Private Endpoint)
  └─ 온프레미스 연결 (ExpressRoute)

2) 시나리오 2: 빠른 프로토타이핑이 필요한 스타트업

요구사항:

• 빠른 시작 및 유지보수 최소화
• 비용 효율성
• 실시간 AI 기능 필요

아키텍처 선택:

• Serverless Compute 우선 사용
• Model Serving 및 Vector Search 활용
• Classic Compute는 대용량 배치 작업에만 사용

구성:

Databricks 계정
  ├─ Serverless SQL Warehouse (일반 쿼리)
  ├─ Model Serving (실시간 추론)
  └─ Vector Search (RAG 애플리케이션)

고객 계정
  └─ Cloud Storage (Delta Tables)

3) 시나리오 3: 하이브리드 워크로드가 있는 대기업

요구사항:

• 다양한 워크로드 지원 (배치, 스트리밍, 실시간)
• 보안과 성능의 균형
• 비용 최적화

아키텍처 선택:

• Classic Compute: 장기 실행 배치 작업, 엄격한 보안 요구사항
• Serverless Compute: 빠른 쿼리, 실시간 AI, 간헐적 워크로드
• Private Link: Serverless에서 스토리지 접근 시 사용

구성:

Databricks 계정
  ├─ Serverless SQL Warehouse (BI 쿼리)
  ├─ Model Serving (실시간 AI)
  └─ Private Link (스토리지 접근)

고객 계정
  ├─ Classic SQL Warehouse (대용량 배치)
  ├─ Workspace Clusters (데이터 엔지니어링)
  └─ Cloud Storage (Delta Tables)

6. 아키텍처 선택 가이드

1) Classic vs. Serverless 비교표

2) 선택 기준

Classic Compute를 선택해야 하는 경우:

• 엄격한 네트워크 격리 요구사항
• 온프레미스 시스템과의 직접 통합 필요
• 장기 실행 작업이나 대용량 데이터 처리
• 특정 규정 준수 요구사항 (예: 데이터가 특정 지역에만 존재해야 함)

Serverless Compute를 선택해야 하는 경우:

• 빠른 시작 시간이 중요
• 유지보수 부담을 줄이고 싶음
• 간헐적인 워크로드
• 실시간 AI 애플리케이션
• 비용 효율적인 리소스 사용

하이브리드 접근:

• 대부분의 조직은 두 방식을 조합하여 사용
• 워크로드 특성에 따라 적절한 Compute 모델 선택
• Classic은 배치 및 데이터 엔지니어링, Serverless는 분석 및 실시간 작업

7. 마무리

Databricks의 데이터 인텔리전스 플랫폼 아키텍처는 Databricks 계정과 고객 계정으로 분리된 하이브리드 구조를 통해 보안, 성능, 유연성을 모두 제공합니다.

핵심 요약:

1. 두 계정 구조: Control Plane과 Serverless Compute는 Databricks 계정에, 데이터와 Classic Compute는 고객 계정에 위치
2. Compute 선택: Classic은 완전한 제어와 격리, Serverless는 빠른 시작과 유지보수 편의성 제공
3. 보안 계층: IDP 통합, 네트워크 격리, 세밀한 접근 제어를 통해 다층 보안 구현
4. 유연한 구성: 워크로드 특성에 따라 Classic과 Serverless를 조합하여 사용

아키텍처를 올바르게 이해하고 구성하면, 보안을 유지하면서도 성능과 비용을 최적화할 수 있습니다. 처음 구축할 때는 작은 규모로 시작하여 점진적으로 확장하는 것이 좋습니다.

참고 자료:

• Databricks 공식 문서: Architecture Overview
• Databricks 공식 문서: Serverless Compute
• Databricks 공식 문서: Network Security

왜 Databricks 관리자 역할을 이해해야 할까?

Databricks를 처음 도입하거나 운영하는 조직에서 가장 자주 마주치는 질문 중 하나는 "누가 어떤 권한을 가져야 하는가?"입니다. Databricks는 엔터프라이즈급 데이터 플랫폼으로, 다양한 관리자 역할을 제공하여 조직의 보안과 운영 효율성을 보장합니다.

각 역할의 책임 범위를 명확히 이해하지 못하면, 불필요하게 높은 권한을 부여하거나 반대로 필요한 권한이 부족하여 업무가 지연되는 문제가 발생할 수 있습니다. 이 글에서는 Databricks의 주요 관리자 역할들을 계층별로 정리하고, 각 역할이 담당하는 영역과 실무에서의 활용 방법을 설명합니다.

Databricks 관리자 역할의 계층 구조

Databricks의 관리자 역할은 크게 네 가지 계층으로 구분할 수 있습니다:

1. 계정 및 인프라 관리 수준: 클라우드 자원과 계정 전체를 관리
2. 워크스페이스 및 데이터 거버넌스 수준: 특정 워크스페이스와 데이터 카탈로그 관리
3. 특정 기능 관리 수준: 결제, 마켓플레이스 등 특화된 기능 관리
4. 소유권 개념: 개별 데이터 객체의 소유자

1. 계정 및 인프라 관리 수준

1) Cloud Administrator (클라우드 관리자)

주요 책임:

• 스토리지 계정(버킷) 및 클라우드 네이티브 자원 관리
• IAM 역할 및 서비스 주체(Service Principal) 설정
• 클라우드 서비스와 Databricks 간의 통합 구성

실무 관점: Cloud Administrator는 Databricks가 아닌 클라우드 플랫폼(AWS, Azure, GCP) 레벨에서 작업합니다. 예를 들어, Databricks가 S3 버킷에 접근하기 위한 IAM 역할을 생성하거나, Azure Storage Account에 대한 접근 권한을 설정하는 것이 이 역할의 주요 업무입니다.

언제 필요한가?

• Databricks 워크스페이스를 처음 생성할 때
• 외부 스토리지(데이터 레이크)와 Databricks를 연결할 때
• 클라우드 네이티브 서비스(예: AWS Glue, Azure Data Factory)와 통합할 때

2) Identity Administrator (ID 관리자)

주요 책임:

• 기업의 ID 공급자(IdP)를 Databricks와 통합
• 사용자 및 그룹을 계정에 자동 프로비저닝
• SSO(Single Sign-On) 설정 및 관리

실무 관점: 대규모 조직에서는 수백 명의 사용자를 수동으로 관리하기 어렵습니다. Identity Administrator는 SAML 또는 SCIM 프로토콜을 통해 기업의 Active Directory나 Okta 같은 IdP와 Databricks를 연동하여, 사용자 추가/삭제/권한 변경을 자동화합니다.

언제 필요한가?

• 10명 이상의 사용자가 있는 조직
• 기존 기업 인증 시스템과 통합이 필요한 경우
• 사용자 생명주기 관리를 자동화하고 싶을 때

3) Account Administrator (계정 관리자)

주요 책임:

• 메타스토어(Metastore) 생성 및 관리
• 워크스페이스 생성 및 삭제
• 계정 레벨의 사용자 및 그룹 구조 관리
• 스토리지 자격 증명(Storage Credentials) 생성
• 다른 사용자에게 계정 관리자 권한 부여

실무 관점: Account Administrator는 Databricks 계정의 최상위 관리자입니다. 이 역할은 조직 내에서 매우 제한적으로 부여되어야 하며, 보통 데이터 플랫폼 팀의 리더나 IT 관리자가 담당합니다. 모든 데이터 객체를 관리할 수 있는 권한을 가지므로, 신중하게 권한을 부여해야 합니다.

언제 필요한가?

• 새로운 워크스페이스를 생성해야 할 때
• Unity Catalog의 메타스토어를 설정할 때
• 계정 전체의 사용자 구조를 재구성할 때

2. 워크스페이스 및 데이터 거버넌스 수준

1) Workspace Administrator (워크스페이스 관리자)

주요 책임:

• 특정 워크스페이스 내의 자산 관리 (노트북, Repo, 클러스터, 작업 등)
• 워크스페이스 레벨의 사용자 추가/제거
• 클러스터 생성 정책 및 인스턴스 프로파일 설정
• 워크스페이스 설정 및 구성 관리

실무 관점: Workspace Administrator는 특정 워크스페이스의 "관리자"입니다. 예를 들어, "개발 워크스페이스"와 "프로덕션 워크스페이스"가 있다면, 각각 별도의 Workspace Administrator를 둘 수 있습니다. 이 역할은 개발자들이 필요한 리소스를 사용할 수 있도록 환경을 구성하고, 비용 관리를 위한 클러스터 정책을 설정합니다.

언제 필요한가?

• 워크스페이스 내에서 사용자 권한을 세밀하게 관리해야 할 때
• 클러스터 자동 종료 정책이나 인스턴스 타입 제한을 설정할 때
• 워크스페이스별로 독립적인 개발 환경을 운영할 때

2) Metastore Administrator (메타스토어 관리자)

주요 책임:

• Unity Catalog의 카탈로그(Catalog) 생성 및 관리
• 외부 위치(External Locations) 생성 및 관리
• 데이터 객체(테이블, 뷰 등)에 대한 권한 부여
• 데이터 객체의 소유권 변경
• 데이터 거버넌스 정책 수립 및 관리

실무 관점: Metastore Administrator는 데이터 거버넌스의 핵심 역할입니다. Unity Catalog를 통해 데이터 레이크의 모든 데이터에 대한 접근 제어를 관리합니다. 예를 들어, "마케팅 팀은 sales 데이터베이스의 특정 테이블만 읽을 수 있다"와 같은 정책을 설정하고 관리합니다.

언제 필요한가?

• Unity Catalog를 사용하여 데이터 거버넌스를 구현할 때
• 여러 워크스페이스에서 공유하는 데이터 카탈로그를 관리할 때
• 데이터 접근 권한을 세밀하게 제어해야 할 때

3. 특정 기능 관리 수준

1) Marketplace Administrator (마켓플레이스 관리자)

주요 책임:

• Databricks Marketplace의 공급자 프로필 생성 및 관리
• 데이터 제품 리스팅 생성 및 관리
• 데이터 공유를 위한 Share 생성 및 관리
• 공급자 콘솔(Provider Console) 접근 및 운영

실무 관점: Marketplace Administrator는 조직이 Databricks Marketplace를 통해 데이터를 공유하거나 판매할 때 필요한 역할입니다. 데이터 제품을 마켓플레이스에 등록하고, 다른 조직과 데이터를 공유하는 비즈니스를 관리합니다.

언제 필요한가?

• 조직의 데이터를 외부에 공유하거나 판매할 때
• Databricks Marketplace를 활용한 데이터 비즈니스를 운영할 때

2) Billing Administrator (결제 관리자)

주요 책임:

• 예산 조회 및 예산 정책 관리
• 구독 및 결제 수단(신용카드 등) 관리
• 사용량 대시보드 모니터링
• 예산 알림 설정 및 비용 최적화

실무 관점: Billing Administrator는 Databricks 사용 비용을 관리하는 역할입니다. 각 워크스페이스나 사용자 그룹별로 예산을 설정하고, 예산 초과 시 알림을 받아 비용을 제어합니다. 재무팀이나 IT 관리팀의 담당자가 주로 이 역할을 수행합니다.

언제 필요한가?

• Databricks 사용 비용을 모니터링하고 제어해야 할 때
• 팀별 또는 프로젝트별 예산을 관리할 때
• 비용 최적화를 위한 정책을 수립할 때

4. 소유권 개념: Owner (소유자)

정의: 메타스토어 내의 각 데이터 객체(테이블, 뷰, 함수 등)를 생성한 주체이거나, 소유권을 이전받은 주체입니다.

권한:

• 명시적인 권한 없이도 자신이 소유한 객체를 읽고 수정할 수 있음
• 타인에게 권한을 부여하거나 회수할 수 있음
• 하위 객체(예: 테이블의 컬럼)를 생성할 수 있음
• 객체의 소유권을 다른 사용자에게 이전할 수 있음

실무 관점: Owner는 역할(Role)이 아니라 객체별로 부여되는 개념입니다. 예를 들어, 데이터 엔지니어가 새로운 테이블을 생성하면 자동으로 그 테이블의 Owner가 됩니다. Owner는 해당 객체에 대한 완전한 제어권을 가지므로, 퇴사자나 역할 변경 시 소유권 이전을 고려해야 합니다.

5. 역할별 관리 대상 요약

6. 실무 활용 가이드

신규 프로젝트 시작 시 권한 구성 예시

시나리오: 중소규모 조직의 Databricks 도입

1. 초기 설정 단계

   • Cloud Administrator: 클라우드 자원 설정 (1-2명)
   • Account Administrator: 계정 및 첫 워크스페이스 생성 (1명)

2. 운영 단계

   • Identity Administrator: 사용자 자동 프로비저닝 설정 (1명)
   • Workspace Administrator: 각 워크스페이스별 관리자 (워크스페이스당 1-2명)
   • Metastore Administrator: 데이터 거버넌스 담당 (1-2명)
   • Billing Administrator: 비용 관리 담당 (1명)

3. 일반 사용자

   • 데이터 엔지니어: 테이블 생성 시 자동으로 Owner 권한 획득
   • 데이터 분석가: Metastore Admin이 부여한 읽기 권한으로 데이터 접근

권한 부여 시 주의사항

1. 최소 권한 원칙: 필요한 최소한의 권한만 부여
2. 역할 분리: Account Admin과 Workspace Admin을 분리하여 권한 집중 방지
3. 정기 검토: 분기별로 권한 목록을 검토하고 불필요한 권한 회수
4. 소유권 관리: 퇴사자나 역할 변경 시 데이터 객체의 소유권 이전

7. 마무리

Databricks의 관리자 역할 체계는 조직의 규모와 요구사항에 따라 유연하게 구성할 수 있습니다. 각 역할의 책임 범위를 명확히 이해하고, 조직의 구조에 맞게 권한을 부여하는 것이 안전하고 효율적인 Databricks 운영의 핵심입니다.

특히 계정 관리자(Account Admin)와 워크스페이스 관리자(Workspace Admin)의 차이, 그리고 메타스토어 관리자(Metastore Admin)의 데이터 거버넌스 역할을 이해하는 것이 중요합니다. 이러한 역할들을 적절히 조합하면, 보안을 유지하면서도 개발자들이 필요한 리소스에 자유롭게 접근할 수 있는 환경을 구축할 수 있습니다.

참고 자료:

• Databricks 공식 문서: Account Admin
• Databricks 공식 문서: Unity Catalog 관리

모든 워크플로우 자동화 도구가 그렇듯, Apache Airflow에서도 테스크(Task) 간에 데이터를 주고받아야 하는 상황이 빈번하게 발생합니다. Airflow에서는 이를 위해 XCom(Cross-Communication)이라는 메커니즘을 제공합니다.

XCom이란?

XCom은 Airflow의 메타데이터베이스에 저장되는 작은 데이터 조각으로, 태스크 간에 데이터를 공유할 수 있게 해줍니다. 예를 들어, 한 태스크에서 생성한 파일 경로, 처리된 데이터의 요약 정보, 또는 다음 태스크에 필요한 설정값 등을 전달할 수 있습니다.

XCom의 크기 제한

XCom을 사용할 때 주의해야 할 중요한 점은 데이터 크기 제한입니다. 표준 XCom 백엔드를 사용할 경우, XCom의 크기 제한은 사용 중인 메타데이터 데이터베이스에 따라 결정됩니다:

보시다시피, 특히 MySQL의 경우 64KB라는 매우 작은 제한이 있어 큰 데이터를 전달하기에는 부적합합니다. 만약 XCom을 통해 전달하려는 데이터가 메타데이터 데이터베이스의 크기 제한을 초과할 가능성이 있다면, Custom XCom 저장소를 구축하여 활용하는 것을 고려해야 합니다.

1. **context 파라미터 활용하기

전통적인 PythonOperator 방식에서 호출되는 함수에 **context를 인자로 받아 ti(Task Instance) 객체에 접근하는 방법입니다.

def push_to_xcom(**context):
    message = "사과"
    ti = context["ti"]

    ti.xcom_push(
        key='message',
        value=message
    )
    return message # 'return_value'라는 키로도 자동 저장됨 

def pull_from_xcom(**context):
    ti = context["ti"]
    xcom_value = ti.xcom_pull(
        task_ids='py1',
        key='message'
    )
    print("py1에서 전달받은 결과 : ", xcom_value)

• 장점: Airflow의 모든 컨텍스트 정보에 명시적으로 접근할 수 있습니다.
• 특징: ti.xcom_push를 사용할 때 고유한 key를 지정할 수 있습니다.

💡 잠깐! **context에는 무엇이 들어있을까요?

많은 분들이 **context를 프린트했을 때 쏟아지는 방대한 양의 메타데이터에 당황하곤 합니다. 이 데이터들은 현재 실행 중인 DAG과 태스크의 '상태 정보'입니다.

주요 항목들을 표로 정리하면 다음과 같습니다:

print(context) 실행 시 실제 출력 예시

{
    'ds': '2024-01-14',
    'ds_nodash': '20240114',
    'logical_date': DateTime(2024, 1, 14, 0, 0, 0, tzinfo=Timezone('UTC')),
    'dag': <DAG: xcom01_dag>,
    'ti': <TaskInstance: xcom01_dag.py1 [running]>,
    'run_id': 'scheduled__2024-01-14T00:00:00+00:00',
    'params': {},
    ... (중략) ...
}

따라서 **context를 사용한다는 것은, "Airflow가 태스크를 실행하면서 들고 있는 모든 보따리(메타데이터)를 다 넘겨줘!"라고 요청하는 것과 같습니다.

2. get_current_context() 사용하기

함수의 인자로 **context를 넘기지 않더라도, get_current_context()를 통해 현재 실행 중인 태스크의 컨텍스트를 동적으로 가져올 수 있습니다.

from airflow.sdk import get_current_context

def push_to_xcom():
    message = "사과"
    context = get_current_context()
    ti = context['ti']

    ti.xcom_push(key='message', value=message)
    return message

def pull_from_xcom():
    context = get_current_context()
    ti = context['ti']

    xcom_value = ti.xcom_pull(task_ids='py1', key='message')
    print("py1에서 전달받은 결과 : ", xcom_value)

• 장점: 함수의 시그니처를 깔끔하게 유지할 수 있습니다.

3. TaskFlow API에서 return 사용하기 (현대적인 방식)

Airflow 2.0 이상에서 권장되는 TaskFlow API를 사용하면, 복잡한 xcom_push/pull 코드 없이 함수의 return 값만으로 데이터를 전달할 수 있습니다.

@task(task_id="first")
def first_func(args):
    join_list = ' '.join(args)
    return join_list # 자동으로 XCom에 push됨

@task(task_id='second')
def second_func(message):
    # 인자로 넘겨받은 message는 이전 태스크의 return 값 (XCom pull)
    changed_list = '!' + message + '!'
    return changed_list

# DAG 내에서 호출
message = first_func(['FLOWER', 'AIRFLOW'])
second_func(message)

• 장점: Python 함수를 호출하듯 직관적으로 태스크 간 데이터 흐름을 정의할 수 있습니다.

4. TaskFlow API 옵션 활용 (multiple_outputs)

태스크가 여러 개의 결과값을 딕셔너리 형태로 반환할 때, multiple_outputs=True 옵션을 주면 각각의 키 값이 개별 XCom 항목으로 저장됩니다.

@task(task_id="first", do_xcom_push=True, multiple_outputs=True)
def first_func(args):
    join_list = ' '.join(args)
    return {"key1": join_list} # 'key1'이라는 이름으로 XCom에 저장됨

• 특징: 반환된 딕셔너리의 키를 통해 특정 데이터만 Pull 할 수 있어 관리가 용이합니다.

5. 다른 Operator에서 Jinja 템플릿 활용하기

Python이 아닌 다른 오퍼레이터(예: BashOperator)에서 XCom 데이터를 사용하고 싶을 때는 Jinja 템플릿 형식을 사용합니다.

t1 = PythonOperator(
    task_id="make_dirname",
    python_callable=make_dirname, # 내부에서 ti.xcom_push(key="dir_path", ...) 수행
)

t2 = BashOperator(
    task_id="make_dir",
    bash_command="mkdir -p {{ ti.xcom_pull(task_ids='make_dirname', key='dir_path') }}"
)

• 장점: 서로 다른 언어나 환경을 사용하는 오퍼레이터 간의 협업이 가능해집니다.

6. 요약

7. 참고 문서

• https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/xcoms.html
• https://www.astronomer.io/docs/learn/airflow-passing-data-between-tasks#intermediary-data-storage

데이터 플랫폼이 커질수록 가장 먼저 복잡해지는 것은 데이터 자체가 아니라 데이터의 관리 방식입니다.

여러 팀이 같은 데이터 레이크를 공유하고, 수많은 테이블·파일·모델이 쌓이기 시작하면 “이 데이터는 누가 만들었는가?”, “누가 접근할 수 있는가?”, “어디에 저장되어 있는가?” 같은 질문에 명확히 답하기 어려워지죠.😅

Databricks Unity Catalog는 이런 문제를 해결하기 위해 등장한 통합 데이터 거버넌스 계층입니다. 단순히 테이블 권한을 관리하는 기능을 넘어, 👉 데이터·파일·모델 전반에 대해 중앙 집중식으로 접근 제어, 감사, 메타데이터 관리를 제공합니다.

기존 Hive Metastore 기반 환경에서는 워크스페이스마다 메타스토어가 분리되거나, 클라우드 스토리지 접근 권한과 Databricks 권한이 이중으로 관리되는 경우가 많았습니다. Unity Catalog는 이를 Account 단위의 단일 Metastore 구조로 통합하여 보안과 운영 복잡도를 크게 줄였습니다.

또한 Unity Catalog는

• SQL 기반의 일관된 권한 모델
• 메이저 클라우드 벤더(AWS / Azure / GCP)에 독립적인 설계
• 테이블뿐 아니라 External Location, Volume, ML 모델까지 관리 가능

이라는 점에서, Databricks를 단순한 분석 도구가 아닌 엔터프라이즈 데이터 플랫폼으로 확장시키는 핵심 구성 요소라고 볼 수 있습니다.

이 글에서는 Unity Catalog의 핵심 개념과 구조를 중심으로, 왜 필요한지, 그리고 기존 방식과 무엇이 다른지 차근차근 살펴보겠습니다.

1. Unity Catalog 계층 구조

UC는 Metastore > Catalog > Schema > Table/Volume의 4단계 계층 구조를 가집니다. 특히 저장 위치(Storage Location)는 상위 계층에서 하위 계층으로 상속되는 구조를 가집니다.

1-1) Metastore

• UC의 최상위 컨테이너로, 모든 권한 관리와 메타데이터의 중심지입니다.
• Account Console 수준에서 생성하며, AWS/GCP/Azure 객체 저장소를 기본 위치로 사용합니다.
• 하나의 Metastore는 여러 Workspace에 연결될 수 있어 조직 전체의 통합 거버넌스를 가능하게 합니다.
• Workspace > Catalog Explorer > External Data > External Locations에서 연결된 저장소 정보를 확인할 수 있습니다.

  1-2) Catalog

• 데이터 자산을 그룹화하는 첫 번째 단위입니다.
• 카탈로그 생성 시 실제 데이터가 저장될 MANAGED LOCATION을 지정할 수 있습니다.
• 지정하지 않으면 Metastore의 경로에 저장됩니다.

  1-3) Schema (Database)

• 카탈로그 내의 하위 단위로 태이블, 뷰, 볼륨을 포함합니다.
• 스키마 생성 시 LOCATION을 지정하면 해당 경로가 하위 Managed 객체의 기본 경로가 됩니다.

  1-4) Objects (Table / Volume)

• 실제 데이터가 담기는 최종 단위입니다.
• 최하위 객체는 상위 스키마나 카탈로그에 설정된 위치를 따라가거나, 직접 LOCATION을 지정(External)할 수 있습니다.

2. Managed vs External 핵심 개념

UC에서 가장 중요한 차이는 데이터의 소유권(Lifecycle Management)입니다.

3. 시나리오별 물리 저장 & 삭제 규칙

스키마가 External(Location 지정)이라 하더라도 그 안의 객체가 Managed인 경우, UC의 관리 원칙이 우선 적용됩니다.

💡 핵심 포인트:

1. 테이블 선언이 Managed라면, 하위 저장소가 어디든 DROP 시 데이터는 삭제됩니다.
2. External Schema 내의 Managed Table은 스키마가 가진 LOCATION 경로 아래에 생성되더라도 'Managed' 특성상 삭제 권한이 UC에 있습니다.

4. DDL 명령어 레퍼런스

4-1) Catalog 생성

카탈로그 수준에서 격리된 저장 공간을 사용하고 싶을 때 MANAGED LOCATION을 사용합니다.

-- 1. 기본형 카탈로그 (메타스토어 저장소 상속)
CREATE CATALOG IF NOT EXISTS prod_catalog;

-- 2. 관리형 위치를 지정한 카탈로그 (분리된 버킷 사용)
CREATE CATALOG IF NOT EXISTS dev_catalog
MANAGED LOCATION 's3://my-dev-bucket/uc-managed/';

4-2) Schema 생성

• Schema는 Catalog 하위에 생성되며, 해당 Schema에서 생성될 Managed 객체들의 기본 저장 경로를 결정할 수 있습니다. ```sql
• • Managed Schema (상위 Catalog의 경로 상속) CREATE SCHEMA IF NOT EXISTS catalog.managed_schema;

-- External Schema (Managed 객체들이 저장될 특정 경로 지정) CREATE SCHEMA IF NOT EXISTS catalog.external_schema MANAGED LOCATION 'gs://my-bucket/external-schema-path/';

### 4-3) Table 생성
- **Managed Table**: `LOCATION`을 지정하지 않으면 상위 Schema/Catalog의 경로 하위에 데이터가 저장됩니다.
- **External Table**: `LOCATION`을 명시해야 하며, **External Location**으로 등록된 경로라면 어디든 저장 가능합니다. (반드시 상위 스키마 경로 아래일 필요는 없음)
```sql
-- Managed Table
CREATE TABLE catalog.schema.man_tbl (id INT, name STRING) USING DELTA;

-- External Table (External Location 등록 선행 필요)
CREATE TABLE catalog.schema.ext_tbl (id INT) 
LOCATION 'gs://my-bucket/data/ext_tbl/';

4-4) Volume 생성

• Managed Volume: UC가 관리하는 기본 경로에 파일이 저장됩니다.
• External Volume: 등록된 외부 경로를 직접 참조하여 비정형 데이터를 관리합니다. ```sql
• • Managed Volume (비정형 데이터용) CREATE VOLUME catalog.schema.man_vol;

-- External Volume (LOCATION 명시 및 External Location 등록 필요) CREATE EXTERNAL VOLUME catalog.schema.ext_vol LOCATION 'gs://my-bucket/files/ext_vol/';

```

5. 관리형 볼륨(Volume) vs 테이블(Table)

💡 권장 워크로드:

• 정형 데이터 및 분석용 지표 데이터는 Table 추천
• 머신러닝 모델 파일, 로그 파일, 원본(Raw) 파일 관리는 Volume 추천

6. 요약 및 주의사항

1. External Location 등록: External Table/Volume을 만들기 전, 클라우드 스토리지 경로가 UC에 External Location으로 등록되어 있어야 합니다.
2. Managed의 삭제 규정: Managed 타입은 "UC가 데이터의 생명주기를 책임진다"는 뜻이므로, DROP 시 데이터가 물리적으로 삭제됩니다. 단, 7일 이내라면 UNDROP 명령어로 복구가 가능합니다.
3. External의 복구: External 테이블은 DROP 시 데이터가 유지되므로 언제든 재연결이 가능하지만, UNDROP을 사용하면 테이블 권한 등 메타데이터까지 함께 복구됩니다.
4. 스키마 위치 상속: 스키마에 LOCATION을 주면 그 아래의 Managed 객체들은 부모 스키마의 경로를 기본값으로 사용합니다.

7. 설계 시 권장 사항(Best Practices)

1. 환경 분리: dev, staging, prod 카탈로그를 만들고 각각 다른 MANAGED LOCATION(S3 버킷 등)을 지정하여 물리적으로 데이터를 격리하세요.
2. Managed 우선: 특별한 이유(외부 시스템 공유, 기존 데이터 등)가 없다면 성능과 관리 편의성을 위해 Managed Table(Delta) 사용을 권장합니다.
3. Volume 활용: .csv, .json 원본 파일이나 머신러닝 모델, 로그 등 비정형 파일은 Table이 아닌 Volume으로 관리하여 보안과 추적성을 확보합니다.
4. 권한 최소화: EXTERNAL LOCATION을 직접 참조하는 권한은 데이터 엔지니어 등 관리자에게만 부여하고, 일반 분석가는 Managed Table을 통해서만 데이터에 접근하도록 설계하세요.

• 이전 글 'Google Cloud Pub/Sub 서비스의 핵심 개념과 실습 튜토리얼'에서는 Pub/Sub 서비스의 기초적인 내용을 살펴보았습니다. 이번 글에서는 Pub/Sub 토픽으로 전송된 메시지를 읽어 BigQuery 테이블에 직접 저장하는 방법을 다뤄보겠습니다.

🔹 1. BigQuery 테이블 생성

▪ 1) 단일 스키마

• 토픽으로 전송되는 메세지를 저장할 수 있는 빅쿼리 테이블을 생성합니다. 이 때 구독 유형 중 스키마 사용 안함 옵션을 선택하게 되면 data라는 컬럼 하나가 있는 테이블의 row에 토픽에서 읽은 메세지가 저장됩니다.

• BigQuery의 bq_data 데이터셋에 data 컬럼 하나만 있는 pubsub_tbl_simple 테이블을 생성합니다.

▪ 2) 커스텀 스키마

• 구독 유형에서 테이블 스키마 사용 옵션을 선택하면, 생성한 BigQuery 테이블의 스키마에 맞춰 토픽으로 데이터를 전송할 수 있고, 구독은 해당 데이터를 읽어 BigQuery 테이블에 적재합니다.
• 이번 실습에 사용할 테이블은 'Baby Names by year' 데이터셋이며, 스키마는 아래와 같습니다.(살짝 수정)

  id         int64
  name      object
  year       int64
  gender    object
  count      int64

• 실습을 위해 동일한 스키마를 가지는 pubsub_tbl_names BigQuery 테이블을 생성합니다.

🔹 2. 구독 생성

• 구독의 유형에는 아래 네가지 종류가 있습니다.
  • 가져오기 : 메세지를 읽어오기
  • 푸시 : 메세지를 다른 Endpoint URL로 전송
  • BigQuery에 쓰기 : 메세지를 BigQuery 테이블에 저장
  • Cloud Storage에 쓰기 : 메세지를 GCS 파일 객체로 저장

그 중 BigQuery에 쓰기 유형의 구독을 생성하여 토픽에 들어오는 데이터를 실시간으로 읽어와 BigQuery 테이블에 저장해보도록 하겠습니다. Google Pub/Sub 서비스의 토픽·발행자·구독자의 개념 및 토픽 생성 방법은 이전 글(링크)을 참고하시면 됩니다. 생성할 구독은 총 2가지로,

1) 단일 스키마 테이블인 pubsub_tbl_simple로 전송하는 구독 2) 커스텀 스키마 테이블인 pubsub_tbl_names로 전송하는 구독 이렇게 생성합니다.

▪ 1) 단일 스키마 테이블에 쓰기

• 위에서 생성한 테이블 중 data라는 컬럼 하나를 가진 pubsub_tbl_simple 테이블로 데이터를 전송하는 구독을 생성해보겠습니다.
  • 구독ID : dev_bq_subscription
  • 전송 유형 : BigQuery에 쓰기
  • 스키마 구성 : 스키마 사용 안함

▪ 2) 커스텀 스키마 테이블에 쓰기

• 5개의 컬럼을 가진 pubsub_tbl_names 테이블로 데이터를 전송하는 구독을 생성합니다.
  • 구독ID : dev_bq_names_subscription
  • 전송 유형 : BigQuery에 쓰기
  • 스키마 구성 : 테이블 스키마 사용

토픽에 들어온 데이터를 받아 각 테이블들로 저장해줄 구독 2개 생성이 완료되었습니다.

🔹 3. 실시간 메세지 전송

• 위에서 생성한 토픽과 구독, BigQuery 테이블의 관계는 아래와 같습니다.
• names 데이터셋을 JSON 형식으로 dev_topic 토픽에 전송하였을 때, 빅쿼리 테이블들에 어떤식으로 데이터가 적재되는지 확인해보도록 하겠습니다.

▪ 1) 토픽에 메세지 전송

• 파이썬 코드를 통해 dev_topic 토픽으로 데이터가 3초에 한 번씩 전송되도록 합니다.

import pandas as pd
from google.cloud import pubsub_v1
from google.oauth2 import service_account
import json, time
from faker import Faker

## 1. PubSub 토픽 관련 설정
PROJECT_ID = "[프로젝트 ID]"
KEY_PATH = "[서비스 계정 JSON KEY 경로]"
CREDENTIALS = service_account.Credentials.from_service_account_file(KEY_PATH)

TOPIC_ID = "dev_topic"
publisher = pubsub_v1.PublisherClient(credentials=CREDENTIALS)
TOPIC_PATH = publisher.topic_path(PROJECT_ID, TOPIC_ID)

# 2. 보낼 데이터셋을 읽어 JSON 형식으로 변환
df = pd.read_csv('names.csv')
df_dict = df.to_dict(orient='records')

# 3. dev_topic으로 데이터를 3초에 한 번씩 전송
for row in df_dict[:10]:
    future = publisher.publish(
        topic=TOPIC_PATH,
        data=json.dumps(row).encode("utf-8")
    )
    print(f"보낸 메시지: {row} / 결과: {future.result()}")
    time.sleep(3)


--- 출력 결과 ---
보낸 메시지: {'id': 1, 'name': 'Mary', 'year': 1880, 'gender': 'F', 'count': 7065} / 결과: 16021080392424521
보낸 메시지: {'id': 2, 'name': 'Anna', 'year': 1880, 'gender': 'F', 'count': 2604} / 결과: 16021545983015478
보낸 메시지: {'id': 3, 'name': 'Emma', 'year': 1880, 'gender': 'F', 'count': 2003} / 결과: 16022743523116683
보낸 메시지: {'id': 4, 'name': 'Elizabeth', 'year': 1880, 'gender': 'F', 'count': 1939} / 결과: 16022824001573059
보낸 메시지: {'id': 5, 'name': 'Minnie', 'year': 1880, 'gender': 'F', 'count': 1746} / 결과: 16022962397816275
보낸 메시지: {'id': 6, 'name': 'Margaret', 'year': 1880, 'gender': 'F', 'count': 1578} / 결과: 16022437774262030
보낸 메시지: {'id': 7, 'name': 'Ida', 'year': 1880, 'gender': 'F', 'count': 1472} / 결과: 16021982500229449
보낸 메시지: {'id': 8, 'name': 'Alice', 'year': 1880, 'gender': 'F', 'count': 1414} / 결과: 16022962752184659
보낸 메시지: {'id': 9, 'name': 'Bertha', 'year': 1880, 'gender': 'F', 'count': 1320} / 결과: 16022046604942106
보낸 메시지: {'id': 10, 'name': 'Sarah', 'year': 1880, 'gender': 'F', 'count': 1288} / 결과: 16022147167489566

▪ 2) 단일 스키마 테이블 확인

• pubsub_tbl_simple 테이블의 경우 전송한 JSON 형식의 데이터가 data라는 단일 컬럼 안에 STRING 형식으로 저장되는 것을 확인할 수 있습니다. 이렇게 스키마 사용 안함 유형의 구독은 추후 분석을 위해 테이블에 대한 추가적인 가공이 필요하며, 정형 데이터보다는 비정형이나 TEXT 형태의 데이터를 실시간으로 저장할 때 유용하게 쓰일 수 있을 것 같습니다.

▪ 3) 커스텀 스키마 테이블 확인

• pubsub_tbl_names 테이블의 경우 테이블 스키마에 맞게 데이터를 전송하면 각 컬럼에 해당 값이 실시간으로 저장되어 분석에 용이한 정형 데이터 테이블 형태로 관리할 수 있습니다.

🔹 4. OUTRO

• 이번 글에서는 토픽으로 받은 메시지를 BigQuery 테이블로 직접 적재하는 방식과 스키마 적용 여부에 따른 차이까지 살펴보았습니다. Pub/Sub의 경우 GCP의 서비스이기 때문에 GCS, 빅쿼리, Data Fusion 등 클라우드 내 서비스들과 원활하게 통합되어 사용된다면 훨씬 더 큰 시너지를 낼 수 있을 것이라 생각합니다.
• 앞으로는 Pub/Sub의 실시간성을 어떻게 데이터 파이프라인 아키텍처에 반영할지 고민해 보는 것이 중요할 것 같습니다.

• 현대 소프트웨어 아키텍처에서 시스템 간의 효율적인 통신은 필수적입니다. 특히 마이크로서비스 환경에서는 각 서비스가 독립적으로 동작하면서도 서로 간의 데이터를 주고받아야 하는 상황이 자주 발생합니다. 이 때 사용할 수 있는 개념으로 Queue 라는 자료구조가 있습니다. Queue를 통해 우리는 시스템 간의 느슨한 결합 가지는 아키텍처를 구성할 수 있습니다. 우리에게 익숙한 AWS 클라우드에서 가장 먼저 런칭된 서비스가 AWS SQS라고 하는 비동기 메세징 서비스였습니다. GCP에도 이와 동일한 서비스가 있는데 바로 'Pub/Sub'이라는 서비스입니다. 이번 글에서는 Google Cloud Pub/Sub 서비스의 개념에 대해 알아보고 python을 활용하여 어떻게 다룰 수 있는지 실습해보도록 하겠습니다.

🔹 1. Google Cloud Pub/Sub이란?

• Google Cloud Pub/Sub은 완전 관리형의 비동기 메시징 서비스로, 발행자(Publisher)와 구독자(Subscriber) 간의 메시지 전달을 할 수 있습니다. 전통적인 동기식 통신 방식과 달리, 메시지를 보내는 쪽과 받는 쪽이 서로를 직접 기다리지 않고도 데이터를 교환할 수 있게 하여 서비스간의 느슨한 결합을 가능하게 해줍니다.

▪ 핵심 개념 설명

토픽(Topic)

• 메세지가 저장되는 저장소로, 발행자가 보낸 메세지를 임시로 보관하는 역할을 합니다.
• 저장되는 메세지의 종류에 따라 여러 토픽 생성이 가능하며, GCP 프로젝트 내에서는 고유한 이름으로 식별되어야 합니다.

발행자(Publisher)

• 데이터를 생성하고, 이 생성된 데이터를 특정 토픽에 전송하는 역할을 하는 컴포넌트입니다.

구독자(Subscriber)

• 특정 토픽을 구독하여 발행자가 보낸 메세지를 수신하고 처리하는 역할을 담당하는 컴포넌트입니다.
• 단순히 메세지를 읽어올 수도 있고, 읽어온 메세지를 BigQuery나 GCS에 저장하도록 설정할 수도 있습니다.

발행자 - 토픽 - 구독자의 관계

🔹 2. Pub/Sub 실습 - UI

▪ 1) 토픽 생성

• 가장 먼저 메세지를 발행하여 저장할 토픽을 생성해줍니다. Pub/Sub → 주제 → +주제 만들기 탭에 들어가서 토픽의 이름만 설정해주면 바로 생성이 가능합니다.

▪ 2) 구독 생성

• 토픽을 생성하였다면 해당 토픽으로 들어오는 메세지를 받아서 소비할 구독을 생성해야 합니다.
• Pub/Sub → 구독 → +구독 만들기 탭에서 구독 생성이 가능합니다. 구독의 유형으로는 아래 4가지를 선택할 수 있습니다.
  • 가져오기 : 메세지를 읽어오기
  • 푸시 : 메세지를 다른 Endpoint URL로 전송
  • BigQuery에 쓰기 : 메세지를 BigQuery 테이블에 저장
  • Cloud Storage에 쓰기 : 메세지를 GCS 파일 객체로 저장(TEXT or AVRO 포맷)
• 구독 ID 와 메세지를 읽어올 토픽을 선택하고, 가장 기본이 되는 가져오기 유형으로 구독을 생성합니다.

▪ 3) 메세지 발행

• 위에서 생성한 토픽에서 테스트 메세지 생성이 가능합니다. 아래와 같이 토픽에 들어가서 메세지 → 메세지 게시 탭을 선택하고, 메세지 본분에 간략히 내용을 적어서 게시하면 해당 토픽에서 메세지를 발행할 수 있습니다.

▪ 4) 메세지 가져오기

• 토픽에서 발행한 메세지는 구독에서 가져올 수 있습니다. 위에서 dev_topic을 구독하는 dev_subscription에서 발행된 메세지 확인이 가능합니다.

🔹 3. Pub/Sub 실습 - Python

• 위에서 처럼 직관적인 GUI 환경에서 설정할 수도 있지만 Google의 Pub/Sub 관련 라이브러리를 설치하면 Python 코드로도 해당 기능 구현이 가능합니다.

▪ 1) 라이브러리 설치 및 기본 설정

• 라이브러리 설치 → pip install google-cloud-pubsub
• 아래 코드에서 GCP와 통신할 Client 및 토픽 객체를 생성합니다.

  from google.cloud import pubsub_v1
  from google.oauth2 import service_account
  

PROJECT_ID = "[프로젝트 ID]" KEY_PATH = "[서비스 계정 JSON KEY 경로]" CREDENTIALS = service_account.Credentials.from_service_account_file(KEY_PATH)

TOPIC_ID = "dev_topic" publisher = pubsub_v1.PublisherClient(credentials=CREDENTIALS) TOPIC_PATH = publisher.topic_path(PROJECT_ID, TOPIC_ID)

print(TOPIC_PATH)

--- 출력 결과 --- projects/codeit-hyunsoo/topics/dev_topic

### ▪ 2) 토픽 생성
- 이미 생성된 토픽이라면 pass 되도록 예외처리까지 포함하여 구성합니다.
```python
# Pub/Sub 토픽 생성 코드
from google.api_core.exceptions import AlreadyExists

try:
    topic = publisher.create_topic(name=TOPIC_PATH)
    print(f"토픽이 생성되었습니다: {topic.name}")
except AlreadyExists:
    print(f"이미 존재하는 토픽입니다: {TOPIC_PATH}")


--- 출력 결과 ---
토픽이 생성되었습니다: projects/codeit-hyunsoo/topics/dev_topic

▪ 3) 샘플 메세지 전송

• Faker 라이브러리를 활용해 랜덤한 문장 생성 후 dev_topic으로 전송하는 발행자를 만들어줍니다. (3초에 한 번씩 총 10개의 메세지 전송)

  from faker import Faker
  import time
  

fake = Faker()

for _ in range(10): msg = fake.sentence() future = publisher.publish( topic=TOPIC_PATH, data=msg.encode("utf-8"), source="app1" ) print(f"보낸 메시지: {msg} / 결과: {future.result()}") time.sleep(3)

--- 출력 결과 --- 보낸 메시지: Include business head send friend him final. / 결과: 15936265876075595 보낸 메시지: Check example left chance approach large. / 결과: 15936256995631331 보낸 메시지: Heart still property. / 결과: 15936267081923148 보낸 메시지: Can production admit with business moment future. / 결과: 15936297693105990 보낸 메시지: Reflect national government between bag part with mission. / 결과: 15936288400415598 보낸 메시지: Meeting drive anyone note. / 결과: 15936290797590549 보낸 메시지: Particularly fact see far election. / 결과: 15936291128627334 보낸 메시지: Hair energy tax whole model head hit. / 결과: 15936290594708674 보낸 메시지: Right range trade score half. / 결과: 15936291572828675 보낸 메시지: Effect where sign popular family media. / 결과: 15936285434903098

### ▪ 4) 구독 생성
- 구독 역시 토픽 생성과 비슷하게 서비스계정 JSON KEY를 기반으로 Client 및 구독 객체를 생성하고, `가져오기` 유형의 구독을 python으로 생성합니다.
```python
from google.cloud import pubsub_v1
from google.oauth2 import service_account
import time
from google.api_core.exceptions import AlreadyExists

PROJECT_ID = "[프로젝트 ID]"
KEY_PATH = "[서비스 계정 JSON KEY 경로]"
CREDENTIALS = service_account.Credentials.from_service_account_file(KEY_PATH)

# 토픽 정보
TOPIC_ID = "dev_topic"
publisher = pubsub_v1.PublisherClient(credentials=CREDENTIALS)
TOPIC_PATH = publisher.topic_path(PROJECT_ID, TOPIC_ID)

# 구독 정보
SUBSCRIPTION_ID = "dev_subscription"
subscriber = pubsub_v1.SubscriberClient(credentials=CREDENTIALS)
SUBSCRIPTION_PATH = subscriber.subscription_path(PROJECT_ID, SUBSCRIPTION_ID)

try:
    subscription = subscriber.create_subscription(
        name=SUBSCRIPTION_PATH,
        topic=TOPIC_PATH
    )
    print(f"구독이 생성되었습니다: {SUBSCRIPTION_PATH}")
except AlreadyExists:
    print(f"이미 존재하는 구독입니다: {SUBSCRIPTION_PATH}")


--- 출력 결과 ---
구독이 생성되었습니다: projects/codeit-hyunsoo/subscriptions/dev_subscription

▪ 5) 토픽에 저장된 메세지 읽기

• dev_topic에 저장된 메세지들은 구독을 통해 읽어올 수 있습니다. 아래 코드는 1초에 한 번씩 메세지를 확인하여 출력해줍니다.

  def callback(message):
    print(f"받은 메시지: {message.data.decode('utf-8')}")
    message.ack()
  

streaming_pull_future = subscriber.subscribe(SUBSCRIPTION_PATH, callback=callback) print(f"구독을 시작합니다: {SUBSCRIPTION_PATH}")

try: while True: time.sleep(1) except KeyboardInterrupt: streaming_pull_future.cancel() print("구독을 중단합니다.") finally: streaming_pull_future.cancel() subscriber.close()

--- 출력 결과 --- 구독을 시작합니다: projects/codeit-hyunsoo/subscriptions/dev_subscription 받은 메시지: This is Sample Message 구독을 중단합니다.

```

🔹 4. OUTRO

• 지금까지 Google Cloud Pub/Sub의 기본 개념부터 실제 구현까지 단계별로 살펴보았습니다. 토픽과 프로듀서, 컨슈머 개념 등 사용해보면서 Kafka와 상당히 비슷하다는 느낌을 받을 수 있었습니다.
• 다음 글에서는 토픽으로 들어온 메세지를 BigQuery와 GCS에 객체로 저장하는 방법에 대해 다뤄보도록 하겠습니다.

• 요즘 데이터 엔지니어링의 핵심 트렌드 중 하나는 단연 레이크하우스(Lakehouse)입니다. Apache Iceberg, Delta Lake, Apache Hudi 등 오픈 테이블 형식(Open Table Format)의 등장으로, 파일 기반 데이터 레이크에서도 ACID 트랜잭션, 스키마 진화, 타임 트래블과 같은 데이터베이스급 기능을 사용할 수 있게 되었죠. 하지만 이러한 솔루션들은 복잡한 JSON/Avro 기반의 메타데이터 시스템을 사용하기 때문에, 작은 변경사항 처리나 트랜잭션 관리에서 한계를 드러내기도 합니다.
• 이러한 문제를 해결하고자 DuckDB 팀은 DuckLake라는 새로운 오픈 테이블 형식을 제안했습니다. 이는 기존 레이크하우스와 유사한 구조를 가지면서도, 핵심적인 차별점은 모든 메타데이터를 표준 SQL 데이터베이스에 저장한다는 점입니다.
• 이번 글에서는 기존의 솔루션들과 차별화되는 DuckLake의 핵심적인 특징들을 살펴보고 python을 이용해 어떻게 ducklake로 관리되는 환경을 구성할 수 있는지에 대한 실습까지 다뤄보도록 하겠습니다.

🔹 1. DuckLake의 핵심 특징

• DuckLake는 메타데이터를 데이터베이스에 저장함으로써, 파일 기반 레이크하우스가 안고 있던 복잡성과 성능 병목 문제를 해결하고 더 단순하고 유연한 대안을 제공합니다.

1) 성능 최적화

• 메타데이터 쿼리가 단일 SQL로 처리되어 빠름
• 수많은 HTTP 파일 요청 대신 메타데이터 인라인(inline) 저장
• 서브 밀리초 단위의 쓰기 및 작은 파일 문제 해결

2) 강력한 일관성과 트랜잭션 지원

• 관계형 DB의 ACID 속성 활용
• 다중 테이블 트랜잭션 및 트랜잭션 DDL 지원

3) 유연한 메타데이터 및 저장소 구성

• PostgreSQL, MySQL, DuckDB 등 다양한 DB 지원
• S3, GCS, 로컬 디스크 등 다양한 저장소와 호환

4) 스냅샷 및 보안 기능 내장

• 데이터베이스 테이블 기반으로 수백만 개의 스냅샷을 효율적 관리
• 기본 내장 암호화로 지속적인 검증(Zero Trust) 환경 구현

🔹 2. 설치 및 환경 구성

• DuckLake를 활용하기 위해서는 DuckDB가 설치되어 있어야 하고, 메타 테이블이 저장될 데이터베이스가 준비되어 있어야 합니다.
• DuckDB의 경우 pip으로 간단하게 설치하여 파이썬에서 활용이 가능하며, 메타 DB의 경우 Docker Compose를 이용하여 컨테이너 환경에서 운영될 수 있도록 구성할 것입니다.

1) DuckDB 및 관련 패키지 설치

• DuckDB 설치 → 공식 문서
• 파이썬으로 간단히 설치 → pip install duckdb (반드시 1.3 버전 이상 설치)
• DuckDB에서 Extension들을 사용하기 위해서는 각 패키지들에 대한 설치를 해야합니다. 이번 실습에서는 아래 내용들에 대한 설치가 필요합니다.

  import duckdb as dd
  

In Memory 상태에서 데이터베이스 연결 생성

mem_con = dd.connect()

Extension들 설치

mem_con.execute("INSTALL postgres") mem_con.execute("INSTALL mysql") mem_con.execute("INSTALL ducklake") mem_con.execute("INSTALL httpfs")

### 2) Meta DB 구성(Docker Compose)
- DuckLake 메타 테이블들이 저장될 데이터베이스는 현재까지는 아래 4가지를 사용할 수 있습니다.
  - `DuckDB`, `Sqlite`, `PostgreSQL`, `MySQL`
- 이번 실습에서는 `PostgreSQL`과 `MySQL`을 각각 활용해보도록 하겠습니다.
- 아래 Docker Compose 파일을 실행해주면 `PostgreSQL`, `MySQL` 컨테이너가 생성됩니다.
```yaml
services:
  mysql:
    image: mysql
    container_name: mysql_db
    ports:
      - "3306:3306"
    environment:
      - MYSQL_ROOT_PASSWORD=123456
      - MYSQL_USER=hyunsoo
      - MYSQL_PASSWORD=velog
      - MYSQL_DATABASE=ducklake_catalog
  postgres:
    image: postgres:16
    container_name: postgres_db
    ports:
      - "5432:5432"
    environment:
      - POSTGRES_USER=hyunsoo
      - POSTGRES_PASSWORD=velog
      - POSTGRES_DB=ducklake_catalog

실행 명령 → docker compose up -d

🔹 3. DuckLake 연동

• DuckDB 설치와 데이터베이스 컨테이너 생성이 완료되었다면 기본적인 준비는 다 되었습니다. 이제부터는 아래 두 과정을 추가적으로 진행해주고 본격적인 연동 작업을 진행하면 됩니다.
  • 1) 데이터가 저장될 클라우드 객체 저장소와의 자격 증명 내용 저장
  • 2) 메타 데이터가 저장될 데이터베이스 연결 내용 저장

1) 객체 저장소 SECRET 생성

• GCS(Google Cloud Storage)의 경우 아래 공식 문서를 참고하여 'HMAC Key' 발급 후 SECRET 생성이 가능합니다.

  • DuckDB-GCS 공식문서

    secret_gcs = """
    CREATE SECRET (
    TYPE GCS,
    KEY_ID 'ABCDE',
    SECRET 'ABCDESECRET'
    );
    """
    mem_con.execute(secret_gcs)

• AWS S3의 경우 IAM 장기 자격증명 발급 후 SECRET 생성이 가능합니다.

  • DuckDB-S3 공식문서

    secret_s3 = """
    CREATE SECRET (
    TYPE s3,
    PROVIDER config,
    KEY_ID 'AKIAIOSFODNN7EXAMPLE',
    SECRET 'wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY',
    REGION 'ap-northeast-2'
    );
    """
    mem_con.execute(secret_s3)

• 생성한 SECRET 목록 확인

  mem_con.execute("FROM duckdb_secrets()").df()

  2) Meta DB SECRET 생성

• DuckLake의 메타 테이블 저장소로 사용되는 PostgreSQL과 MySQL 역시 SECRET으로 등록해주어야 합니다.

• 이번 글에서는 데이터베이스 각각에 대해서 연결하는 코드를 모두 작성하였지만 실제로 DuckLake 사용을 위해서는 하나의 데이터베이스만 선택해서 사용하면 됩니다.

PostgreSQL 등록 → 공식 문서

pg_conn = """
CREATE SECRET (
    TYPE postgres,
    HOST '127.0.0.1',
    PORT 5432,
    DATABASE ducklake_catalog,
    USER 'codeit',
    PASSWORD 'sprint'
);
"""
mem_con.execute(pg_conn)

MySQL 등록 → 공식 문서

mysql_conn = """
CREATE SECRET (
    TYPE mysql,
    HOST '127.0.0.1',
    PORT 3306,
    DATABASE ducklake_catalog,
    USER 'codeit',
    PASSWORD 'sprint'
);
"""
mem_con.execute(mysql_conn)

3) DuckLake 연동

• ATTACH 명령어를 통해 DuckLake 설정이 가능하며, DATA_PATH 다음에는 DuckLake에서 관리되는 데이터가 저장될 경로를 넣어주시면 됩니다. 아래 예시와 같이 로컬 경로, 클라우드 스토리지 경로 등 사용 환경에 맞게 설정이 가능합니다.
  • 로컬 경로 예시 : /home/ducklake/data/
  • AWS S3 예시 : s3://aws_ducklake_bucket/data_dir/
  • GCS 예시 : gs://gcs_ducklake_bucket/data_dir/
• 아래 실습에서는 Google Cloud Storage의 gs://hyunsoo_de_bucket/ducklake/postgres/ 경로를 데이터 저장소로 사용해보겠습니다.
• 형식은 아래와 같습니다.

  ATTACH 'ducklake:<연결 DB 종류>:dbname=<데이터베이스 이름>'
  AS <생성할 DB 이름>(DATA_PATH <객체 저장소 경로>);

PostgreSQL을 메타 DB로 사용하는 경우

ducklake_conn = """
ATTACH 'ducklake:postgres:dbname=ducklake_catalog' 
AS my_ducklake(DATA_PATH 'gs://hyunsoo_de_bucket/ducklake/postgres/');
"""
mem_con.execute(ducklake_conn)

MySQL을 메타 DB로 사용하는 경우

ducklake_conn = """
ATTACH 'ducklake:mysql:db=ducklake_catalog' 
AS my_ducklake(DATA_PATH 'gs://hyunsoo_de_bucket/ducklake/mysql/');
"""
mem_con.execute(ducklake_conn)

• 위의 코드가 에러 없이 잘 실행이 되었다면 DuckLake 사용환경 설정이 완료가 되었습니다. 선택한 데이터베이스의 ducklake_catalog DB를 조회해보면 아래와 같이 DuckLake 관련 메타 테이블들이 세팅되어 있는 것을 확인할 수 있습니다.

🔹 4. 데이터 조회 및 테이블 생성

1) Cloud 객체 저장소 데이터 조회

• 등록한 객체 저장소 SECRET을 활용해 객체 저장소에 있는 파일(JSON, parquet, CSV 등) 데이터를 DuckDB로 바로 조회할 수 있습니다.

  _query = """
  FROM read_parquet('gs://hyunsoo_de_bucket/dataset/emp.parquet')
  """
  

mem_con.execute(_query).df()

### 2) 데이터베이스 조회
- 데이터베이스를 조회해보면 DuckLake로 관리되는 데이터베이스가 목록에 포함되어 있는 것을 확인할 수 있습니다.
```python
mem_con.execute("SHOW DATABASES").df()

3) DuckLake에 테이블 저장

• 먼저 위에서 생성한 my_ducklake 데이터베이스를 선택한 후 CREATE TABLE ~ AS 쿼리를 통해 기존에 있는 테이블을 기반으로 새로운 테이블을 생성합니다.

## my_ducklake 데이터베이스 사용 설정
mem_con.execute("USE my_ducklake")

## 새로운 테이블 생성
create_table = """
CREATE TABLE duck_emp
AS 
SELECT * FROM read_parquet('gs://hyunsoo_de_bucket/dataset/emp.parquet')
"""

mem_con.execute(create_table)

• 테이블 생성 후 등록한 DATA_PATH 경로를 확인해보면 main/<테이블 이름> 디렉토리가 새롭게 생성되면서 데이터가 parquet 파일 형태로 저장이 된 것을 확인할 수 있습니다. (‼️ 위 저장되는 디렉토리 구조는 MacOS 기준이며, 실습 결과 Windows와 Linux 플랫폼에서는 디렉토리 구조가 약간 달랐습니다. ‼️)

• 또한 메타 테이블의 ducklake_table을 조회해보면 아래와 같이 방금 생성한 duck_emp 테이블을 확인할 수 있습니다.

🔹 5. 버전 확인 및 Time Travel

1) 테이블 수정

• Time Travel 쿼리를 작성하기에 앞서 데이터를 수정해보도록 하겠습니다.

  ## 1) 새로운 컬럼 추가(new_col)
  mem_con.execute("ALTER TABLE duck_emp ADD COLUMN new_col INT DEFAULT 0")
  

2) UPDATE

mem_con.execute("UPDATE duck_emp SET new_col = 1 WHERE Salary >= 100000")

3) DELETE

mem_con.execute("DELETE FROM duck_emp WHERE Salary < 50000")

- 테이블 수정 발생시 연결된 객체 저장소에도 parquet 파일이 추가적으로 생성되어 아래와 같이 확인이 가능합니다.
![](https://velog.velcdn.com/images/newnew_daddy/post/da14f06b-4666-4cc5-bf35-01e35f64f5e4/image.png)

### 2) 테이블 버전 확인
- 테이블 수정 내용 및 버전 확인을 위해서는 메타테이블의 `ducklake_snapshot`, `ducklake_snapshot_changes` 테이블을 조회해보면 됩니다.
#### ducklake_snapshot 테이블
![](https://velog.velcdn.com/images/newnew_daddy/post/7cf66df8-a9c0-4d64-b7c7-884561f7c939/image.png)

#### ducklake_snapshot_changes 테이블
![](https://velog.velcdn.com/images/newnew_daddy/post/54fc4b72-acd1-4579-a358-abf968f9d8dd/image.png)


- 또한 아래 파이썬 코드를 통해서도 확인이 가능합니다.
```python
mem_con.execute("FROM my_ducklake.snapshots()").df()

3) Time Travel 쿼리

• Time Travel 관련 공식 문서 내용

버전 기반 Time Travel

• 테이블 스냅샷 기준 3버전 상태로 되돌립니다.

  mem_con.execute("SELECT * FROM duck_emp AT (VERSION => 3)").df()

시간 기반 Time Travel

• 2025-07-04 05:43:49.371+00 → 이 시점에서의 테이블 데이터를 보여줍니다.

  mem_con.execute("SELECT * FROM duck_emp AT (TIMESTAMP => '2025-07-04 05:43:49.371+00')").df()

🔹 6. OUTRO

• 기존에는 Iceberg나 Delta Lake 같은 Open Table Format(OTF)을 사용할 때, 그 구조에 큰 의문을 가지지 않았습니다. 파일 단위로 저장된 데이터가 ACID 트랜잭션, 롤백, 그리고 Time Travel까지 지원된다는 점이 그저 신기하게 느껴졌죠. 동일한 경로에 메타데이터가 파일 형식으로 저장된다는 것도 "이력 관리를 위해서는 당연히 이렇게 저장이 되어야지!" 하고 넘어갔습니다.
• 하지만 DuckLake를 접하고 기존 도구들과의 차이점에 대해 알게 되면서 이런 생각이 들었습니다.
• "메타데이터를 꼭 파일로만 저장해야 할까? 왜 DB로 관리할 수 있다고 생각을 못했을까?"* DuckLake처럼 메타데이터를 데이터베이스에 저장하면, 이력 조회나 관리 작업이 훨씬 직관적이고 유연해질 수 있겠다는 가능성을 느꼈습니다.
• 물론 아직까지는 기존 OTF만큼 성숙한 생태계를 갖췄다고 보긴 어렵습니다. 하지만 DuckDB의 빠른 발전 속도와 함께 DuckLake 역시 함께 성장한다면, 향후에는 널리 사용되는 차세대 테이블 포맷으로 자리 잡을 수도 있겠다는 생각을 하게 되었습니다.

🔹 7. 참고 자료

• DuckLake 공식 문서
• DuckDB DuckLake Extensions
• Understanding DuckLake: A Table Format with a Modern Architecture

UV 공식 문서

• Rust로 작성된 고성능 Python 패키지 및 프로젝트 관리 도구로 pip, virtualenv, poetry 등을 대체할 수 있는 빠르고 효율적인 패키지 관리 도구입니다.

• 흔히 사용되는 pip이나 poetry 같은 다른 도구들에 비해 패키지 설치가 엄청나게 빠른 것이 특징입니다.

• uv init만으로 가상환경이 자동으로 구성되어, 별도의 Python 가상환경 설치 없이 즉시 개발을 시작할 수 있습니다.

• pyproject.toml, uv.lock 파일을 활용해 의존성 명시 및 재현 가능한 환경 구성 보장합니다.

🔹 2. UV 설치 및 초기 설정

1) 설치

• pip 설치가 되어 있지 않은 경우

  # macOS/Linux
  curl -LsSf https://astral.sh/uv/install.sh | sh
  

Windows (PowerShell)

powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"

- pip 설치가 되어 있는 경우
```shell
pip install uv

• 설치 확인

  uv --version
  

uv 0.7.14 (e7f596711 2025-06-23)

### 2) 작업 디렉토리 초기 설정
Git에서 `git init` 명령어를 통해 디렉토리를 Git으로 관리하겠다고 선언하듯, uv 역시 `uv init` 명령어를 통해 작업 디렉토리를 초기화해야 uv 기능을 사용할 수 있습니다.
즉, uv를 사용하기 위해서는 최초에 `uv init` 명령어로 해당 디렉토리를 설정해주는 과정이 필요합니다.
```shell
# 작업 디렉토리 설정 초기화
uv init

# 디렉토리 생성 + 초기화
uv init my-project

• .python-version: 프로젝트에서 사용할 Python 버전이 명시된 파일입니다.
• main.py: 실행 또는 테스트를 위한 기본 Python 스크립트 파일입니다.
• pyproject.toml: 프로젝트의 메타데이터와 빌드 설정, 의존성 등을 정의하는 표준 구성 파일입니다.
• README.md: 프로젝트의 목적과 사용법 등을 설명하는 문서 파일입니다.

🔹 3. 패키지 관리 및 실행

1) 패키지 설치 & 삭제

• 기존에 pip install 명령어를 통해 설치했던 파이썬 라이브러리들 역시 uv에서는 uv 문법에 맞게 설치 및 삭제를 진행해주어야 합니다.

  # 패키지 추가
  uv add [패키지 명]
  uv add requests pandas faker
  

패키지 제거

uv remove [패키지 명]

- `requirements.txt` 파일에 정리된 라이브러리들을 한 번에 설치하고자 한다면 아래 명령어로 가능합니다.
```shell
uv pip install -r requirements.txt

• 추가된 패키지는 pyproject.toml 파일의 dependencies 항목에 저장되며, 삭제시엔 해당 항목에서 사라집니다.

2) python 파일 실행

• uv로 패키지 관리가 되는 디렉토리의 python 파일을 실행할 경우 uv run 명령어를 활용하는 것이 가장 좋습니다.
• 물론 python 명령어를 통해 실행하는 것도 가능하지만 터미널에서 가상환경을 따로 활성화시킨 후 실행을 해야합니다.

uv를 통한 실행

python을 통한 실행 (가상환경 활성화 + 실행)

• uv로 생성한 가상환경 활성화시 uv init 때 명시한 작업 디렉토리 이름으로 가상환경이 활성화됩니다.

🔹 4. Python 버전 변경

• pip을 사용하는 경우, 중간에 가상환경의 Python 버전을 변경하려면 기존 가상환경 삭제 → 새로운 버전의 가상환경 생성 → 라이브러리 설치 → python 파일 실행 이 순서로 작업이 진행되어야 합니다.
• 특히 기존 가상환경이 무겁거나 복잡하게 구성된 경우, 삭제하거나 새로 설정하는 데 시간이 오래 걸려 Python 버전 변경 작업이 매우 번거롭고 귀찮게 느껴질 수 있습니다.
• 하지만 uv를 사용하는 경우, 작업 디렉토리 내의 .python-version과 pyproject.toml 파일만 수정하면, 기존 디렉토리를 그대로 유지한 채 가상환경의 Python 버전만 간편하게 변경할 수 있습니다.

  .python-version

  

pyproject.toml

• 위 두 파일에서 Python 버전 변경 후 uv run main.py 명령어를 치게되면 아래의 과정을 자동으로 수행하게 됩니다.
  • 기존 가상환경 제거 → 새로운 가상환경 생성 → 패키지 설치 → 실행

🔹 5. pip vs uv

🔹 6. OUTRO

• 기존에는 대부분의 Python 프로젝트에서 pip과 venv 조합을 사용했고, 지금도 많이 사용되고 있습니다. 하지만 느린 속도, Python 버전 변경의 어려움 등 pip이 가지는 명확한 한계점이 있었습니다.
• uv는 이를 보완하고 Python의 패키지 관리를 더욱 효율적이고 빠르게 할 수 있도록 개발된 도구입니다. 특히 Python 버전 관리나 가상환경 설정, 의존성 설치 속도 등에서 uv는 매우 강력한 이점을 제공합니다.
• uv는 독자적인 디렉토리 구조를 가지고 있지만 기존 pip로 설치된 디렉토리와도 높은 호환성을 유지해, 기존 프로젝트를 완전히 갈아엎지 않아도 자연스럽게 도입할 수 있습니다. 이처럼 부담 없이 도입이 가능하면서도 더 나은 개발 환경을 제공한다는 점에서 매우 매력적입니다.
• 최근에는 MCP를 비롯해 다양한 오픈소스 프로젝트와 팀들이 uv를 표준 툴로 채택해가고 있는 추세입니다.
• 이제 Python 프로젝트를 시작하거나 기존 환경을 개선할 계획이 있다면, uv를 적극 고려해보세요. 빠르고 가벼우며, 무엇보다 개발자의 귀찮음을 줄여주는 도구입니다.

• Airflow의 XCom(Cross Communication)은 DAG 내의 태스크 간 데이터를 공유하기 위한 기능입니다. Airflow 시스템이 실행되면 기본 메타데이터 저장소인 PostgreSQL에 다양한 메타 정보가 저장되며, XCom 데이터 역시 이 PostgreSQL 데이터베이스에 기록됩니다.
• 하지만 XCom은 어디까지나 경량 메시지 전달 용도로 설계된 기능이기 때문에, 사용 시 몇 가지 권장사항을 반드시 고려해야 합니다.
  • str, int, float, bool, list, dict 등의 단순하고 직렬화 가능한 데이터 타입만 저장
  • 수 MB 이하 수준의 작고 가벼운 데이터만 저장
  • 실제 데이터 전달보다는 상태 또는 신호 전달 용도로 사용 (경로, 처리 여부, 간단한 통계치 등)
• Python 함수 간 데이터 전달은 같은 세션의 메모리를 공유하기 때문에 pandas.DataFrame, numpy.ndarray, 이미지 등 대용량 복합 객체도 무리 없이 전달이 가능합니다. 하지만 XCom은 메타DB를 공유하므로, 대용량 데이터를 그대로 저장하면 DB 성능과 안정성에 악영향을 줄 수 있습니다.
• 데이터 파이프라인에서는 함수 간 DataFrame 객체 전달이 자주 발생하는데 이런 경우는 어떻게 해야 할까요? Airflow는 이를 위해 Xcom 데이터를 PostgreSQL이 아닌 AWS S3, Google GCS, Azure Blob Storage 같은 클라우드 객체 저장소에 저장할 수 있는 XComObjectStorageBackend 설정을 제공합니다.
• 이번 글에서는 Airflow 3.0 버전에서 Xcom을 AWS S3와 Google Cloud Storage로 설정하는 방법을 각각 다뤄보도록 하겠습니다. (Airflow 3.0 버전을 Docker Compose로 설치 후 진행하므로 설치 관련해서는 공식 문서 참고 바랍니다.)

  1. Airflow 3.0 설정 파일 확인

• Airflow에는 각종 세팅들을 저장해놓는 airflow.cfg 파일이 있습니다. 2.10 버전까지는 Airflow의 도커 컨테이너 내부 /opt/airflow/airflow.cfg 경로에 존재했었는데 3 버전이 되면서 /opt/airflow/config/airflow.cfg 경로로 위치가 바뀌었습니다.
• 사용자의 로컬 디렉토리와 볼륨 매핑이 되어있는 경로로 위치가 바뀌어 예전이라면 도커 컨테이너 내부에 들어가서 수정해야 했었던 것이 이제 로컬 디렉토리에서 바로 수정이 가능해졌습니다.
• XComObjectStorageBackend 설정시 airflow.cfg 파일을 수정하고 docker compose restart 명령으로 시스템 재시작을 해주어야 제대로 적용이 됩니다.

2. Connections 설정

• Airflow UI에서 AWS와 GCP에 대한 connections 설정을 해주도록 하겠습니다. (Admin > Connections)

  1) AWS Connection

• Connection Type : aws
• AWS Access Key ID : IAM USER로 발급받은 Key의 ID 부분
• AWS Secret Access Key : IAM USER로 발급받은 Key의 Secret 부분
• Extra Fields : {"region_name" : "ap-northeast-2"}
• 

2) GCP Connection

• Connection Type : google_cloud_platform
• Project Id : GCP 프로젝트 ID
• Keyfile Path : 서비스 계정 JSON 키 경로

3. DAG 작성

• 실습 DAG의 경우 아래 두 가지 Task로 구성되어 있습니다.
  • 1번 Task(upload_df) > 컬럼 4개, 행 10개짜리 샘플 DataFrame을 생성하여 이를 Xcom에 저장
  • 2번 Task(read_df) > Xcom에 저장된 DataFrame을 읽어와 로그에 출력

    import pendulum
    from datetime import timedelta
    import datetime
    from airflow.sdk import DAG, task, get_current_context, ObjectStoragePath, Param
    from airflow.operators.python import PythonOperator
    import pandas as pd
    

샘플 데이터프레임 생성

def create_sample_df(): data = { 'name': ['John', 'Emma', 'Michael', 'Sarah', 'David', 'Lisa', 'James', 'Emily', 'Daniel', 'Anna'], 'age': [25, 32, 28, 35, 41, 29, 33, 27, 38, 31], 'city': ['Seoul', 'Boston', 'London', 'Paris', 'Tokyo', 'Sydney', 'Berlin', 'Toronto', 'Rome', 'Madrid'], 'score': [85, 92, 78, 95, 88, 83, 91, 87, 94, 89] } return pd.DataFrame(data)

with DAG( dag_id="xcom_backend_dag", schedule="@once", start_date=pendulum.datetime(2025, 6, 1, tz="Asia/Seoul"), catchup=False ) as dag: ## DataFrame을 Xcom에 저장 @task(task_id='upload_df') def upload_df(): df = create_sample_df() return df

## 저장된 Xcom을 가져와 로그에 출력
@task(task_id='read_df')
def read_df():
    context = get_current_context()
    df = context['task_instance'].xcom_pull(task_ids='upload_df')
    print(df.head())
    return df

upload_df() >> read_df()

## 4. airflow.cfg 파일 수정 후 적용
### 1) AWS S3
- AWS S3를 Xcom Backend로 설정하고자 하는 경우, `airflow.cfg` 파일의 내용을 아래와 같이 설정하면 됩니다.

xcom_backend = airflow.providers.common.io.xcom.backend.XComObjectStorageBackend

예시) xcom_objectstorage_path = s3://aws_connection@airflow-hyunsoo-bucket/xcom

xcom_objectstorage_path = s3://[AWS Connection 명]@[S3 버킷 경로]

xcom_objectstorage_threshold = 0

xcom_objectstorage_compression = gzip

- `xcom_objectstorage_path` : Xcom 데이터가 저장될 객체 스토리지 경로 지정
- `xcom_objectstorage_threshold` : Xcom 데이터가 객체 스토리지에 저장되지 위한 최소값 지정
  - `threshold 값 미만` : 메타DB에 저장
  - `threshold 값 이상` : 객체 스토리지에 저장
  - `-1` : 항상 메타DB에 저장
  - `0` : 항상 객체 스토리지에 저장
- `xcom_objectstorage_compression` : Xcom 데이터의 압축 방식을 지정
### 2) Google Cloud Storage
- 위 설정과 비슷하며 `xcom_objectstorage_path`만 GCS 경로로 바꾸면 됩니다.

xcom_backend = airflow.providers.common.io.xcom.backend.XComObjectStorageBackend

예시) xcom_objectstorage_path = gs://my_gcp_conn@codeit_sprint/xcom

xcom_objectstorage_path = gs://[GCP Connection 명]@[GCS 버킷 경로]

xcom_objectstorage_threshold = 0

xcom_objectstorage_compression = gzip

```

3) 설정 내용 적용

• airflow.cfg 파일의 위 4가지 Key값을 설정하였다면 시스템 재시작이 필요합니다. docker compose로 설치되어 있다면 docker compose restart 명령어를 통해 재시작해주세요.
• 재시작 후 위의 DAG를 실행하면 아래와 같이 객체 저장소에 버킷 경로/[DAG ID]/[작업 시간]/[TASK ID]/[Xcom 데이터]와 같이 디렉토리 구조가 생성되며 Task에서 return한 데이터가 저장됩니다.
  • AWS S3
  • Google Cloud Storage

5. 참고 자료

• S3에 XCom 저장하기: XComObjectStorageBackend 설정 가이드
• Set up a custom XCom backend using object storage

• 이번 블로그 글에서 다뤄볼 내용은 Airflow Dag 내의 Task 실행시 선행 작업과 후행 작업을 설정할 수 있는 setup 과 teardown 기능입니다.
• 예를 들어 아래와 같은 흐름의 파이프라인을 구성해보겠습니다. 1) MySQL 데이터베이스에 연결 2) 테이블 데이터 조회 후 집계 테이블로 변환하여 저장 3) 데이터베이스 연결 종료 여기서 실질적인 작업을 하는 Task는 2번이고 1,3번은 2번 Task 앞뒤에 따라서 실행되는 작업입니다.
• 이런 경우, 1~3까지의 로직을 각각의 Task로 구성해도 상관없지만 setup 과 teardown 기능을 활용한다면 2번 작업에 대한 선행, 후행 작업으로 선언하는 것이 이후에 이어질 파이프라인 작업을 선언할 때도 훨씬 직관적으며, 이후 Clear 같은 Task 단위의 재실행 작업시에도 깔끔하게 실행이 가능해집니다.

1. 기본 작업 구성

import pendulum
from airflow import DAG
from airflow.sdk import DAG, task

with DAG(
        dag_id="setup_n_teardown",
        schedule="@once",
        start_date=pendulum.datetime(2025, 6, 1, tz="Asia/Seoul"),
        catchup=False,
) as dag:
    @task(task_id='pre_task')
    def pre_task():
        print('INITIALIZE')

    @task(task_id='real_task')
    def real_task():
        MSG = """
        HELLO WORLD!

        THIS IS REAL PIPELINE TASK!
        """
        print(MSG)

    @task(task_id='post_task')
    def post_task():
        print('FINALIZE')

pre_task().as_setup() >> real_task() >> post_task().as_teardown()

• pre_task : real_task 이전에 선행되는 작업

• real_task : DAG의 실질적인 작업

• post_task : real_task가 끝나고 후행되는 작업

• 위와 같이 real_task에 대한 선행 작업은 as_setup(), 후행 작업은 as_teardown() 메소드를 통해 각각 설정이 가능합니다.

• Airflow UI의 그래프를 확인해보면 일반적인 Task와는 다르게 setup이 설정된 Task는 ↗, teardown이 설정된 작업은 ↘ 모양의 화살표가 붙어있는 것을 확인할 수 있습니다.

• 또한 위의 경우, real_task만을 Clear로 재실행하면 해당 Task의 선후행 작업이 모두 같이 실행되는 것을 볼 수 있습니다.

2. Task Group 작업 구성

• 이번엔 Task Group으로 묶인 Task 내에서 setup 과 teardown 선언시 어떻게 작업이 처리되는지 보도록 하겠습니다.

  import pendulum
  from airflow.sdk import DAG, task, task_group
  
  

with DAG( dag_id="setup_n_teardown_tg", schedule="@once", start_date=pendulum.datetime(2025, 6, 1, tz="Asia/Seoul"), catchup=False, ) as dag:

@task_group(group_id='first_group')
def first_group():
    @task(task_id='pre_task')
    def pre_task():
        print('INITIALIZE')

    @task(task_id='real_task')
    def real_task():
        MSG = """
        HELLO WORLD!
        THIS IS REAL PIPELINE TASK!
        """
        print(MSG)

    @task(task_id='post_task')
    def post_task():
        print('FINALIZE')

    pre_task().as_setup() >> real_task() >> post_task().as_teardown()

@task(task_id='task_outer_tg')
def task_outer_tg():
    print('outer tg')

@task_group(group_id='second_group')
def second_group():
    @task(task_id='pre_task')
    def pre_task():
        print('INITIALIZE')

    @task(task_id='real_task')
    def real_task():
        MSG = """
        HELLO WORLD!
        THIS IS REAL PIPELINE TASK!
        """
        print(MSG)

    @task(task_id='post_task')
    def post_task():
        print('FINALIZE')

    pre_task().as_setup() >> real_task() >> post_task().as_teardown()

first_group() >> task_outer_tg() >> second_group()

```

• first_group과 second_group 내에 각각 선후행 작업이 있는 Task가 존재합니다. 이 두 Task Group 사이에는 task_outer_tg라는 Task가 연결되어 있습니다.
• 아래 그래프에서 볼 수 있듯 Task Group 내에서 setup과 teardown이 설정된 경우, 이후에 따라오는 작업(task_outer_tg)은 이전 작업의 real_task와 곧바로 연결되는 것을 확인할 수 있습니다.
• 즉, 그룹이 이후 다른 Task과 연결될 때, 그 Task는 그룹내에 있는 선후행 작업이 아닌 실제 작업과 연결되어 실행되고, 후행 작업은 그와 독립적으로 실행되는 것을 확인할 수 있습니다.

3. teardown 작업 옵션 설정

• 1에서 다룬 예제와 같이 선후행 작업이 설정되어 있을 때, 선행 작업, 본작업, 후행 작업 각각 Fail시 어떻게 진행되는지 보도록 하겠습니다.

1) 선행 작업 실패시

• 선행 작업 실패 → 본작업 실패 + 전체 DAG 실패 → 후행 작업은 성공 or 실패
• 선행 작업이 실패하면 본작업 역시 제대로 진행되지 못하여 전체 DAG가 FAILED 됩니다. 하지만 후행 작업은 성공될 수 있습니다.

2) 본작업 실패시

• 선행 작업 성공 → 본작업 실패 + 전체 DAG 실패 → 후행 작업 성공 or 실패
• 본작업이 실패하면 전체 DAG는 FAILED되지만 후행 작업은 성공될 수 있습니다.

3) 후행 작업 실패시

• 선행 작업 성공 → 본작업 성공 + 전체 DAG 성공 → 후행 작업 성공

• 후행 작업의 경우 성공, 실패 여부와 상관 없이 본작업이 성공하게 되면 전체 DAG는 SUCCESS로 표시됩니다.

• 하지만, 아래 설정을 통해 후행 작업 실패시 전체 DAG가 FAILED 되도록 설정할 수도 있습니다.

  • as_teardown(on_failure_fail_dagrun=True)

4. 참고 자료

• Setup and Teardown 공식 문서

• GCS, S3와 같은 클라우드 객체 스토리지를 Data Lake로 활용하는 경우, 지속적으로 유입되는 데이터를 디렉토리 단위로 파티셔닝하여 저장하는 방식이 일반적입니다. 예를 들어, 일 단위로 적재되는 데이터는 yyyy=/mm=/dd= 형식으로, 시간 단위 배치 데이터는 yyyy=/mm=/dd=/hh= 와 같은 구조로 저장할 수 있습니다. 이러한 디렉토리 구조는 데이터의 배치 주기와 일치하도록 설계되어 효율적인 관리와 조회를 가능하게 합니다.
• 이처럼 주기적으로 누적되는 데이터를 분석용 테이블로 적재할 때는, 증가된 데이터만 APPEND 방식으로 적재하는 방법이 리소스와 처리 시간 면에서 훨씬 효율적입니다. 단, 이 방식은 원천 데이터에 삭제나 수정 없이 오직 데이터가 추가만 되는 경우에 안정적으로 작동할 수 있으며, 로그나 이벤트 기록처럼 이력성 데이터를 처리할 때 특히 유용한 전략입니다.
• 이번 글에서는 GCS(Google Cloud Storage)에 일 단위로 적재되는 데이터를 중분(Incremental) 방식으로 테이블에 적재하는 방법에 대해 다뤄보도록 하겠습니다.

1. GCS 적재 데이터 구조

• GCS에 적재되는 데이터의 형상은 아래와 같습니다.

• 컬럼 4개로 이루어진 테이블이며, 그 중 proc_ymd 컬럼이 적재된 날짜(년월일)를 알려줍니다.

• 적재 기간은 2023-04-01 ~ 2023-04-30 입니다.

• GCS에는 yyyy=2023/mm=04/dd=01/data.parquet 이런 형식으로 일단위로 디렉토리가 나뉘어 적재됩니다.

2. EXTERNAL TABLE 활용하기

1) External Table이란?

• External Table은 BigQuery에서 외부 저장소(GCS, Google Drive 등)에 있는 데이터를 직접 참조하여 쿼리할 수 있는 가상 테이블입니다. 즉, 데이터를 BigQuery 테이블로 만들어 저장해놓지 않고도, 마치 BigQuery 내부 테이블처럼 SQL을 사용하여 조회할 수 있도록 해주는 기능입니다.
• 따라서 GCS에 parquet 파일 형식(CSV, JSON 등 다양한 형식 지원)으로 데이터가 적재되는 경우, 해당 디렉토리를 BigQuery External Table로 등록해놓으면 새롭게 유입된 데이터를 바로 조회해볼 수 있습니다.
• 하지만 BigQuery 내부 테이블에 비해서는 조회 속도가 느리며, 파티셔닝/클러스터링 기능 활용이 불가능하여 성능 최적화에 한계가 있습니다.

2) 사용 방법

• GCS에 적재되어 있는 데이터를 External Table로 등록하려면 아래와 같은 형식으로 SQL 쿼리를 실행하면 됩니다.

  CREATE OR REPLACE EXTERNAL TABLE `프로젝트ID.데이터셋.테이블명`
  OPTIONS (
    format = 'PARQUET', -- CSV, JSON 등
    uris = [데이터에 대한 gsutil 주소]
    );

• 아래 코드는 GCS의 gs://hyunsoo_sprint_bucket/dataform_data/gcp_part_parquet/ 경로 이하에 yyyy=/mm=/dd= 형식으로 일 단위 적재되는 데이터들을 dataform 데이터셋에 demo_source 라는 이름의 테이블로 등록하는 쿼리입니다.

    CREATE OR REPLACE EXTERNAL TABLE `codeit-hyunsoo.dataform.demo_source`
    OPTIONS (
      format = 'PARQUET',
      uris = ['gs://hyunsoo_sprint_bucket/dataform_data/gcp_part_parquet/yyyy=2023/mm=04/*']
      );

3) 등록 확인

• 위의 SQL 쿼리문을 통해 External Table 등록이 완료되면 BigQuery 콘솔 화면에서 확인이 가능합니다.
• 이렇게 등록된 External Table의 경우 연결된 GCS 경로에 특정 날짜의 추가 데이터가 적재되면, 해당 적재분에 대한 조회가 바로 가능합니다.

3. DataForm으로 증분 데이터 처리하기

• 위와 같이 External Table을 생성하면, BigQuery는 GCS에 저장된 데이터를 직접 참조하기 때문에 별도의 증분 처리 없이도 최신 데이터를 즉시 조회할 수 있습니다. 그러나 이 데이터를 기반으로 다른 테이블을 생성하거나 가공된 결과를 저장해야 하는 경우, 해당 결과 테이블은 증분 처리가 필요합니다.
• 적재되는 데이터의 양이 많지 않다면 매번 전체 데이터를 다시 적재하는 full-refresh 방식도 고려할 수 있습니다. 하지만, 불필요한 계산과 비용을 줄이고 쿼리 성능을 최적화하려면 신규 데이터만 처리하는 증분 방식을 선택하는 것이 훨씬 효율적입니다.
• 이러한 증분 처리를 SQL 기반의 선언형 방식으로 손쉽게 구현할 수 있도록 도와주는 도구가 바로 Dataform입니다. Dataform을 활용하면 BigQuery 위에서 안정적이고 확장성 있는 증분 데이터 파이프라인을 간편하게 구성할 수 있습니다.

1) 증분 데이터 처리 코드 작성

definition/demo_increment.sqlx

config {
  type: "incremental",
  name: "demo_increment"
}

SELECT
  *
FROM codeit-hyunsoo.dataform.demo_source

${when(incremental(), `WHERE proc_ymd > (SELECT MAX(proc_ymd) FROM ${self()})`)}

• type: "incremental" → 이 모델은 증분 처리 대상 테이블임을 선언합니다.

• name: "demo_increment" → 생성될 테이블의 이름은 demo_increment 입니다.

• SELECT 쿼리문 → demo_source 테이블의 전체 컬럼을 조회합니다.

• when(incremental() 쿼리문 → 증분 실행 조건으로, demo_source 테이블의 proc_ymd 컬럼에 새로운 날짜 데이터가 들어오면 해당 데이터를 demo_increment로 삽입합니다.

즉, 위의 스크립트는 demo_source 테이블에 새로운 데이터가 들어오면 추가된 데이터만 demo_increment 테이블에 적재합니다.

2) 실행 확인 확인

• type: "incremental" 조건의 증분 처리 스크립트의 경우Run` 버튼을 누르면 아래와 같이 증분 조건일 경우와 아닐 경우, 두 가지 경우에 대해 실행 테스트를 진행해볼 수 있습니다.

4. OUTRO

Dataform Incremental Table 구성 공식 문서

• 이번 글에서는 GCS에 일 단위로 데이터가 파일 형식으로 적재되는 경우, BigQuery에서 이를 효율적으로 분석하기 위한 두 가지 증분 처리 방식에 대해 살펴보았습니다.

  • External Table : 데이터를 BigQuery로 로드하지 않고도 GCS의 최신 데이터를 직접 조회할 수 있는 방식으로, 별도의 적재 작업 없이 빠르게 분석을 시작할 수 있다는 장점이 있습니다. 단, 대용량 데이터나 정교한 가공이 필요한 경우에는 성능이나 비용 측면에서 주의가 필요합니다.
  • Dataform incremental 모델 : 데이터가 쌓이는 구조를 반영하여, 새로운 데이터만을 선택적으로 BigQuery 테이블에 추가하는 방식입니다. 스케줄링과 병합 조건을 선언형으로 관리할 수 있어, 지속적인 데이터 파이프라인 운영에 적합합니다.

• 데이터 엔지니어링에서 증분 처리는 비용 효율성과 쿼리 성능을 동시에 잡기 위한 핵심 전략입니다. 데이터의 속성, 쿼리 목적, 사용 빈도 등을 고려하여 상황에 맞는 방식을 선택하는 것이 중요합니다.

• 앞선 글(BigQuery Dataform으로 빅쿼리 데이터 플로우 자동화하기!)에서는 BigQuery에서 제공해주는 Dataform 이라는 서비스에 대해 알아보고 간단한 실습까지 진행해 보았습니다. 빅쿼리 데이터를 기반으로 자동화된 파이프라인을 만들어준다는 것 외에도 Dataform에는 매력적인 기능들이 많이 있는데요, 대표적인 것이 Github Repo와의 연동입니다.
• Dataform에 작성한 코드를 Github Repo와 연결하여 UI에서 바로 commit이나 push 작업이 가능하며, 특정 브랜치에 push하는 것도 가능합니다.
• 이번 글에서는 Github Repo와 연동하는 방법에 대해 알아보고, .sqlx의 문법에 대해 추가적으로 다뤄보도록 하겠습니다.

🔹 1. Github와 연동

🔸 1) Github Repo 생성 및 토큰 발행

• dataform-practice 라는 이름의 깃허브 Repo를 생성합니다.

• Settings > Developer settings > Personal access tokens > Fine-grained tokens 메뉴로 가서 토큰을 발행합니다.

• 토큰 발행시 아래와 같이 설정하고 발행합니다.

  • 토큰 이름 : dataform-prac-token

  • Repository access : Only select repositories 선택 후 위에서 생성한 Repo를 선택합니다.

  • Repository permissions : Contents 선택 후 Read and write로 변경

• 발행이 완료되면 github_pat_xxxxx 와 같은 토큰값을 얻을 수 있습니다.

🔸 2) Secret Manager에 토큰 등록

• 위에서 발행한 토큰을 비밀이나 민감정보를 관리해주는 GCP 서비스인 Secret Manager에 등록해줘야 Dataform에서 github와 연동시 사용이 가능합니다.

• Secret Manager > +보안 비밀 만들기를 선택합니다.

• 이름을 정한 후 보안 비밀 값 항목에 위에서 발급받은 토큰값을 붙여넣고 보안 비밀을 생성합니다.

🔸 3) Dataform 저장소 생성 및 연동

• 위에서 생성한 github Repo와 연동될 Dataform 저장소를 생성합니다.
• 생성한 저장소로 들어가 Settings > Git과 연결 을 누르면 github Repo와 연동할 수 있는 설정창이 나옵니다. 아래 사진과 같이 github Repo 저장소 URL, 기본 브랜치 이름, 그리고 위에서 등록한 secret manager 키를 선택해주고 링크를 누르면 연동이 완료됩니다.

🔸 4) 작업 공간(workspace) 생성 후 초기화

• 연동된 Dataform 저장소에 작업 공간을 생성합니다. 이 작업 공간은 이후 Github Repo와 연동될 때 브랜치명으로 사용됩니다. 즉, dataform-github-repo라는 저장소 내에 작업 공간을 A, B, C 이렇게 3개를 만들었다면, 이들 각각이 연동된 github Repo의 브랜치가 되는 것입니다.
• tutorial이라는 작업 공간을 생성하고 작업공간 초기화 버튼을 클릭하면 초기 세팅 파일들이 생성됩니다.

🔸 5) Github Repo로 push

• tutorial 작업 공간에 생성된 초기 파일들을 연동된 Github Repo로 push 할 수 있습니다. UI에서 버튼 몇 번만 클릭하면 간단하고 직관적으로 commit 후 push가 완료됩니다.
• tutorial 브랜치로 push하는 것이 디폴트지만 설정을 통해 master(혹은 main) 브랜치로 바로 push하는 것도 가능합니다.
• Github에서 확인해보면 앞에서 생성한 dataform-practice Repo의 tutorial 브랜치에 해당 파일들이 잘 올라가있는 것을 확인할 수 있습니다.
• 이후 Dataform 작업시 파일이나 디렉토리에 변경이 일어나는 경우, 위와 같이 UI를 통해 github 저장소로 바로 push를 할 수 있어 코드 형상관리가 아주 편리해진다는 장점이 있습니다.

🔹 2. config 설정

• .sqlx 파일의 문법이 일반 SQL과 다른점은 상단에 정의되는 config 때문입니다. 이 설정을 통해 쿼리의 의존성, 실행 순서, 파티셔닝, 태그 지정, 품질 검사 등 파이프라인 작업에 필요한 다양한 세부 사항들을 정의할 수 있어, 단순한 SQL 쿼리를 넘어 복잡한 데이터 워크플로우를 구성이 가능하게 해줍니다.

🔸 1) 기본 설정 오버라이딩

• workflow_settings.yaml 파일에 작성했던 기본 설정값들을 파일 내에서 재정의 할 수 있습니다. 설정을 따로 하지 않으면, 기본 설정값에 명시된 사항에 따라 테이블이 생성됩니다.

  config {
  type: "table",
  description : "This is Sample Table"
  database: "my-gcp-project-id",
  schema: "education",
  name: "sample_tbl",
  columns: {
      user_id: {
        description: "Unique identifier for the user"
      event_timestamp: {
        description: "Timestamp of the battle event"
      }
  }

• type : 파일이 어떤 방식으로 실행되고 결과를 생성할지 유형을 정의
  • table : 물리적 테이블 생성
  • incremental : 테이블에 증분 방식으로 데이터를 추가하거나 업데이트
  • view : 논리적 뷰 생성
  • operations : 테이블이나 뷰를 생성하지 않고, 정의된 SQL 작업을 실행
  • assertion : 데이터 품질 검사를 정의(조건 불충족시 작업 실패)
• description : 생성될 테이블에 대한 설명 추가
• database : 작업이 저장될 project ID 설정
• schema : 저장될 빅쿼리 dataset 이름 설정
• name : 저장될 테이블 이름 설정(없다면 sqlx 파일명으로 저장)
• columns : 테이블의 컬럼에 대한 메타데이터(설명, 태그 등)를 정의

🔸 2) 추가 설정

config {
  type: "table",
  database: "my-gcp-project-id",
  schema: "education",
  name: "sample_tbl",
  columns: {
      user_id: {
        description: "Unique identifier for the user"
      },
      event_timestamp: {
        description: "Timestamp of the battle event"
      }
    },
  disabled: true,
  hasOutput: true,
  dependencies: ["raw_battle_data", "user_profile"]
  tags : ['dev']
    }

• disabled : 해당 파일에서의 테이블 생성 비활성화 여부
• hasOutput : type: "operations"인 SQLX 파일에서 출력 테이블을 생성하도록 지정
• dependencies : SQLX 파일 간의 명시적 의존성을 정의
• tags : 특정 작업을 선택적으로 실행하거나 그룹화할 때 사용되는 태그를 설정

🔹 3. 작업 전후 쿼리 설정

• pre_operations, post_operations 설정을 통해 파일 본문의 SQL 쿼리 작업이 실행되기 전과 후에 실행할 SQL 문을 정의할 수 있습니다.
  • pre_operations : 테이블 생성 전에 실행할 SQL 문을 정의
  • post_operations : 테이블 생성 후에 실행할 SQL 문을 정의

config {
  type: "table",
  description : "This is Sample Table"
  database: "my-gcp-project-id",
  schema: "education",
  name: "sample_tbl",
  columns: {
      user_id: {
        description: "Unique identifier for the user"
      event_timestamp: {
        description: "Timestamp of the battle event"
      }
}

-- 본문의 SQL 쿼리 실행 전 작업 정의
pre_operations {
    CREATE OR REPLACE TABLE sprintda05-hyunsoo.dataform.pre AS SELECT * FROM codeit-hyunsoo.dataform.source
  }

-- 본문의 SQL 쿼리 실행 이후 작업 정의
post_operations {
    CREATE OR REPLACE TABLE sprintda05-hyunsoo.dataform.post AS SELECT * FROM codeit-hyunsoo.dataform.source
}

-- SQL 쿼리
SELECT 1 AS number;

🔹 4. Assertions 설정

• Assertions 공식 문서
• 작업시 생성되는 테이블의 품질 검사를 정의하는 항목으로, 지정된 조건이 충족되지 않으면 워크플로우 실행이 실패하게 됩니다.
• 정의에 따른 쿼리의 결과값이 0이면 통과, 한 행 이상을 반환하게되면 실패로 간주됩니다.
• 일반적으로 아래와 같은 내용들을 체크할 때 많이 활용됩니다.
  • 데이터 무결성 : 필수 필드에 null 값이 없는지 확인.
  • 중복 데이터 : 고유 키(unique key)에 중복이 없는지 확인.
  • 데이터 범위 검증 : 값이 예상 범위 내에 있는지 확인.
  • 데이터 최신성 확인 : 데이터가 특정 시간 내에 적재되었는지 확인.
• Assertions 작업이 실패하게 되면 빅쿼리에 dataform_assertions이라는 dataset이 만들어지고 작업 파일명과 동일한 이름의 VIEW가 생성되어 실패한 행 확인이 가능합니다.

🔸 1) 수동 Assertions 정의

• 수동 Assertions는 독립적인 SQLX 파일에 정의되며, 특정 테이블의 품질을 테스트하는 데 사용됩니다.

• 예를 들어 demo_table 이라는 테이블의 status 행에 null값이 있는지 확인하는 Assertions를 작성할 수 있습니다.

• definitions/single_assert.sqlx

  config { type: "assertion" }
  
  SELECT status
  FROM dataform.demo_table
  WHERE status IS NULL

  • config { type: "assertion" } : 이 파일이 Assertion임을 지정.
  • 쿼리는 status 컬럼에서 null인 행을 찾음.
  • 쿼리가 0행을 반환하면 Assertion 성공, 1행 이상 반환하면 실패.
  • 실패 시 dataform_assertions.single_assert 뷰를 BigQuery에 생성하여 실패한 행을 확인할 수 있도록 함.

• 수동으로 Assertions를 정의하는 경우, 워크플로우 그래프를 보게되면 아래와 같이 하나의 작업만 보이는 것을 확인할 수 있습니다.

🔸 2) 테이블 내 Assertions 정의

• 테이블 정의 내에서 Assertions를 설정하면, 해당 테이블 생성 후 자동으로 데이터 품질 테스트를 수행합니다. config 블록에 assertions 속성을 추가하여 정의합니다.

• definitions/multi_assert.sqlx

  config {
    type: "table",
    assertions: {
      uniqueKey: ["user_id"],  // user_id가 고유해야 함
      nonNull: ["user_id", "customer_id"],  // user_id, customer_id가 null이 아니어야 함
      rowConditions: [
        "create_date > '2019-01-01'",  // create_date 조건
        "email LIKE '%@%.%'"  // 이메일 형식 검증
      ]
    }
  }
  
  SELECT
    user_id,
    customer_id,
    create_date,
    email
  FROM dataform.assertion_table

  • uniqueKey : user_id 열에 중복 값이 없어야 함.
  • nonNull : user_id와 customer_id가 null이 아니어야 함.
  • rowConditions : 컬럼의 값들이 지정된 SQL 조건을 만족해야 함

• 아래 정의된 SQL 문을 바탕으로 파일명과 동일한 multi_assert 테이블 생성 후 assertions 항목에 정의된 검증을 실행합니다.

• assertions 항목에 정의된 내용의 통과 유무는 이후 연결된 작업에 영향을 미치기 때문에 해당 파일 하나로는 품질 검증에 의미가 크게 없고, 각 검증 절차 이후 연결되는 또 다른 작업 파일이 있을 때 활용성이 높아집니다.

• 워크플로우 그래프를 확인해보면, 아래와 같이 multi_assert 작업 이후 assertions 검증 항목이 연결되어 보이는 것을 확인할 수 있습니다.

🔹 5. OUTRO

• 이번 글에서는 Dataform의 저장소를 Github와 연동하고, .sqlx 파일에 정의할 수 있는 config 요소들에 대해 상세히 알아보았습니다. 위의 기능들만 잘 활용해도 BigQuery 테이블을 바탕으로 효율적인 ETL 파이프라인을 구성할 수 있지 않을까 싶습니다.

• Apache Airflow, dbt, Step Functions, Databricks Workflow 등 데이터 관련 여러 작업들을 자동화하고 관리해주는 다양한 오케스트레이션 도구들이 있습니다. 그 중 dbt와 유사한 방식으로 BigQuery 환경에서 활용할 수 있는 도구가 있습니다. 바로 Dataform입니다.
• Dataform은 Google BigQuery에 저장된 데이터를 변환하고 워크플로우를 관리할 수 있도록 도와주는 도구입니다. BigQuery에 저장된 테이블들을 대상으로 자동화된 데이터 파이프라인 구축이 가능하죠.
• SQL 쿼리문에 config가 더해진 SQLX라는 확장된 SQL 문법과 JavaScript를 활용해 데이터 변환 로직을 정의할 수 있으며, 개발 작업공간을 통해 팀원들이 독립적으로 작업하고 변경 사항을 버전 관리(Git)로 관리할 수 있도록 지원합니다.
• 이번 글에서는 BigQuery 기반의 Dataform을 처음 사용하는 분들을 위해, 기본적인 사용법을 차근차근 소개해보려고 합니다.

🔹 1. 기본 세팅

▪ 1) 저장소 만들기

• Dataform에서 저장소(Repository)는 가장 상위 계층에 위치하며 파이프라인 작업들을 논리적으로 구분하는 단위입니다.
• 작업 공간(Workspace)은 저장소 내에 위치하며 저장소의 작업에 대해 버전 관리나 분기가 필요할 때 이를 구분하는 데 사용됩니다.
• BigQuery Dataform UI에서 +저장소 만들기 버튼을 누르고 저장소 이름과 리전(Region)을 선택하면 생성이 가능합니다.
• quickstart-repository 라는 이름으로 저장소를 생성하였습니다.

▪ 2) 서비스 계정 권한 부여

• 저장소 생성이 되면 이어서 Dataform 서비스가 사용할 서비스 계정에 대한 권한 확인 작업이 진행됩니다. 만약 Dataform을 최초로 이용하는 것이라면 생성된 서비스 계정에 대해 최소한 roles/bigquery.user 역할 추가가 필요합니다.

▪ 3) 작업 공간 만들기

• 작업 공간 역시 위에서 만든 저장소에 들어가 +개발 작업공간 만들기 버튼을 눌러 생성이 가능합니다.
• quickstart-workspace라는 이름으로 작업 공간을 생성하였습니다.

4) 작업 공간 초기화

• 위에서 생성한 quickstart-workspace에 들어가 작업공간 초기화 버튼을 누르게 되면 Dataform 작업에 필요한 초기 파일들이 생성되게 됩니다.
• 초기 세팅 파일/디렉토리 설명
  • definitions/ : Dataform 프로젝트의 SQLX 파일과 JavaScript 파일을 저장하는 디렉토리로, 본격적인 데이터 ETL 파이프라인 코드가 작성되는 디렉토리입니다.
  • includes/ : 재사용 가능한 JavaScript 함수나 공통 SQL 로직을 저장하는 디렉토리로, 프로젝트 전반에서 참조되는 파일들이 위치하는 디렉토리입니다. workflow_settings.yaml : Dataform 워크플로의 실행 설정(예: 스케줄, BigQuery 위치, 기본 스키마 등)을 정의하는 구성 파일입니다.

🔹 2. 단일 테이블 작업 진행

▪ 1) workflow_settings.yaml

• 기본적으로 아래와 같이 5개의 키들이 세팅되어 있고 그 중 중요한 키는 3번째 defaultDataset으로, Dataform에서 작성한 .sqlx 파일내의 쿼리의 결과물이 빅쿼리의 어떤 Dataset에 저장될지를 명시하는 키입니다.

  defaultProject: codeit-hyunsoo
  defaultLocation: asia-northeast3
  defaultDataset: dataform
  defaultAssertionDataset: dataform_assertions
  dataformCoreVersion: 3.0.0

• workflow_settings.yaml 관련 더 추가적인 내용은 공식 문서에서 확인할 수 있습니다.

▪ 2) .sqlx 파일 구성

• .sqlx 파일은 상단의 config 설정 부분과 하단의 SQL 쿼리 작성 부분으로 나뉩니다.

• 🛠 config 설정부

  • .sqlx 파일의 최상단에 위치하며, 중괄호 {} 안에 작성됩니다.
  • 이 부분은 해당 파일에서 생성하는 테이블, 뷰, 혹은 선언 등에 대한 메타데이터 및 실행 설정을 정의합니다.
  • 주요 설정 내용
    • 테이블 이름, 타입(view/table/assertion 등) 지정
    • 파티션/클러스터링 설정
    • 태그, 설명, 라벨 추가
    • 의존성 설정(dependencies)
    • 외부 쿼리 옵션 지정
    • workflow_settings.yaml 내용 오버라이딩

• 📄 SQL 쿼리 작성부

• config에 정의한 설정을 기반으로 실제로 실행될 SQL 쿼리문을 작성합니다.

▪ 3) source.sqlx 파일 작성

• 컬럼이 2개(fruit, count)있는 샘플 테이블을 생성하는 쿼리를 작성합니다.

• definitions/ 디렉토리 아래에 source.sqlx 파일 생성 후 아래 내용을 작성하면 됩니다.

• 아래 파일은 빅쿼리의 dataform 데이터셋에 source라는 테이블로 쿼리의 내용을 저장합니다. config상에 별도로 명시가 되어있지 않다면, .sqlx 파일의 이름과 동일하게 테이블의 이름이 정해집니다.

  config {
    type: "table", // SQL을 통해 생성될 테이블의 종류 (table/view)
    description: "Description of the table", // 쿼리에 대한 설명
  }
  
  SELECT
    "apples" AS fruit,
    3 AS count
  UNION ALL
  SELECT
    "oranges" AS fruit,
    5 AS count
  UNION ALL
  SELECT
    "pears" AS fruit,
    1 AS count
  UNION ALL
  SELECT
    "bananas" AS fruit,
    0 AS count

▪ 4) 작업 실행 및 확인

• 쿼리 작성이 완료되면 자동으로 해당 .sqlx 파일의 문법을 체크하여 틀린 부분이 없는지 보여줍니다. config나 SQL 문법상 오류가 없다면 아래 사진과 같이 초록색 체크 표시가 나옵니다.

• 해당 쿼리를 실행하려면 실행 시작 > 작업 실행 버튼을 누른 후 실행할 작업을 선택하여 시작할 수 있습니다.

• Compiled graph 탭으로 가면 작성한 .sqlx 파일의 워크플로우를 시각적으로 확인할 수 있습니다.

• 실행이 정상적으로 완료되면 기본 설정과 같이 dataform.source 테이블이 생성되어 빅쿼리에서 확인할 수 있습니다.

🔹 3. 여러 테이블 작업 진행

• 이번에는 위에 작성한 source.sqlx 파일로 생성된 테이블을 대상으로 집계를 하는 aggregation.sqlx 파일을 추가로 생성하여 2개의 .sqlx 파일이 연동되어 작업되는 파이프라인을 구성해볼 것입니다.

▪ 1) aggregation.sqlx 파일 작성

• source 테이블의 count 컬럼 값을 SUM 하는 쿼리입니다.

• dataform 데이터셋에 파일명과 동일한 aggregation이라는 이름의 VIEW로 저장되도록 config를 구성하였습니다.

• ${ref("source")} 코드를 통해 같은 디렉토리 내에서 생성되는 테이블을 참조할 수 있습니다. 참조 정의가 되면 파일의 실행 순서가 자연스럽게 정해집니다.

  config {
    type: "view", // SQL을 통해 생성될 테이블의 종류 (table/view)
    description: "Aggregation of source table", // 쿼리에 대한 설명
  }
  
  SELECT
      SUM(count) AS fruit_cnt
  FROM ${ref("source")}

• Compiled graph 탭을 확인하면 작업 순서에 맞게 UI로 표시가 되는 것을 확인할 수 있습니다.

  ▪ 2) 작업 실행 및 확인

• 실행 시작 > 작업 실행 버튼을 누른 후 실행할 작업(2개)을 선택하여 작업을 실행합니다.

• 실행이 완료되면 BigQuery 스튜디오에서 생성된 테이블을 확인할 수 있습니다.

🔹 4. OUTRO

• 지금까지 Dataform의 기본 개념부터 실제 사용법까지, 간단한 실습을 통해 함께 살펴보았습니다. 처음에는 다소 생소하게 느껴질 수 있지만, 익숙해지면 SQL 기반으로 깔끔하고 체계적인 데이터 파이프라인을 구축할 수 있다는 점에서 매우 매력적인 도구입니다. 특히 dbt를 사용해본 경험이 있다면, Dataform의 구조와 사용 방식에 훨씬 빠르게 익숙해질 수 있을 것입니다.
• Dataform 서비스 자체는 무료이며 .sqlx 파일을 통한 빅쿼리 엔진 사용에 대한 비용만 발생합니다. Github와의 호환성, 시각적인 워크플로우 관리 기능 등도 지원되어, BigQuery를 데이터 웨어하우스로 사용 중이라면 충분히 활용해볼 만한 자동화 도구라고 생각됩니다.

0. INTRO

• 앞선 글 🌊 Delta Lake 입문자를 위한 가이드 - 실전편(Part 1. 로컬 환경)에서는 Pyspark Docker Container 환경에서 Pyspark를 활용하여 delta 유형의 파일들을 생성하고 다뤄보는 실습을 진행하였습니다.
• 이번 Part 2 실습에서는 세부 내용은 유사하지만, Delta 형식의 데이터를 Spark로 보다 간편하게 다룰 수 있도록 도와주는 delta-spark 라이브러리를 활용해보는 내용을 담았습니다.
• 실습 환경의 경우 hyunsoolee0506/pyspark-cloud:3.5.1 이미지로 컨테이너를 생성하시는 것을 권장드립니다만 이번 실습의 경우는 Google Colab에서 진행해도 무방합니다.
• 실습에서는 아래 delta_data.zip파일 안에 있는 두 가지 CSV 파일 데이터를 사용하였습니다.

  👉 delta_data.zip

1️⃣ 기본 환경 세팅

▪ 1) Docker Container 생성

• 사용자의 컴퓨터에서 volume으로 사용할 디렉토리와 컨테이너 내부 /workspace/spark 디렉토리가 매핑되도록 설정합니다.

  docker run -d \
    --name pyspark \
    -p 8888:8888 \
    -p 4040:4040 \
    -v [사용자 디렉토리]:/workspace/spark \
    hyunsoolee0506/pyspark-cloud:3.5.1

• 위 명령어 실행 후 8888 포트로 접속하면 juypter lab 개발 환경에 접속할 수 있습니다.

  ▪ 2) 라이브러리 설치

• 실습에 필요한 라이브러리들을 설치합니다. hyunsoolee0506/pyspark-cloud:3.5.1 이미지에는 이미 설치되어 있지만 colab의 경우 아래 코드 실행을 통해 라이브러리들을 설치해야 합니다.

  pip install pyspark==3.5.1 delta-spark==3.2.0 pyarrow findspark

▪ 3) Pyspark delta lake 환경 설정

• https://docs.delta.io/latest/quick-start.html#set-up-apache-spark-with-delta-lake
• pyspark에서 delta lake를 사용하기 위해서 SparkSession 생성시 관련 extension들에 대한 설정을 합니다.

  from delta import *
  from pyspark.sql import SparkSession
  import pyspark.sql.functions as F
  

builder = SparkSession.builder.appName("DeltaLakeLocal")
.enableHiveSupport()
.config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
.config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")

spark = configure_spark_with_delta_pip(builder).getOrCreate()

---
## 2️⃣ delta 형식 테이블 생성
### ▪ 1) 데이터베이스 생성
```python
spark.sql("CREATE DATABASE IF NOT EXISTS deltalake_db")

spark.sql("SHOW DATABASES").show()

---
+------------+
|   namespace|
+------------+
|     default|
|deltalake_db|
+------------+

▪ 2) 일반 테이블 생성

"""
테이블 이름 : trainer
스키마 : 
  - id → INT
  - name → STRING
  - age → INT
  - hometown → STRING
  - prefer_type → STRING
  - badge_count → INT
  - level → STRING
""" 

spark.sql(f"""
CREATE TABLE IF NOT EXISTS deltalake_db.trainer (
  id INT,
  name STRING,
  age INT,
  hometown STRING,
  prefer_type STRING,
  badge_count INT,
  level STRING
)
USING csv
OPTIONS (
  path '/workspace/spark/deltalake/dataset/trainer_data.csv',
  header 'true',
  inferSchema 'true',
  delimiter ','
)
""")

spark.sql("SHOW TABLES FROM deltalake_db").show()

---
+------------+-------------+-----------+
|   namespace|    tableName|isTemporary|
+------------+-------------+-----------+
|deltalake_db|      trainer|      false|
+------------+-------------+-----------+

▪ 3) delta 테이블 생성

LOCAL_DELTA_PATH 변수에 delta 테이블이 저장될 디렉토리를 저장합니다. 이 디렉토리는 로컬 디렉토리가 될 수도 있고 s3나 GCS 같은 클라우드 스토리지의 경로가 될 수도 있습니다.

LOCAL_DELTA_PATH = '/workspace/spark/deltalake/delta_local/trainer_delta/'

query = f"""
CREATE TABLE IF NOT EXISTS deltalake_db.trainer_delta (
  id INT,
  name STRING,
  age INT,
  hometown STRING,
  prefer_type STRING,
  badge_count INT,
  level STRING
)
USING delta
LOCATION '{LOCAL_DELTA_PATH}'
"""
spark.sql(query)

spark.sql("SHOW TABLES FROM deltalake_db").show()

---
+------------+-------------+-----------+
|   namespace|    tableName|isTemporary|
+------------+-------------+-----------+
|deltalake_db|      trainer|      false|
|deltalake_db|trainer_delta|      false|
+------------+-------------+-----------+

▪ 4) delta 테이블에 데이터 삽입

• CSV로 생성한 일반 테이블의 데이터를 delta 테이블에 삽입합니다.

  query = """
  INSERT INTO deltalake_db.trainer_delta
  SELECT * FROM deltalake_db.trainer;
  """
  spark.sql(query)
  

spark.sql('SELECT * FROM deltalake_db.trainer_delta').show(5)

+---+-----------+---+--------+-----------+-----------+------------+ | id| name|age|hometown|prefer_type|badge_count| level| +---+-----------+---+--------+-----------+-----------+------------+ | 1| Brian| 28| Seoul| Electric| 8| Master| | 2| Sabrina| 23| Busan| Water| 6| Advanced| | 3| Susan| 18| Gwangju| Rock| 7| Expert| | 4| Martin| 20| Incheon| Grass| 5| Advanced| | 5| Gabrielle| 30| Daegu| Flying| 6| Advanced| +---+-----------+---+--------+-----------+-----------+------------+

- 디렉토리를 확인하면 아래와 같이 `_delta_log` 폴더가 생성된 것을 확인할 수 있습니다.
![](https://velog.velcdn.com/images/newnew_daddy/post/e771ecd0-cc66-4fa0-b429-887ba999dabd/image.png)

---
## 3️⃣ DeltaTable로 데이터 읽기
### ▪ 1) delta 형식인지 확인
- 디렉토리에 저장된 파일이 `delta` 형식인지 확인합니다.
- 여기서 파라미터에 들어가는 `spark`는 위에 생성한 SparkSession에 대한 값이 담겨있는 변수입니다.
```python
DeltaTable.isDeltaTable(spark, LOCAL_DELTA_PATH)

---
True

▪ 2) delta 테이블 읽기

• delta 테이블을 읽은 방식은 두 가지가 있습니다.

1) 저장된 테이블 이름으로 읽어오기

dt = DeltaTable.forName(spark, "deltalake_db.trainer_delta")

2) 테이블이 저장된 경로로 읽어오기

dt = DeltaTable.forPath(spark, LOCAL_DELTA_PATH)

▪ 3) DeltaTable을 spark dataframe으로 변환

• 위의 코드로 읽어오게 되면 delta.tables.DeltaTable 타입으로 저장됩니다. 따라서 데이터를 조회하기 위해서는 spark dataframe으로 변환 후 show() 메소드로 조회합니다.

  dt.toDF().show(5)
  

+---+---------+---+--------+-----------+-----------+--------+ | id| name|age|hometown|prefer_type|badge_count| level| +---+---------+---+--------+-----------+-----------+--------+ | 1| Brian| 28| Seoul| Electric| 8| Master| | 2| Sabrina| 23| Busan| Water| 6|Advanced| | 3| Susan| 18| Gwangju| Rock| 7| Expert| | 4| Martin| 20| Incheon| Grass| 5|Advanced| | 5|Gabrielle| 30| Daegu| Flying| 6|Advanced| +---+---------+---+--------+-----------+-----------+--------+

---
## 4️⃣ DeltaTable 형식 테이블 생성
### ▪ 1) create
- `delta.tables.DeltaTable` 타입의 비어있는 테이블을 생성합니다.
- create 관련해서는 아래 세 가지 종류가 있는데 활용법이 비슷하므로 `create` 메소드만 실습해보도록 하겠습니다.
  - `create` : 새로운 DeltaTable을 생성합니다. 테이블이 이미 존재하면 오류가 발생합니다.
  - `createIfNotExists` : 새로운 DeltaTable을 생성합니다. 테이블이 이미 존재해도 오류가 나지 않습니다.
  - `createOrReplace` : 새로운 DeltaTable을 생성하거나 동일한 이름의 기존 테이블을 대체합니다.
```python
my_dt = DeltaTable.create(spark) \
          .tableName("my_table") \
          .addColumn("id", "INT") \
          .addColumn("name", "STRING") \
          .addColumn("age", "INT") \
          .execute()

my_dt.toDF().show()

---
+---+----+---+
| id|name|age|
+---+----+---+
+---+----+---+

▪ 2) replace

• 위의 createOrReplace 메소드와 비슷하게 기존 DeltaTable을 새로운 스키마의 테이블로 대체할 때 사용합니다.

  df = spark.createDataFrame([('Ryan', 31), ('Alice', 27), ('Ruby', 24)], ["name", "age"])
  

my_dt = DeltaTable.replace(spark)
.tableName("my_table")
.addColumns(df.schema)
.execute()

my_dt.show()

+----+---+ |name|age| +----+---+ +----+---+

## 5️⃣ DeltaTable 업데이트, 삭제, 병합
### ▪ 1) UPDATE
- dataframe에서 id가 5~10에 해당하는 row의 level 컬럼을 'Delta_Update'로 변경합니다.
```python
dt.update(
    condition="id >= 5 AND id <= 10",
    set={'level' : "'Delta_Update'"}
)

▪ 2) DELETE

prefer_type 컬럼에서 데이터가 `Rock'인 행을 모두 삭제합니다.

dt.delete(
    condition="prefer_type = 'Rock'"
)

▪ 3) MERGE

• merge()는 테이블에 데이터를 upsert(업데이트 또는 삽입)하거나 삭제하는 데 매우 유용한 기능입니다.
• merge 메소드의 옵션
  • whenMatchedDelete : 소스와 대상 테이블의 레코드가 매칭될 때, 해당 레코드를 삭제
  • whenMatchedUpdate : 소스와 대상 테이블의 레코드가 매칭될 때, 해당 레코드를 업데이트
  • whenMatchedUpdateAll : 소스와 대상 테이블의 레코드가 매칭될 때, 모든 컬럼을 소스 데이터로 업데이트
  • whenNotMatchedBySourceDelete : 소스 데이터에 없는 대상 테이블의 레코드를 삭제
  • whenNotMatchedBySourceUpdate : 소스 데이터에 없는 대상 테이블의 레코드를 업데이트
  • whenNotMatchedInsert : 소스 데이터가 대상 테이블에 없는 경우, 새로운 레코드를 삽입
  • whenNotMatchedInsertAll : 소스 데이터가 대상 테이블에 없는 경우, 모든 컬럼을 삽입
  • withSchemaEvolution : 스키마가 변경된 경우(예: 소스 데이터에 새로운 컬럼이 추가됨), 대상 테이블의 스키마를 자동으로 업데이트

1) 대상 테이블과 소스 테이블 생성

# 대상 테이블 → 기존 trainer 테이블에서 5개 행만 추출
dt.delete(
    condition="id > 5"
)

# 소스 테이블
data = [
    (1, "Brian", 29, "Seoul", "Electric", 9, "GrandMaster"),
    (3, "Susan", 19, "Gwangju", "Rock", 8, "Master"),
    (7, "Alex", 25, "Jeju", "Fire", 3, "Beginner"),
    (8, "Emily", 22, "Ulsan", "Psychic", 5, "Intermediate")
]

columns = ["id", "name", "age", "hometown", "prefer_type", "badge_count", "level"]
source_df = spark.createDataFrame(data, columns)

2) merge 작업 수행

• source_df 테이블 데이터 중 기존 dt와 id가 겹치는 행은 업데이트, id가 없는 행은 추가하는 작업을 수행합니다.

  dt.alias("target") \
    .merge(
        source=source_df.alias("source"),
        condition="target.id = source.id"
    ) \
    .whenMatchedUpdate(
        set={
            "name": "source.name",
            "age": "source.age",
            "hometown": "source.hometown",
            "prefer_type": "source.prefer_type",
            "badge_count": "source.badge_count",
            "level": "source.level"
        }
    ) \
    .whenNotMatchedInsert(
        values={
            "id": "source.id",
            "name": "source.name",
            "age": "source.age",
            "hometown": "source.hometown",
            "prefer_type": "source.prefer_type",
            "badge_count": "source.badge_count",
            "level": "source.level"
        }
    ) \
    .execute()
  

dt.toDF().show()

+---+---------+---+--------+-----------+-----------+------------+ | id| name|age|hometown|prefer_type|badge_count| level| +---+---------+---+--------+-----------+-----------+------------+ | 1| Brian| 29| Seoul| Electric| 9| GrandMaster| | 2| Sabrina| 23| Busan| Water| 6| Advanced| | 3| Susan| 19| Gwangju| Rock| 8| Master| | 4| Martin| 20| Incheon| Grass| 5| Advanced| | 5|Gabrielle| 30| Daegu| Flying| 6| Advanced| | 7| Alex| 25| Jeju| Fire| 3| Beginner| | 8| Emily| 22| Ulsan| Psychic| 5|Intermediate| +---+---------+---+--------+-----------+-----------+------------+

## 6️⃣ DeltaTable 메타데이터 조회
### ▪ 1) detail
- Delta 테이블의 상세 정보(스키마, 속성, 메타데이터 등)를 확인할 때 사용합니다.
```python
dt.detail().show(truncate=False)

---
+------+------------------------------------+----+-----------+---------------------------------------------------------+-----------------------+-----------------------+----------------+-----------------+--------+-----------+----------+----------------+----------------+------------------------+
|format|id                                  |name|description|location                                                 |createdAt              |lastModified           |partitionColumns|clusteringColumns|numFiles|sizeInBytes|properties|minReaderVersion|minWriterVersion|tableFeatures           |
+------+------------------------------------+----+-----------+---------------------------------------------------------+-----------------------+-----------------------+----------------+-----------------+--------+-----------+----------+----------------+----------------+------------------------+
|delta |e3db23a9-cc4c-4700-95ee-a8b4e06dfbf9|NULL|NULL       |file:/workspace/spark/deltalake/delta_local/trainer_delta|2025-03-28 07:31:56.941|2025-04-01 01:15:22.165|[]              |[]               |1       |3984       |{}        |1               |2               |[appendOnly, invariants]|
+------+------------------------------------+----+-----------+---------------------------------------------------------+-----------------------+-----------------------+----------------+-----------------+--------+-----------+----------+----------------+----------------+------------------------+

▪ 2) history

• Delta 테이블에 수행된 작업 기록(쓰기, 업데이트, 삭제 등)을 확인할 때 사용합니다.

  dt.history().show(truncate=False)
  

+-------+-----------------------+------+--------+------------+-----------------------------------------------------------------------------------------------+----+--------+---------+-----------+--------------+-------------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+------------+-----------------------------------+ |version|timestamp |userId|userName|operation |operationParameters |job |notebook|clusterId|readVersion|isolationLevel|isBlindAppend|operationMetrics |userMetadata|engineInfo | +-------+-----------------------+------+--------+------------+-----------------------------------------------------------------------------------------------+----+--------+---------+-----------+--------------+-------------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+------------+-----------------------------------+ |3 |2025-04-01 01:15:22.165|NULL |NULL |DELETE |{predicate -> ["(prefer_type#3178 = Rock)"]} |NULL|NULL |NULL |2 |Serializable |false |{numRemovedFiles -> 1, numRemovedBytes -> 3997, numCopiedRows -> 89, numDeletionVectorsAdded -> 0, numDeletionVectorsRemoved -> 0, numAddedChangeFiles -> 0, executionTimeMs -> 789, numDeletionVectorsUpdated -> 0, numDeletedRows -> 1, scanTimeMs -> 494, numAddedFiles -> 1, numAddedBytes -> 3984, rewriteTimeMs -> 294} |NULL |Apache-Spark/3.5.1 Delta-Lake/3.2.0| |2 |2025-04-01 01:13:45.637|NULL |NULL |UPDATE |{predicate -> ["((id#3174 >= 5) AND (id#3174 <= 10))"]} |NULL|NULL |NULL |1 |Serializable |false |{numRemovedFiles -> 1, numRemovedBytes -> 3980, numCopiedRows -> 84, numDeletionVectorsAdded -> 0, numDeletionVectorsRemoved -> 0, numAddedChangeFiles -> 0, executionTimeMs -> 1604, numDeletionVectorsUpdated -> 0, scanTimeMs -> 984, numAddedFiles -> 1, numUpdatedRows -> 6, numAddedBytes -> 3997, rewriteTimeMs -> 618}|NULL |Apache-Spark/3.5.1 Delta-Lake/3.2.0| |1 |2025-03-28 07:32:09.347|NULL |NULL |WRITE |{mode -> Append, partitionBy -> []} |NULL|NULL |NULL |0 |Serializable |true |{numFiles -> 1, numOutputRows -> 90, numOutputBytes -> 3980} |NULL |Apache-Spark/3.5.1 Delta-Lake/3.2.0| |0 |2025-03-28 07:31:57.505|NULL |NULL |CREATE TABLE|{partitionBy -> [], clusterBy -> [], description -> NULL, isManaged -> false, properties -> {}}|NULL|NULL |NULL |NULL |Serializable |true |{} |NULL |Apache-Spark/3.5.1 Delta-Lake/3.2.0| +-------+-----------------------+------+--------+------------+-----------------------------------------------------------------------------------------------+----+--------+---------+-----------+--------------+-------------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+------------+-----------------------------------+

### ▪ 3) generate
- Delta Lake 테이블의 데이터를 외부 시스템(예: Apache Hive, Presto, Amazon Athena 등)에서 읽을 수 있도록 매니페스트 파일을 생성하는 함수입니다.
- `generate` 함수를 실행하면 delta 테이블 디렉토리에 `_symlink_format_manifest` 디렉토리가 생성되고, 그 안에 현재 버전의 parquet 파일을 가리키는 메니페스트 파일이 작성됩니다.
- Presto, Trino, Amazon Athena, Apache Hive와 같은 엔진은 Delta Lake의 로그 기반 트랜잭션 시스템을 직접 지원하지 않습니다. 하지만 `_symlink_format_manifest`를 사용하면 delta 테이블을 parquet 파일로 표현한 매니페스트를 제공하므로, 이러한 엔진에서 테이블을 쿼리할 수 있습니다.
```python
dt.generate("symlink_format_manifest")

7️⃣ Time Travel 쿼리

• 데이터의 과거 버전을 조회하는 Time Travel 쿼리의 경우 특정 버전을 기준으로 조회하는 경우와 특정 시간(timestamp)을 기준으로 조회하는 경우, 이렇게 두 가지가 가능합니다.

  ▪ 1) Version

  dt.restoreToVersion(1)
  

dt.toDF().show(10)

### ▪ 2) Timestamp
- `SEARCH_TIME`에 저장된 시간에 존재했던 테이블 모습을 조회합니다.
```python
SEARCH_TIME = '2025-03-28 07:32:00'

dt.restoreToTimestamp(SEARCH_TIME)

dt.toDF().show(10)

8️⃣ 파일 상태 최적화

• 작게 나눠서 저장된 parquet 파일들을 합쳐 용량은 크게, 파일 수는 적게 만들어 데이터를 관리합니다. 이 과정을 통해 Delta 테이블의 쿼리 성능을 향상시킬 수 있습니다.
• executeZOrderBy 옵션은 지정된 컬럼에 대해 Z-Order 클러스터링을 적용하여 데이터를 물리적으로 재배치시킵니다. 쿼리에서 해당 컬럼에 대한 필터링이나 조인이 빈번할 때 효과적입니다.

  ▪ 1) OPTIMIZE

  # 표준 최적화 방법
  dt.optimize().executeCompaction()
  

특정 컬럼을 대상으로 최적화

dt.optimize().executeZOrderBy('level')

### ▪ 2) VACUUM
- 더 이상 필요 없는 오래된 데이터 버전을 삭제하여 저장 공간을 확보할 때 사용합니다.
- 기본은 7일(168시간)이며, `retentionHours` 파라미터를 통해 특정 시간 이후의 데이터를 지우도록 할 수 있습니다.
```python
dt.vacuum(
    retentionHours=10
)

9️⃣ Parquet to Delta 변환

▪ 1) parquet 파일 저장

• trainer_data.csv 파일을 읽어와 테이블과 경로에 parquet 타입으로 저장합니다.

  # 1. CSV 파일을 DataFrame으로 읽기
  DATA_PATH = '/workspace/spark/deltalake/dataset/trainer_data.csv'
  SAVE_PATH = '/workspace/spark/deltalake/delta_local/spark-warehouse/trainer_parquet/'
  

df = spark.read.csv(DATA_PATH, header=True, inferSchema=True)

2. 테이블 생성

query = f""" CREATE TABLE IF NOT EXISTS deltalake_db.trainer_parquet ( id INT, name STRING, age INT, hometown STRING, prefer_type STRING, badge_count INT, level STRING ) USING parquet LOCATION '{SAVE_PATH}' """ spark.sql(query)

3. DataFrame을 테이블과 경로에 저장

df.write.mode("overwrite")
.option("path", SAVE_PATH)
.saveAsTable("deltalake_db.trainer_parquet")

### ▪ 2) delta로 변환
```python
# 저장된 테이블을 delta로 변환
DeltaTable.convertToDelta(spark, "deltalake_db.trainer_parquet")

# 디렉토리에 저장된 파일 데이터를 delta로 변환
DeltaTable.convertToDelta(spark, f"parquet.`{SAVE_PATH}`")

참고자료

• Delta Lake’s Python documentation page

0. INTRO

• 앞선 글 🌊 Delta Lake 입문자를 위한 가이드 - 이론편에서는 Delta Lake에 대한 이론적인 내용을 상세하게 다루어 보았습니다. 이번 글에서는 Pyspark Docker Container 환경에서 Delta Lake의 기능을 실습해보도록 하겠습니다.
• 이번 실습에서는 데이터를 다루는 도구로 Pyspark를 사용하며, delta 유형의 파일들은 로컬 디렉토리에 저장되어 관리됩니다.
• 도커 컨테이너의 경우 이후 클라우드 환경을 연동한 실습까지 고려하였을 때 제가 따로 생성한 hyunsoolee0506/pyspark-cloud:3.5.1 이미지로 생성하시는 것을 권장드립니다. 하지만 이번 로컬 환경 실습의 경우는 Google Colab에서 진행해도 무방합니다.
• 실습에서는 아래 delta_data.zip파일 안에 있는 두 가지 CSV 파일 데이터를 사용하였습니다.

  👉 delta_data.zip

1️⃣ 실습 환경 설정

▪ 1) Docker Container 생성

• 사용자의 컴퓨터에서 volume으로 사용할 디렉토리와 컨테이너 내부 /workspace/spark 디렉토리가 매핑되도록 설정합니다.

  docker run -d \
    --name pyspark \
    -p 8888:8888 \
    -p 4040:4040 \
    -v [사용자 디렉토리]:/workspace/spark \
    hyunsoolee0506/pyspark-cloud:3.5.1

• 위 명령어 실행 후 8888 포트로 접속하면 juypter lab 개발 환경으로 들어올 수 있습니다.

  ▪ 2) 라이브러리 설치

• 실습에 필요한 라이브러리들을 설치합니다. hyunsoolee0506/pyspark-cloud:3.5.1 이미지에는 이미 설치되어 있지만 colab의 경우 아래 코드 실행을 통해 라이브러리들을 설치해야 합니다.

  pip install pyspark==3.5.1 delta-spark==3.2.0 pyarrow findspark

▪ 3) Pyspark delta lake 환경 설정

• https://docs.delta.io/latest/quick-start.html#set-up-apache-spark-with-delta-lake
• pyspark에서 delta lake를 사용하기 위해서 SparkSession 생성시 관련 extension들에 대한 설정을 합니다.

  from delta import *
  from pyspark.sql import SparkSession
  import pyspark.sql.functions as F
  

builder = SparkSession.builder.appName("DeltaLakeLocal")
.enableHiveSupport()
.config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
.config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")

spark = configure_spark_with_delta_pip(builder).getOrCreate()

---
## 2️⃣ 데이터베이스 및 테이블 생성
### ▪ 1) 데이터베이스 생성
- `deltalake_db`라는 이름의 새로운 데이터베이스를 생성합니다.
```python
spark.sql("CREATE DATABASE IF NOT EXISTS deltalake_db")

spark.sql("SHOW DATABASES").show()

---
+------------+
|   namespace|
+------------+
|     default|
|deltalake_db|
+------------+

▪ 2) csv 타입 테이블 생성

• trainer_data.csv 파일의 데이터가 저장될 trainer 테이블을 생성합니다. ```
• 테이블 이름 : trainer
• 스키마 :
  • id → INT
  • name → STRING
  • age → INT
  • hometown → STRING
  • prefer_type → STRING
  • badge_count → INT
  • level → STRING

    ```python
    query = f"""
    CREATE TABLE IF NOT EXISTS deltalake_db.trainer (
    id INT,
    name STRING,
    age INT,
    hometown STRING,
    prefer_type STRING,
    badge_count INT,
    level STRING
    )
    USING csv
    OPTIONS (
    path '[trainer_data.csv 파일 경로]',
    header 'true',
    inferSchema 'true',
    delimiter ','
    )
    """
    

spark.sql(query)

테이블 생성 확인

spark.sql("SHOW TABLES FROM deltalake_db").show()

+------------+---------+-----------+ | namespace|tableName|isTemporary| +------------+---------+-----------+ |deltalake_db| trainer| false| +------------+---------+-----------+

---
## 3️⃣ delta 타입 테이블 생성
- 기존에 있는 `csv` 파일의 데이터를 바로 `delta` 유형의 테이블 생성과 동시에 넣을 수는 없기 때문에 아래와 같이 두 단계를 거쳐 `delta` 테이블을 생성하여야 합니다.
  1. `delta` 유형 빈 테이블 생성
  2. `delta` 테이블에 `csv` 테이블 데이터 삽입
### ▪ 1) 테이블 생성
- `trainer_delta`라는 이름의 `delta` 테이블을 생성합니다.
- `/workspace/spark/deltalake/delta_local/trainer_delta/` 해당 경로 아래에 `delta` 테이블 관련 데이터가 저장되도록 설정하였습니다.
```python
query = f"""
CREATE TABLE IF NOT EXISTS deltalake_db.trainer_delta (
  id INT,
  name STRING,
  age INT,
  hometown STRING,
  prefer_type STRING,
  badge_count INT,
  level STRING
)
USING delta
LOCATION '/workspace/spark/deltalake/delta_local/trainer_delta/'
"""
spark.sql(query)

▪ 2) 데이터 삽입

• 위에서 생성하였던 trainer 테이블의 데이터를 trainer_delta 테이블에 삽입합니다.

  query = """
  INSERT INTO deltalake_db.trainer_delta
  SELECT * FROM deltalake_db.trainer;
  """
  spark.sql(query)

• 데이터 삽입이 완료가 되면 delta 테이블 디렉토리에 _delta_log/ 폴더와 parquet 파일이 새롭게 생성이 된 것을 확인해볼 수 있습니다.

4️⃣ delta 타입 테이블 읽기

• pyspark에서는 조금씩 다른 방식으로 저장된 테이블을 읽어올 수 있습니다.

  ▪ 1) spark에서 기본 읽기

  LOCAL_DELTA_PATH = '/workspace/spark/deltalake/delta_local/trainer_delta'
  

df = spark.read.format("delta").load(LOCAL_DELTA_PATH)

df.show(5)

+---+------+---+--------+-----------+-----------+------------+ | id| name|age|hometown|prefer_type|badge_count| level| +---+------+---+--------+-----------+-----------+------------+ | 1| Brian| 28| Seoul| Electric| 8| Master| | 3| Susan| 18| Gwangju| Rock| 7| Expert| | 6| Vicki| 17| Daejeon| Ice| 4|Intermediate| | 9|Olivia| 45| Incheon| Psychic| 3|Intermediate| | 10| Mark| 16| Gangwon| Fire| 4|Intermediate| +---+------+---+--------+-----------+-----------+------------+ only showing top 5 rows

### ▪ 2) `delta.`으로 읽기
```python
query = f"SELECT * FROM delta.`{LOCAL_DELTA_PATH}`"

spark.sql(query)

▪ 3) hive catalog에서 읽기

spark.table('deltalake_db.trainer_delta')

5️⃣ 테이블 수정 후 저장

• 이번에는 테이블 내용을 수정하여 기존에 저장되어 있던 디렉토리에 덮어쓰는 과정을 진행합니다. 이후에 있을 테이블 변경 이력 조회나 delta lake의 핵심 기능인 Time Travel 쿼리를 실습해보기 위한 과정입니다.

  ▪ 1) 'Beginner' 제외 후 저장

  # Beginner 제외한 dataframe 생성
  df_1 = df.filter(F.col('level') != 'Beginner')
  

기존 경로에 덮어쓰기

df_1.write
.format('delta')
.mode('overwrite')
.save(LOCAL_DELTA_PATH)

데이터 확인

df = spark.read.format("delta").load(LOCAL_DELTA_PATH) df.select('level').distinct().show()

+------------+ | level| +------------+ | Expert| | Advanced| | Master| |Intermediate| +------------+

### ▪ 2) 'Advanced' 제외 후 저장
```python
# Advanced 제외한 dataframe 생성
df_2 = df_1.filter(F.col('level') != 'Advanced')

# 기존 경로에 덮어쓰기
df_2.write \
    .format('delta') \
    .mode('overwrite') \
    .save(LOCAL_DELTA_PATH)

# 데이터 확인
df = spark.read.format("delta").load(LOCAL_DELTA_PATH)
df.select('level').distinct().show()

---
+------------+
|       level|
+------------+
|      Expert|
|      Master|
|Intermediate|
+------------+

• 데이터가 덮어씌워짐에 따라 parquet 파일이 추가되고, _delta_log/ 폴더 내에 메타데이터(.json 파일) 역시 추가되는 것을 확인할 수 있습니다.

6️⃣ 변경 이력(history) 조회 및 Time Travel 쿼리

▪ 1) History 조회

• delta 테이블에 대한 변경 이력을 조회합니다.
• 현재까지 테이블은 처음 생성(CREATE) 후 WRITE 가 총 3번 발생한 구조입니다. 따라서 version 역시 0,1,2,3 이렇게 존재합니다.
  • VERSION 0 → trainer_delta 테이블 생성 상태, 데이터 X
  • VERSION 1 → trainer 테이블에서 데이터 삽입된 최초 상태
  • VERSION 2 → 'Beginner' 행 제외 후 저장된 상태
  • VERSION 3 → Advanced' 행 제외 후 저장된 상태

    query = "DESCRIBE HISTORY deltalake_db.trainer_delta"
    

spark.sql(query).show(vertical=True, truncate=False)

-RECORD 0-------------------------------------------------------------------------------------------------------------- version | 3
timestamp | 2025-03-21 02:09:39.085
userId | NULL
userName | NULL
operation | WRITE
operationParameters | {mode -> Overwrite, partitionBy -> []}
job | NULL
notebook | NULL
clusterId | NULL
readVersion | 2
isolationLevel | Serializable
isBlindAppend | false
operationMetrics | {numFiles -> 1, numOutputRows -> 42, numOutputBytes -> 3125}
userMetadata | NULL
engineInfo | Apache-Spark/3.5.1 Delta-Lake/3.2.0
-RECORD 1-------------------------------------------------------------------------------------------------------------- version | 2
timestamp | 2025-03-21 02:08:32.646
userId | NULL
userName | NULL
operation | WRITE
operationParameters | {mode -> Overwrite, partitionBy -> []}
job | NULL
notebook | NULL
clusterId | NULL
readVersion | 1
isolationLevel | Serializable
isBlindAppend | false
operationMetrics | {numFiles -> 1, numOutputRows -> 85, numOutputBytes -> 3868}
userMetadata | NULL
engineInfo | Apache-Spark/3.5.1 Delta-Lake/3.2.0
-RECORD 2-------------------------------------------------------------------------------------------------------------- version | 1
timestamp | 2025-03-21 01:46:21.446
userId | NULL
userName | NULL
operation | WRITE
operationParameters | {mode -> Append, partitionBy -> []}
job | NULL
notebook | NULL
clusterId | NULL
readVersion | 0
isolationLevel | Serializable
isBlindAppend | true
operationMetrics | {numFiles -> 1, numOutputRows -> 90, numOutputBytes -> 3980}
userMetadata | NULL
engineInfo | Apache-Spark/3.5.1 Delta-Lake/3.2.0
-RECORD 3-------------------------------------------------------------------------------------------------------------- version | 0
timestamp | 2025-03-21 01:43:25.471
userId | NULL
userName | NULL
operation | CREATE TABLE
operationParameters | {partitionBy -> [], clusterBy -> [], description -> NULL, isManaged -> false, properties -> {}} job | NULL
notebook | NULL
clusterId | NULL
readVersion | NULL
isolationLevel | Serializable
isBlindAppend | true
operationMetrics | {}
userMetadata | NULL
engineInfo | Apache-Spark/3.5.1 Delta-Lake/3.2.0

### ▪ 2) Time Travel - Version
- 테이블 변경시마다 부여된 버전 번호를 기반으로 특정 버전에 해당하는 테이블의 내용을 불러옵니다.

👉 최초 버전(version 0) 테이블 불러오기
```python
df_pre = spark.read \
    .format("delta") \
    .option("versionAsof", 0) \
    .load(LOCAL_DELTA_PATH)

df_pre.select('level').distinct().show()

---
+-----+
|level|
+-----+
+-----+

👉 version 2 테이블 불러오기

df_pre = spark.read \
    .format("delta") \
    .option("versionAsof", 2) \
    .load(LOCAL_DELTA_PATH)

df_pre.select('level').distinct().show()

---
+------------+
|       level|
+------------+
|      Expert|
|    Advanced|
|      Master|
|Intermediate|
+------------+

👉 SQL로 Time Travel 쿼리하기

df_pre = spark.sql("SELECT * FROM deltalake_db.trainer_delta VERSION AS OF 3")

df_pre.select('level').distinct().show()
---
+------------+
|       level|
+------------+
|      Expert|
|      Master|
|Intermediate|
+------------+

▪ 3) Time Travel - Timestamp

• 테이블 변경 시간을 기준으로 조회하는 방법으로, 지정한 시간(TIMESTAMP) 기준으로 해당 시점에 존재했던 Delta 테이블의 상태(버전)를 조회합니다.

👉 지정한 시간대의 테이블 상태 불러오기

TABLE_TIMESTAMP = "2025-03-21T02:09:00"

spark.read.format("delta") \
    .option("timestampAsOf", TABLE_TIMESTAMP) \
    .table("deltalake_db.trainer_delta")

👉 SQL로 지정 시간대의 테이블 불러오기

TABLE_TIMESTAMP = "2025-03-21T02:09:00"

spark.sql(f"SELECT * FROM deltALake_db.trainer_delta TIMESTAMP AS OF '{TABLE_TIMESTAMP}'")

7️⃣ 스키마 변경 작업

• Delta Lake는 기존 테이블 스키마와 다른 데이터를 쓰려고 하면 에러가 나도록 하는 '스키마 강제(Strict Schema Enforcement)' 옵션을 사용합니다. 따라서 기존 작업하던 테이블의 스키마 변경이 일어났다면, 특정 옵션을 추가해주어야 덮어쓰기가 가능합니다.
• 실습 내용은 아래와 같습니다. ```
• 기존 컬럼 : ['id', 'name', 'age', 'hometown', 'prefer_type', 'badge_count', 'level']
• 변경된 테이블 컬럼 : ['id', 'name', 'age', 'hometown', 'prefer_type', 'badge_count', 'level', 'dummy_col']

👉 'dummy_col' 이라는 컬럼이 추가되어 스키마가 변경된 테이블 덮어쓰기

### ▪ 1) 표준 쓰기 - 작업 실패
```python
LOCAL_DELTA_PATH = '/workspace/spark/deltalake/delta_local/trainer_delta'

# 테이블 불러오기
df = spark.table("deltalake_db.trainer_delta")

# 'dummy_col' 컬럼 추가
df_diff = df.withColumn('dummy_col', F.lit(1))

# 스키마 합치기 시도
df_diff.write \
    .format('delta') \
    .mode('overwrite') \
    .save(LOCAL_DELTA_PATH)

# 덮어쓰려는 테이블의 스키마가 달라 아래의 에러 발생
# 👇👇👇👇👇
---
AnalysisException: [_LEGACY_ERROR_TEMP_DELTA_0007] A schema mismatch detected when writing to the Delta table (Table ID: 31dbae5e-d042-467b-9454-e483fdad97bb).
To enable schema migration using DataFrameWriter or DataStreamWriter, please set:
'.option("mergeSchema", "true")'.
For other operations, set the session configuration
spark.databricks.delta.schema.autoMerge.enabled to "true". See the documentation
specific to the operation for details.

Table schema:
root
-- id: integer (nullable = true)
-- name: string (nullable = true)
-- age: integer (nullable = true)
-- hometown: string (nullable = true)
-- prefer_type: string (nullable = true)
-- badge_count: integer (nullable = true)
-- level: string (nullable = true)


Data schema:
root
-- id: integer (nullable = true)
-- name: string (nullable = true)
-- age: integer (nullable = true)
-- hometown: string (nullable = true)
-- prefer_type: string (nullable = true)
-- badge_count: integer (nullable = true)
-- level: string (nullable = true)
-- dummy_col: integer (nullable = true)


To overwrite your schema or change partitioning, please set:
'.option("overwriteSchema", "true")'.

Note that the schema can't be overwritten when using
'replaceWhere'.

▪ 2) 스키마 합치기 옵션과 함께 쓰기

• 스키마가 다른 테이블을 덮어쓰기 위해서는 option("mergeSchema", "true") 옵션을 추가해주어야 합니다.

  df_diff.write \
    .format('delta') \
    .mode('overwrite') \
    .option("mergeSchema", "true") \
    .save(LOCAL_DELTA_PATH)

8️⃣ 파일 상태 최적화

• https://docs.databricks.com/aws/en/sql/language-manual/delta-optimize

• Delta 테이블은 기본적으로 계속 파일이 적재되는 형식이기 때문에 시간이 지남에 따라 작은 파일들이 많이 생기게 됩니다. 이렇게 되면 쿼리 성능 저하와 읽기 오버헤드 증가가 발생합니다. 이 때 OPTIMIZE를 통해 데이터를 큰 파일로 병합하여 성능을 향상시킬 수 있습니다.

  최적화 방식	설명
  기본 Optimize	작은 파일 병합, 읽기 성능 향상
  Z-Ordering	자주 필터링하는 컬럼 기준 정렬 → 스캔 줄여 쿼리 성능 향상
  파티션 기반 Optimize	특정 날짜/지역 등 자주 조회되는 파티션만 선택적 최적화
  ### ▪ 1) 표준 최적화

• delta lake가 기본적으로 수행하는 표준 최적화 방식을 적용합니다.

  query = "OPTIMIZE deltalake_db.trainer_delta"
  

spark.sql(query)

### ▪ 2) Z-Ordering 최적화
- 특정 컬럼 기준으로 데이터의 물리적 저장 순서를 최적화하는 기능.
- 자주 필터링하는 컬럼을 기준으로 Z-Order를 걸면 쿼리시 불필요한 파일 스캔을 줄일 수 있습니다.
```python
query = """
OPTIMIZE deltalake_db.trainer_delta
ZORDER BY (trainer_id, region)  
"""

spark.sql(query)

▪ 3) 파티션 최적화

• 전체 테이블을 대상으로 최적화하지 않고, 파티셔닝된 특정 범위의 데이터만 병합합니다.

  query = """
  OPTIMIZE deltalake_db.trainer_delta
  WHERE level = 'Master' 
  """
  

spark.sql(query)

---
## 9️⃣ 과거 데이터 삭제(VACUUM)
- https://docs.databricks.com/aws/en/sql/language-manual/delta-vacuum
- Delta Lake는 데이터의 수정이나 삭제 등이 발생하더라도 과거의 parquet 파일들은 다 남아있게 됩니다.
- 과거 버전의 파일들을 더 이상 사용하지 않는데 계속 남겨놓는 것은 낭비이기 때문에 VACUUM 기능을 활용하여 특정 기간 이전의 데이터를 삭제하는 작업을 수행할 수 있습니다.
- `delta` 유형의 데이터의 경우 과거의 파일 데이터를 지울 때는 디렉토리에서 직접 삭제하면 안되고 `VACUUM` 명령을 통해 지워야 테이블의 정합성을 해치지 않고 이후에도 원활한 작업이 가능해집니다.
- `VACUUM` 작업 발생시 삭제된 날짜 이전으로는 Time Travel 하여 조회하는 것이 불가능해집니다.
- 파일의 기본 유지 기간은 168시간(7일)이며, spark config 수정을 통해 유지 기간을 조정할 수 있습니다.
### ▪ 1) 기본 유지 기간 설정 해제
- spark에 기본적으로 설정되어 있는 설정을 해제해 주어야 retention 기간을 커스텀하게 관리할 수 있습니다.
```python
# 설정 확인
spark.conf.get("spark.databricks.delta.retentionDurationCheck.enabled")
-> 'true'

# 유지 기간 설정 해제
spark.conf.set("spark.databricks.delta.retentionDurationCheck.enabled", "false")

▪ 2) VACUUM 명령어 실행

• 사용자가 지정한 기간 이전에 생성된 parquet 파일은 삭제하도록 VACUUM 명령을 수행합니다.
• DRY RUN 옵션은 실제로 작업은 되지 않도록 하는 설정입니다.

  # 기본 VACUUM 명령 (168시간 이전의 파일 삭제)
  spark.sql("VACUUM deltalake_db.trainer_delta").show(truncate=False)
  

현재 버전 이전의 파일들 삭제

spark.sql("VACUUM deltalake_db.trainer_delta RETAIN 0 HOURS DRY RUN").show(truncate=False)

2일 이전의 파일들 삭제

spark.sql("VACUUM deltalake_db.trainer_delta RETAIN 2 DAYS DRY RUN").show(truncate=False)

### ▪ 3) 테이블 생성시 유지 기간 설정
- `delta` 유형 테이블 생성시 기본 retention 기간을 설정합니다.
```python
query = f"""
CREATE TABLE IF NOT EXISTS deltalake_db.trainer_delta_2 (
  id INT,
  name STRING,
  age INT,
  hometown STRING,
  prefer_type STRING,
  badge_count INT,
  level STRING
)
USING delta
LOCATION '/workspace/spark/deltalake/delta_local/trainer_delta_2/'
TBLPROPERTIES ('delta.deletedFileRetentionDuration' = 'interval 2 days');
"""

spark.sql(query)

🔟 Parquet to Delta 변환

• https://docs.databricks.com/aws/en/sql/language-manual/delta-convert-to-delta
• parquet 형태로 저장되어 있던 데이터를 delta 유형의 테이블 데이터로 변환하는 기능입니다.

▪ 1) 일반 parquet 데이터 변환

👉 fish_data.csv 데이터를 parquet으로 저장합니다.

# csv 파일 읽어오기
fish = spark.read.option('header', 'true').csv('fish_data.csv')

# 로컬 디렉토리 저장 + Catalog 저장
fish.write \
    .mode('overwrite') \
    .format('parquet') \
    .option('path', '/workspace/spark/deltalake/delta_local/fish_parquet/') \
    .saveAsTable('deltalake_db.fish_parquet')

👉 parquet으로 저장되어 있던 데이터를 delta로 변환

query = """
CONVERT TO DELTA
parquet.`/workspace/spark/deltalake/delta_local/fish_parquet/`
"""

spark.sql(query)

▪ 2) 파티션 된 parquet 데이터 변환

👉 Species 컬럼으로 파티션 된 parquet 데이터 쓰기

fish_df.write.mode('overwrite')\
    .format('parquet') \
    .partitionBy('Species') \
    .option('path', '/workspace/spark/deltalake/delta_local/fish_parquet_partitioned/') \
    .saveAsTable('deltalake_db.fish_parquet_partitioned')

👉 파티션 된 parquet 데이터를 delta로 변환

query = """
CONVERT TO DELTA 
parquet.`/workspace/spark/deltalake/delta_local/fish_parquet_partitioned/`
PARTITIONED BY (Species STRING)
"""
spark.sql(query)

참고자료

• Delta Lake Documentation
• Delta Lake Quickstart - Apache Spark
• 𝑫𝒆𝒍𝒕𝒂 𝑳𝒂𝒌𝒆 𝒊𝒏 𝑨𝑾𝑺 - 𝑭𝒓𝒐𝒎 𝒁𝒆𝒓𝒐 𝒕𝒐 𝑯𝒆𝒓𝒐 𝒊𝒏 4 𝒉𝒐𝒖𝒓𝒔

1️⃣ 데이터를 흐름으로 이해한다

이들은 데이터를 단순한 값이 아니라, 흐름(flow) 으로 봅니다.

분산 시스템(HDFS, S3), 배치 vs. 스트림 처리 차이를 직관적으로 이해함.

저장 포맷(Parquet, Avro) 장단점을 빠르게 판단할 수 있음.

👉 데이터를 "어떻게 최적의 방식으로 전달할까?"를 먼저 고민합니다.

2️⃣ 단순한 코더가 아니라, 문제 해결자다

SQL만 잘하는 게 아닙니다. 복잡한 문제를 논리적으로 해결하는 능력이 뛰어납니다.

Python, Scala, Java를 능숙하게 다루며, 가독성 좋은 코드를 작성.

자료구조와 알고리즘을 이해하고, 성능 최적화 감각이 있음.

👉 이들은 "어떻게 하면 더 효율적으로 문제를 해결할까?"를 끊임없이 고민합니다.

3️⃣ 성능 최적화를 당연하게 여긴다

이들은 파이프라인을 만들고 끝이 아니라, 최적화를 필수 과정으로 생각합니다.

Spark 디버깅 능력이 뛰어나고, 병목 현상을 빠르게 찾음.

파티셔닝, 버케팅을 활용해 대용량 데이터도 빠르게 처리.

👉 그냥 작동하는 코드가 아니라, "최적의 성능을 내는 코드"를 만듭니다.

4️⃣ 클라우드를 제대로 활용한다

단순히 클라우드를 사용하는 게 아니라, 비용과 성능을 고려한 최적의 조합을 찾습니다.

AWS, GCP, Azure 활용에 능숙하며, EMR, Databricks, Kubernetes의 차이를 명확히 이해.

Terraform 같은 IaC(Infrastructure as Code) 도구를 능숙하게 다룸.

👉 "어떤 서비스를 사용할까?"가 아니라, "가장 효율적인 조합은 무엇일까?"를 고민합니다.

5️⃣ 데이터 품질을 철저히 관리한다

많은 데이터보다 신뢰할 수 있는 데이터가 중요함을 압니다.

데이터 검증 및 모니터링(Great Expectations 등) 활용.

데이터가 언제, 어디서, 어떻게 생성되었는지 명확히 추적 가능.

👉 "이 데이터가 정확한가?"를 항상 먼저 고려합니다.

💡 최상위 데이터 엔지니어들의 마인드셋

✅ 주인의식 – 문제를 끝까지 해결하는 태도. ✅ 호기심 – 단순한 사용이 아니라, 원리를 깊이 이해. ✅ 실행력 – 이론이 아니라, 실제로 실험하고 개선. ✅ 비즈니스 감각 – 데이터를 통해 실제 가치를 창출하는 사고방식.

👉 이들은 "좋은 엔지니어"가 아니라, "비즈니스에 기여하는 엔지니어"입니다.

✨ 최상위 1% 데이터 엔지니어는 이렇게 다르다

✅ 데이터를 흐름으로 이해하고, 최적의 설계를 고민함. ✅ 성능 최적화가 몸에 배어 있음. ✅ 클라우드를 단순 사용이 아니라, 효율적으로 활용함. ✅ 데이터 품질을 철저히 관리하며, 신뢰할 수 있는 데이터를 다룸. ✅ 단순한 코더가 아니라, 문제 해결 능력을 갖춘 전문가.

💡 최고의 데이터 엔지니어들은 단순히 툴을 잘 쓰는 것이 아니라, 데이터로 가치를 만드는 사람들입니다.

▶ 출처 : https://blog.det.life/i-interviewed-200-data-engineers-heres-what-separates-the-best-from-the-rest-3092524e5875

0. Delta Lake란 무엇인가?

Delta Lake는 data lakehouse 아키텍처를 구축할 수 있는 오픈 소스 스토리지 레이어로, 기존의 데이터 레이크(Lake)에 트랜잭션 기능과 데이터 무결성 보장을 추가하여 데이터 웨어하우스 수준의 신뢰성과 성능을 제공하는 기술입니다.

주로 데이터 엔지니어링, 데이터 분석, 데이터 파이프라인 등 데이터를 다루기 위한 다양한 에서 활용되며, 데이터의 변경 이력을 관리하는 기능까지 제공합니다.

1. Delta Lake가 등장한 배경

1) 클라우드 객체 저장소의 부상

과거에는 HDFS 기반의 물리적인 서버 자원을 활용한 데이터 레이크가 널리 사용되었지만, 최근에는 Amazon S3, Google Cloud Storage와 같은 클라우드 오브젝트 스토리지가 대체하고 있습니다. 이들의 장점은 다음과 같습니다.

✅ 사실상 무제한의 확장성 ✅ 사용한 만큼만 비용 지불 ✅ 높은 내구성 및 안정성 보장

데이터 처리시 사용되는 Apache Spark, Presto, Trino, Pandas, DuckDB 등의 도구들 역시 클라우드 객체 저장소와의 연동을 지원해주고 있으며, 클라우드 객체 저장소를 기반으로 데이터를 읽고 저장하는 과정이 표준으로 자리잡고 있습니다.

2) 기존 데이터 레이크의 한계

전통적인 데이터 레이크(Hadoop HDFS, AWS S3, Azure Data Lake 등)는 대량의 데이터를 저장하기에는 적합하지만, 파일 기반이기 때문에 데이터 웨어하우스 수준의 쿼리나 분석 기능을 제공하는데는 한계가 있었습니다. 주요한 문제점들은 아래와 같습니다.

1️⃣ 데이터 정합성 부족

• 데이터가 여러 파일로 분산 저장되기 때문에, 중간에 작업이 실패하면 데이터가 불완전하게 저장될 가능성이 존재합니다.
• 예를 들어, 한 테이블에 100개의 파일이 저장되어 있고, 새로운 데이터를 추가하는 도중 작업이 중단되거나 실패하게 된다면 데이터의 일부만 기록되어 데이터 정합성이 깨질 수 있는 것이죠.

2️⃣ ACID 트랜잭션 미지원

• 파일 형태로 관리되는 데이터레이크는 데이터베이스처럼 원자성(Atomicity), 일관성(Consistency), 격리성(Isolation), 지속성(Durability)을 보장하지 않습니다.
• 따라서 여러 사용자가 동시에 데이터를 수정하거나 추가할 경우, 예측할 수 없는 충돌이나 데이터 손실이 발생할 가능성이 높습니다.

3️⃣ Schema Evolution(스키마 변경) 문제

• 기존에 저장되어 있던 데이터의 스키마(컬럼, 데이터 타입 등)를 변경하려면 전체 데이터를 새롭게 덮어쓰는 등 큰 작업을 요하는 경우가 많습니다.

4️⃣ 데이터 버전 관리 부재

• 동일한 저장 경로에 데이터가 추가되거나 삭제될 경우, 기존 데이터와의 변경 사항을 쉽게 추적할 수 없습니다.
• 이전의 데이터로 되돌아가는 롤백 기능이 없어 작업중 데이터를 삭제하거나 덮어썼다면 다시 복구하기가 어렵습니다.

이러한 문제들을 해결하기 위해 Databricks는 Delta Lake를 개발하였고, 이를 2017년에 고객에게 제공한 후 2019년 오픈소스로 공개했습니다.

2. Delta Lake의 핵심 설계 원리

Delta Lake의 핵심 원리는 객체 저장소에 파일 기반으로 저장되어 있는 데이터들에 대하여 ACID 트랜잭션을 보장하는 것입니다. Delta Lake는 Apache Parquet을 기반으로 동작하며, Delta Log라는 추가적인 메타데이터 로그 파일을 사용하여 ACID 트랜잭션과 데이터 버전 관리를 수행합니다.

1️⃣ Raw Data → 원본 데이터 파일 (Parquet) 2️⃣ Delta Log → 모든 변경 사항이 기록되는 로그 파일

✅ ACID 트랜잭션 보장

• Delta Lake는 Parquet 파일을 기반으로 데이터를 저장하면서도 데이터와 같이 관리되는 delta_log 메타데이터 로그를 통해 ACID 트랜잭션을 지원합니다.
• 따라서 아래와 같은 기능 구현이 가능해졌습니다.
  • 여러 데이터를 동시에 업데이트 가능.
  • 이전 데이터로의 자유로운 롤백 가능.
  • 데이터 일관성을 유지하면서도 높은 성능 제공.

이렇게 Delta Log를 활용하면 데이터가 언제, 어떻게 변경되었는지 추적할 수 있고, 트랜잭션 충돌을 방지할 수도 있습니다.

3. Delta Lake의 핵심 기능

1️⃣ ACID 트랜잭션 지원

• Delta Lake는 ACID 트랜잭션을 보장하여 데이터 정합성을 유지합니다. 즉, 여러 사용자가 동시에 데이터를 수정하거나 삽입해도 데이터가 일관되게 유지됩니다.

2️⃣ 데이터 버전 관리 (Time Travel)

• Delta Lake는 모든 데이터 변경 사항을 버전 관리합니다. 따라서 파일 기반으로 저장되어 있지만 특정 시점의 데이터로 롤백이 가능합니다.
• 실수로 데이터를 삭제하거나 덮어 쓰더라도 과거 버전으로 복구가 가능한 것이죠. 예를 들어, "어제의 데이터 상태로 돌아가고 싶다!"라고 할 때, 별도의 백업 없이도 특정 버전을 선택하여 데이터 조회가 가능합니다.

3️⃣ 스키마 관리 (Schema Management)

• Schema Enforcement(스키마 강제 적용)
  • 기존 데이터와 맞지 않는 스키마가 들어오는 경우 오류를 발생시켜 데이터 무결성을 유지합니다.
• Schema Evolution(스키마 변경 지원)
  • 기존 테이블의 컬럼을 추가/삭제하는 등의 스키마 변경 작업을 쉽게 적용할 수 있습니다.

4️⃣ 데이터 파일 최적화 (File Compaction)

• 데이터 레이크에서는 작은 용량의 파일들이 디렉토리 내에 많이 생성될 경우 성능이 저하될 수 있습니다. Delta Lake는 자동으로 작은 파일들을 병합하여 데이터 저장을 최적화시킬 수 있습니다.
• 아래의 최적화 메소드를 활용하여 대규모 데이터를 다룰 때 성능 문제를 줄일 수 있습니다.
  • OPTIMIZE : 작은 파일을 합쳐서 성능을 개선
  • Z-Ordering : 여러 개의 컬럼을 기준으로 데이터를 정렬하여 쿼리 성능 향상

5️⃣ 데이터 정리 및 삭제 (Data Vacuuming)

• 저장된 데이터에 Retention Policy를 설정하면 일정 기간이 지난 불필요한 데이터를 자동으로 삭제할 수 있습니다.
• Delta Lake에서 데이터 파일을 삭제하게되면 정합성 문제가 발생할 수 있으므로 특정 시점 이전의 데이터를 삭제해야 할 경우에는 반드시 VACUUM 기능을 활용하여 파일을 삭제합니다.

6️⃣ 캐싱 (Caching)

• Delta Lake는 클러스터 내에서 데이터 및 메타데이터 캐싱을 제공하여 성능을 더욱 향상시킵니다.

7️⃣ 감사 로그 (Audit Logging)

• 모든 데이터 변경 내역을 기록하여 누가 언제 어떤 데이터를 수정했는지 추적 가능합니다.

8️⃣ 스트리밍 데이터 처리

• 배치 처리와 스트리밍 처리를 모두 지원합니다.
• Delta Lake 테이블을 스트리밍 원본 및 싱크(Sink)로 사용할 수 있습니다.

4. 기존의 데이터 아키텍처와의 비교

5. OUTRO

Delta Lake, Hudi, Iceberg와 같은 Open Table Format이 나온지는 몇 년이 되었지만 저는 최근에서야 본격적으로 학습을 시작했습니다. 이전 회사에서도 데이터를 Parquet 형태로 관리했지만, 주로 업데이트된 테이블을 덮어쓰거나 새로운 디렉토리를 생성하는 방식으로 버전 관리를 했었죠. 그런 경험이 있어서인지, 파일 기반 저장 + AICD 트랜잭션 지원 + 롤백 이 모든 기능이 파일 기반으로 가능한 Open Table Format은 저에게는 혁신적인 개념으로 다가왔습니다. 가까운 미래까지는 이 기술이 데이터를 저장하고 관리하는데 있어 표준으로 자리매김 할 것이라는 생각이 들었습니다. 이번 글에서는 Delta Lake의 이론적 개념을 정리해 보았고, 다음 글에서는 직접 실습을 진행하며 실무에서 어떻게 활용할 수 있을지 살펴보겠습니다.

참고 자료

• https://www.qlik.com/us/data-lake/delta-lake
• https://medium.com/@ansam.yousry/understand-delta-lake-a-comprehensive-guide-a0b4beaeafe3
• https://blog.det.life/i-spent-5-hours-understanding-more-about-the-delta-lake-table-format-b8516c5091eb

❗데이터 엔지니어링의 현실, 화려함 뒤에 숨겨진 10가지 뼈 때리는 진실🤕

대용량 데이터, 최첨단 기술, 실시간 분석, 멋진 대시보드…... 데이터 엔지니어링에 대한 이야기는 늘 화려하죠. 하지만 그 뒤에 숨겨진 매일의 고군분투, 혼란, 그리고 눈에 띄지 않는 노력들은 수면 위로 잘 드러나지 않습니다. 데이터 엔지니어링에 대한 솔직하고 현실적인 이야기들을 들려드릴게요.

1️⃣ 예상보다 훨씬 더 엉망진창인 데이터 🗑️

▪ 아무리 최첨단 ETL 파이프라인을 구축해도, 결국 시간의 80%는 데이터를 정리하고, 변환하고, 검증하는 데 쓰게 될 거예요. 오타, 누락된 값, 잘못된 형식, 중복된 기록들이 매일매일 여러분을 괴롭힐 겁니다. 👉 진실 → 최고의 데이터 파이프라인도 엉망인 소스 데이터를 커버할 순 없어요. 쓰레기를 넣으면 쓰레기가 나올 뿐!

2️⃣ 마법을 기대하는 사람들 🤖

▪ 기술 지식이 없는 사람들은 데이터 엔지니어를 마법사라고 생각해요. 손가락만 튕기면 망가진 보고서를 고치고, 쿼리 속도를 높이고, 실시간 대시보드를 뚝딱 만들 수 있다고 믿죠. 👉 진실 → 기대치를 관리하는 것도 업무의 일부입니다. "안돼요!" 라고 말하지 못하면, 영원히 불 끄는 소방관 신세를 벗어날 수 없을 거예요.

3️⃣ 확장은 구축보다 10배 더 어렵다 🤯

▪ 하루에 1만 건의 레코드를 처리하는 멋진 데이터 파이프라인을 만드는 건 쉬운 일이에요. 하지만 그 규모가 수백만, 수십억 건으로 늘어나면 완전히 다른 이야기가 되죠. 지연 시간 급증, 예상치 못한 병목 현상, 폭발적인 스토리지 비용 증가가 매일 여러분을 괴롭힐 거예요. 💣 👉 진실 → 효율적인 파이프라인뿐만 아니라, 회복 탄력성이 뛰어난 파이프라인이 필요해요.

4️⃣ 지루하지만 중요한 데이터 거버넌스 📜

▪ 모두가 실시간 분석과 AI 기반 인사이트를 원하지만, 데이터 업데이트 하나 잘못되면 핵심 대시보드가 망가지고 관련 부서들에서 연락이 올 수도 있어요. 🚨 👉 진실 → 메타데이터 관리, 데이터 계보 추적, 접근 제어 등은 재미없지만, 여러분의 직장을 지켜줄 거예요.

5️⃣ 배치 vs 스트리밍? "경우에 따라 다르다" 🤔

▪ 기술 블로그는 스트리밍 파이프라인을 엄청나게 홍보하지만, 실제 비즈니스 활용 사례의 90%는 배치 처리로도 충분히 해결할 수 있어요. 하지만 사람들은 비용과 복잡성을 고려하지 않고 무작정 "실시간"을 외치죠. 👉 진실 → 대부분의 경우 배치 처리가 더 저렴하고, 간단하고, 안정적이에요. 스트리밍은 정말 필요할 때만 구축하세요!

6️⃣ 반복적인 업무의 연속 🔁

▪ 항상 최첨단의 프로젝트들만 하는 건 아니에요. 때로는 똑같은 파이프라인을 유지 관리하고, 실패한 작업을 수정하고, 느린 쿼리를 열 번이나 다시 작성해야 할 수도 있죠. 👉 진실 → 최고의 데이터 엔지니어는 지루한 업무를 기꺼이 감수해요. 참신함보다는 안정성이 더 중요하니까요.

7️⃣ 모든 잘못은 데이터 엔지니어 탓? 🤷‍

▪ 대시보드에 오류가 있으면, 데이터 엔지니어가 가장 먼저 비난을 받아요. 소스 데이터가 잘못됐는지, 변환 설정이 잘못됐는지, 분석가가 결과를 잘못 해석했는지 아무도 묻지 않죠. 👉 진실 → 선제적으로 대처하세요. 모니터링 시스템을 구축하고, 알림을 설정하고, 작업 내용을 문서화하세요.

8️⃣ 만능 도구는 없다! 🛠️

▪ Databricks, Snowflake, Airflow, dbt 등 다양한 도구가 자동화와 간편함을 제공하지만, 결국 도구는 사용하는 사람만큼의 가치만 있을 뿐이에요. 🧰 👉 진실 → 새로운 도구를 쫓아다니는 대신, 기본 원리(SQL, 분산 시스템, 데이터 모델링 등)를 마스터하세요.

9️⃣ 성공적인 데이터 엔지니어링은 눈에 띄지 않는다 👻

▪ 모든 것이 순조롭게 실행되면 아무도 알아채지 못해요. 문제가 발생했을 때만 관심을 받죠. 성공적인 데이터 엔지니어는 너무나 안정적으로 일을 처리해서, 사람들이 그 존재를 잊어버리게 만드는 사람이에요. 👉 진실 → 끊임없는 칭찬을 원한다면, 데이터 엔지니어링은 당신에게 맞지 않을 거예요.

🔟 AI가 당신을 대체하는 게 아니라, 게으른 엔지니어가 대체될 것이다. 🤖

▪ AI가 ETL 작업을 자동화하고 SQL 쿼리를 생성하면서, 데이터 엔지니어링 일자리가 사라질 것이라고 믿는 사람들도 있어요. 😥 진실은... 게으른 엔지니어는 대체되겠지만, 훌륭한 엔지니어는 계속 성장해 나갈 거예요. 🌟 👉 진실 → AI는 위협이 아니라 도구예요. AI를 활용하는 방법을 배우세요.

그럼에도 불구하고 데이터 엔지니어링은 기술 분야에서 큰 영향력을 가진 직업 중 하나입니다. 현실적인 문제를 해결하고, 혼란을 다스리고, 안정적인 시스템을 구축하는 데 희열을 느낀다면, 분명 이 일을 사랑하게 될 거예요!

▶원글 출처 : https://medium.com/@shenoy.shashwath/10-hard-truths-about-data-engineering-no-one-tells-you-a9e080ecfef1