데이터 엔지니어를 위한 DuckDB

https://2stndard.tistory.com/notice/203

[발간예정][EPL과 유튜브로 배우는 DuckDB] 실습 코드와 데이터

 

데이터 엔지니어링에는 Spark를 쓰기에는 너무 작고, Pandas로 처리하기에는 너무 느리며, Snowflake 웨어하우스를 띄우기에는 너무 번거로운 문제들이 있습니다. 2GB짜리 Parquet 파일을 읽어 스키마를 확인하는 일, 벤더가 전달한 CSV 덤프에서 간단한 집계를 실행하는 일, 복잡한 SQL 변환 로직을 운영 환경에 배포하기 전에 로컬에서 테스트하는 일 등이 그렇습니다. 수년 동안 이런 문제에 대한 답은 “그냥 Pandas를 쓰자” 또는 “그냥 웨어하우스에 연결하자”였습니다. 하지만 대부분의 경우 DuckDB가 더 나은 답입니다.

DuckDB는 인프로세스 분석 데이터베이스입니다. Python 프로세스 안에서 실행되며, 서버도, Docker 컨테이너도, 별도의 설정도 필요하지 않습니다. 컬럼 기반이고 벡터화되어 있으며, 노트북 환경에서도 수백 MB 이상의 분석 워크로드에서는 대부분 Pandas보다 빠릅니다. SQL을 사용할 수 있고, Parquet과 CSV를 네이티브로 읽으며, Pandas 및 Polars DataFrame과 데이터 복사 없이 통합됩니다. 설치는 pip install 한 줄이면 충분합니다.

이 글에서는 DuckDB가 데이터 엔지니어링 워크플로에서 자리를 차지할 만한 네 가지 사용 사례를 살펴봅니다.

 

사용 사례 1: 로컬 파일 처리를 위해 Pandas 대체하기

Python에서 대용량 CSV 또는 Parquet 파일을 처리하는 일반적인 패턴은 보통 다음과 같습니다:

import pandas as pd

df = pd.read_parquet("orders.parquet")
result = (
    df[df["status"] == "completed"]
    .groupby("customer_id")
    .agg(total_spend=("amount", "sum"), order_count=("order_id", "count"))
    .sort_values("total_spend", ascending=False)
    .head(100)
)

 

파일이 충분히 커져서 read_parquet가 사용 가능한 메모리보다 더 많은 데이터를 로드하게 되면, 이 방식은 더 이상 제대로 동작하지 않습니다. Pandas는 필터링이 일어나기 전에 전체 파일을 먼저 읽습니다. 10개 컬럼을 가진 4GB Parquet 파일에서 실제로 필요한 컬럼이 2개뿐이라도, 여전히 10개 컬럼 전체를 메모리에 로드합니다.

DuckDB는 이런 방식으로 동작하지 않습니다. 조건식과 컬럼 선택을 파일 읽기 단계로 밀어 넣기 때문에, 실제로 필요한 데이터만 로드됩니다:

import duckdb

result = duckdb.sql("""
    select
        customer_id,
        sum(amount)    as total_spend,
        count(order_id) as order_count
    from read_parquet('orders.parquet')
    where status = 'completed'
    group by customer_id
    order by total_spend desc
    limit 100
""").df()

 

마지막의 .df()는 결과를 Pandas DataFrame으로 변환합니다. 메모리에 여유롭게 적재 가능한 파일이라면 성능 차이는 크지 않습니다. 하지만 그렇지 않은 경우에는 DuckDB는 계속 동작하는 반면, Pandas는 메모리가 부족해지거나 디스크 스와핑이 발생하기 시작합니다.

DuckDB는 또한 글롭(glob) 패턴을 사용해 하나의 쿼리에서 여러 파일을 읽을 수 있으며, 이는 파티셔닝된 데이터셋을 다룰 때 특히 유용합니다.

result = duckdb.sql("""
    select
        date_trunc('month', order_date) as month,
        sum(amount)                     as revenue
    from read_parquet('data/orders/year=2024/month=*/*.parquet')
    group by 1
    order by 1
""").df()

 

파일을 반복문으로 순회할 필요도 없고, pd.concat도 필요 없으며, 중간 DataFrame도 만들 필요가 없습니다.

사용 사례 2: SQL로 DataFrame 조회하기

때로는 DataFrame이 이미 메모리에 존재하는 경우가 있습니다. 예를 들어 이전 단계의 변환 결과이거나, Pandas가 잘 처리하는 데이터 소스로부터 생성된 경우입니다. DuckDB는 별도의 데이터 이동 없이 이를 직접 조회할 수 있습니다:

import pandas as pd
import duckdb

orders = pd.read_csv("orders.csv")
customers = pd.read_csv("customers.csv")
result = duckdb.sql("""
    select
        c.country,
        count(distinct o.customer_id)   as customer_count,
        sum(o.amount)                   as total_revenue,
        avg(o.amount)                   as avg_order_value
    from orders o
    join customers c on o.customer_id = c.id
    where o.status = 'completed'
      and o.order_date >= '2024-01-01'
    group by c.country
    having sum(o.amount) > 10000
    order by total_revenue desc
""").df()

 

DuckDB는 SQL 쿼리 안에서 Python 변수인 orders와 customers를 직접 참조합니다. to_sql() 호출도 필요 없고, SQLite 연결도 필요 없으며, 임시 테이블도 만들 필요가 없습니다. DataFrame 객체를 메모리 상에서 그대로 스캔합니다.

이 패턴은 변환 로직 자체는 SQL로 표현하는 것이 자연스럽지만, 주변 코드가 Python으로 구성되어 있을 때 특히 유용합니다. SQL 로직을 여러 단계의 Pandas 연산 체인으로 다시 구현할 필요 없이, SQL을 그대로 실행하고 결과를 다시 DataFrame으로 받아올 수 있습니다.

사용 사례 3: dbt와 웨어하우스 SQL의 로컬 테스트

데이터 엔지니어링에서 비용이 많이 드는 습관 중 하나는 Snowflake나 BigQuery 웨어하우스를 개발 환경처럼 사용하는 것입니다. 개발 중 실행되는 모든 dbt run은 웨어하우스를 대상으로 컴파일되고 실행되며, 이는 비용이 발생하고 네트워크 연결이 필요하며 로컬에서 실행하는 것보다 느립니다.

DuckDB에는 dbt 모델을 프로덕션 웨어하우스 대신 로컬 DuckDB에서 실행할 수 있게 해주는 dbt 어댑터가 있습니다. 설정에 필요한 것은 두 가지입니다. 어댑터와 로컬 프로필입니다.

 

pip install dbt-duckdb

# profiles.yml
my_project:
  target: dev
  outputs:
    dev:
      type: duckdb
      path: /tmp/dev.duckdb
    prod:
      type: snowflake
      account: "{{ env_var('SNOWFLAKE_ACCOUNT') }}"
      # ... rest of Snowflake config

 

이렇게 설정하면 dbt run --target dev 명령이 로컬 DuckDB 파일을 대상으로 실행됩니다. 표준 SQL과 dbt 매크로를 사용하는 모델은 별도의 수정 없이 그대로 실행됩니다. 개발자는 로컬 환경에서 반복적으로 개발하고 검증한 뒤, 로직이 준비되었을 때만 Snowflake에서 실행하면 됩니다.

 

로컬 개발 환경에 실제와 유사한 데이터를 준비하려면, 인증 정보가 설정되어 있는 경우 DuckDB는 S3에서 직접 데이터를 읽을 수 있습니다:

import duckdb

conn = duckdb.connect("/tmp/dev.duckdb")
conn.execute("install httpfs; load httpfs;")
conn.execute("""
    create or replace table raw_orders as
    select *
    from read_parquet('s3://data-lake/raw/orders/2024-01/*.parquet')
    limit 100000
""")

 

이 방식은 데이터 레이크에서 대표성 있는 샘플 데이터를 로컬 DuckDB 파일로 가져오고, 이후 dbt가 이를 소스 데이터로 사용하도록 합니다.

워크플로는 다음과 같이 단순해집니다.
한 번 샘플링하고, 로컬에서 반복 개발하고, 준비되면 배포합니다.

사용 사례 4: 경량 데이터 품질 검사

웨어하우스를 대상으로 데이터 품질 검사를 실행하는 것은 dbt를 로컬에서 실행할 때와 같은 문제를 가집니다. 실시간 연결이 필요하고, 비용이 발생하며, 로컬 파일만으로 수행할 수 있는 검사임에도 불구하고 불필요하게 느립니다.

DuckDB를 사용하면 데이터가 웨어하우스에 적재되기 전에 Parquet 파일을 직접 대상으로 품질 검사를 실행하는 것이 현실적인 선택이 됩니다:

import duckdb
from dataclasses import dataclass

@dataclass
class CheckResult:
    check: str
    passed: bool
    failing_rows: int

def run_quality_checks(path: str) -> list[CheckResult]:
    conn = duckdb.connect()
    checks = {
        "no_null_order_ids": """
            select count(*) from read_parquet(?)
            where order_id is null
        """,
        "positive_amounts": """
            select count(*) from read_parquet(?)
            where amount <= 0
        """,
        "valid_status": """
            select count(*) from read_parquet(?)
            where status not in ('pending', 'completed', 'cancelled')
        """,
        "no_future_dates": """
            select count(*) from read_parquet(?)
            where order_date > current_date
        """,
    }
    results = []
    for name, sql in checks.items():
        failing = conn.execute(sql, [path]).fetchone()[0]
        results.append(CheckResult(
            check=name,
            passed=failing == 0,
            failing_rows=failing
        ))
    return results

results = run_quality_checks("data/orders_2024_01.parquet")
for r in results:
    status = "PASS" if r.passed else f"FAIL ({r.failing_rows} rows)"
    print(f"{r.check}: {status}")

 

이 방식은 웨어하우스 연결 없이, Spark 없이, 그리고 파일을 Pandas DataFrame으로 로드하지 않고도 실행됩니다. 수백 MB에서 수 GB 규모의 파일이라면 몇 초 안에 검사를 완료할 수 있습니다.

같은 패턴은 적재(load) 단계 이전에 실행되는 Airflow 태스크에도 통합할 수 있습니다. 이를 통해 품질 검사를 통과하지 못한 파일은 웨어하우스에 도달하기 전에 미리 차단할 수 있습니다:

from airflow.decorators import dag, task
from airflow.exceptions import AirflowFailException
import duckdb

@dag(schedule="@daily")
def orders_pipeline():
    @task()
    def validate_file(path: str):
        conn = duckdb.connect()
        failing = conn.execute("""
            select count(*) from read_parquet(?)
            where order_id is null or amount <= 0
        """, [path]).fetchone()[0]
        if failing > 0:
            raise AirflowFailException(
                f"Quality check failed: {failing} invalid rows in {path}"
            )

    @task()
    def load_to_snowflake(path: str):
        ...
    path = "s3://data-lake/raw/orders/{{ ds }}.parquet"
    validate_file(path) >> load_to_snowflake(path)

 

DuckDB가 아닌 것 (What DuckDB is not)

DuckDB는 인프로세스(in-process) 데이터베이스입니다. 하나의 프로세스 안에서 실행되기 때문에 수평 확장(horizontal scaling)을 지원하지 않습니다. 데이터셋이 단일 머신의 사용 가능한 RAM을 초과하면, Pandas와 동일한 방식으로 한계에 부딪히게 됩니다. 다만 그 한계점이 더 높을 뿐입니다. 진정한 대규모 처리에서는 여전히 Spark나 웨어하우스 기반의 컴퓨트 환경이 적절한 선택입니다.

또한 DuckDB는 프로덕션 웨어하우스를 대체하는 도구도 아닙니다. 접근 제어 기능이 없고, 프로세스 간 공유가 불가능하며, 쿼리 이력 관리 기능도 없고, 동시 쓰기(concurrent writes)도 지원하지 않습니다. 앞에서 소개한 사용 사례들은 모두 로컬 또는 단일 프로세스 환경의 워크로드입니다. 예를 들어 개발, 검증, 탐색, 적재(load) 이전 단계의 경량 변환 작업 등이 이에 해당합니다.

이러한 범위 안에서 DuckDB는 데이터 엔지니어링 워크플로에 매우 유용한 도구 중 하나입니다. 별다른 설정 없이 바로 사용할 수 있으며, 과거에는 전체 웨어하우스 연결이나 Pandas 기반 우회 방법이 필요했던 문제 영역을 효과적으로 해결해주기 때문입니다.

 

<출처: https://medium.com/data-engineer-things/duckdb-for-data-engineers-8e885367fcd1>