Arrow, Pandas, DuckDB로 다루는 연합 데이터

https://product.kyobobook.co.kr/detail/S000220221456

LUVIT EPL과 유튜브 데이터로 배우는 DuckDB | 이기준 - 교보문고

 

저는 Go 개발자입니다. 이 사실을 먼저 밝히고 싶습니다. 왜냐하면 이것이 앞으로 이야기할 많은 내용을 설명해 주기 때문입니다. 특히 Apache Arrow를 pandas 파이프라인에 추가하면 pandas가 진짜 실행 엔진처럼 동작할 것이라고 생각했던 이유와, 그것이 왜 제가 이해하는 데 필요 이상으로 오래 걸린 방식으로 틀렸는지를 설명해 줍니다.

또한 저는 이 실험 이전부터, 그리고 앞으로도 계속 Apache Arrow의 열성 팬입니다. Arrow 명세서를 읽었고, 컬럼형 포맷 문서도 읽었습니다. 메모리 레이아웃, 제로 카피 읽기, 그리고 왜 Arrow가 분석 시스템을 위한 공통 언어(lingua franca)인지에 대해 컨퍼런스에서 사람들을 지루하게 만든 적도 있습니다.

하지만 이 모든 경험도 제가 곧 하게 될 일을 준비시켜 주지는 못했습니다.

전제

Arrow 팬보이였던 저는 Arrow 기반 pandas가 일반 pandas보다 의미 있게 더 빠를 것이라고 믿었습니다. 메모리 레이아웃이 더 효율적이고, IO 경계에서의 변환 비용도 더 적으며, 표현 방식도 컬럼형이기 때문에 분석 워크로드에 적합합니다. 원칙적으로 이것이 사실이라는 것은 충분히 읽어서 알고 있었습니다. 다만 그것이 언제부터 중요하지 않게 되는지는 충분히 읽지 못했던 것 같습니다.

여기서 한 가지 언급해야 할 점이 있습니다. 저는 pandas를 이전에 한 번밖에 사용해 본 적이 없고, 그것도 또 다른 실패한 실험에서였습니다.

이번 실험은 그 믿음을 현실적인 워크로드를 통해 검증하기 위해 설계되었습니다.

세 개의 파이프라인. 동일한 연합 데이터 소스. 동일한 계산. 다른 실행 모델입니다.

• pandas — 대부분의 Python 데이터 엔지니어가 알고 있는 기준선
• pandas + Arrow — 전체 메모리가 Arrow 기반이며 로직은 동일
• DuckDB — 전체 파이프라인을 하나의 SQL 쿼리로 표현

이것은 도구 비교가 아닙니다. 실행 모델 비교입니다. 그리고 이것이 전체 실험의 핵심입니다. 솔직히 말하면, 올바른 실험을 설계할 수 있을 정도로 이 차이를 명확히 설명하는 데 꽤 오랜 시간이 걸렸습니다.

데이터셋은 충분히 현실적이었습니다.

• CSV의 사용자 100만 명
• PostgreSQL의 주문 1,000만 건
• Parquet의 이벤트 1억 건

이 세 파이프라인의 차이가 드러날 곳이 있다면 바로 여기였습니다.

파이프라인의 목표

세 파이프라인 모두 세션화된 이벤트 로그와 주문 이력을 바탕으로 국가별 행동 및 거래 요약 정보를 계산합니다.

사용자 단위로:

• 세션 이벤트 기반 참여도
• 모바일 대비 비모바일 상호작용 비율
• 주문 빈도 및 평균 주문 금액

국가 단위로 집계:

• 총 주문 수
• 평균 주문 금액
• 총 이벤트 수

전형적인 연합 데이터 기반 enrichment → reduction → aggregation 파이프라인입니다. 특별한 것은 없습니다. 의도적으로 그렇게 했습니다. 이런 작업에서 어려움을 겪는 시스템은 운영 환경에 갈 자격이 없습니다.

파이프라인 1: Pandas

이것은 대부분의 Python 데이터 엔지니어가 익숙하게 보는 기준선입니다. 표현력도 좋고 읽기도 쉽지만, 규모가 커지면 비용이 급격히 드러납니다.

users = pd.read_csv("users.csv", parse_dates=["signup_date"])
orders = pd.read_sql_query("SELECT * FROM orders", conn, parse_dates=["created_at"])
events = pd.read_parquet("events.parquet")

그 다음 필터링, 컬럼 생성, 세션 수준 집계를 수행합니다.

events = events[
 (events["event_type"].isin(["click", "view", "add_to_cart"])) &
 (events["duration_ms"] > 200)
]

events["is_mobile"] = (events["device"] == "mobile").astype(int)
events["is_engaged"] = (events["interaction_count"] > 5).astype(int)

session_agg = (
 events.groupby(["user_id", "session_id"])
 .agg(
 session_duration=("duration_ms", "sum"),
 session_events=("id", "count"),
 engaged_events=("is_engaged", "sum"),
 mobile_events=("is_mobile", "sum"),
 )
 .reset_index()
)

그 후 세션 집계를 다시 사용자 단위로 집계합니다.

user_event_features = (
 session_agg.groupby("user_id")
 .agg(
 avg_session_duration=("session_duration", "mean"),
 total_sessions=("session_id", "count"),
 total_events=("session_events", "sum"),
 )
 .reset_index()
)

아직 조인도 하지 않았습니다. 이미 1억 건의 이벤트에 대해 두 번의 전체 스캔을 수행했고, 두 번 모두 메모리에 완전히 적재한 뒤 다음 단계로 넘어갑니다. 이후 주문 데이터도 같은 방식으로 처리하고, 조인, 결측값 처리, 최종 집계를 모두 Python이 단계별로 조정합니다.

이것이 제가 말하는 오케스트레이션 엔진입니다. Pandas는 계산 계획을 세우지 않습니다. 작성된 순서 그대로 실행하며, Python이 모든 단계 사이에 존재하고 중간 결과는 메모리에 완전히 저장됩니다.

결과는 다음과 같습니다.

133.96초, 최대 메모리 6,480MB

파이프라인 2: Pandas + Arrow (이쯤에서 눈치챘어야 했습니다)

가설은 단순했습니다.

Arrow는 pandas에 더 좋은 메모리 레이아웃을 제공하므로 전체 파이프라인도 더 빨라질 것입니다. 복사가 줄고, 메모리 압박이 줄고, 경계 변환도 감소합니다. 더 깔끔한 시스템이 될 것입니다.

저는 이것이 불완전한 생각이라는 사실을 알 만큼 Arrow를 충분히 읽었지만, 그대로 진행했습니다.

users = pd.read_csv(
 "users.csv",
 engine="pyarrow",
 dtype_backend="pyarrow",
 parse_dates=["signup_date"]
)

orders = pd.read_sql_query(
 "SELECT * FROM orders",
 conn,
 dtype_backend="pyarrow",
 parse_dates=["created_at"]
)

events = pd.read_parquet(
 "events.parquet",
 engine="pyarrow",
 dtype_backend="pyarrow"
)

변환 로직은 전혀 바꾸지 않았습니다. 새로운 로직을 시험한 것이 아니라, 더 나은 데이터 표현 방식이 시스템 동작을 바꿀 수 있는지를 확인하고 싶었습니다.

결과는 다음과 같습니다.

133.07초, 최대 메모리 6,859MB

Arrow가 개선하는 것은 다음과 같습니다.

• 메모리 지역성(memory locality)
• 제로 카피 경계 처리
• 상호 운용 비용

Arrow가 개선하지 못하는 것은 다음과 같습니다.

• 실행 계획
• 연산자 융합(operator fusion)
• 중간 결과 물질화 전략

Arrow는 데이터가 메모리에 어떻게 배치되는지를 최적화합니다. 하지만 계산이 데이터를 통해 어떻게 흐르는지는 최적화하지 않습니다.

(단일 실행 기준 측정이며, 이 워크로드에서 변동성은 시스템 간 차이에 비해 매우 작았습니다.)

실제로 일어난 일은 제가 기대했던 것보다 훨씬 단순했습니다.

Arrow 기반 pandas가 쓸모없다는 뜻은 아닙니다. 일부 복사를 줄이고, 넓은 테이블에서는 메모리 부담을 줄일 수 있습니다. 하지만 실행 모델은 변하지 않습니다. Python은 여전히 모든 단계를 조정합니다. 모든 groupby는 여전히 물질화됩니다. 모든 중간 결과는 굳이 필요하지 않은 회의에 참석하듯 메모리에 등장합니다.

비용 곡선은 약간 움직였지만, 곡선의 형태 자체는 변하지 않았습니다.

1억 건의 이벤트와 두 번의 집계 작업 앞에서는 이 차이가 모든 것을 삼켜버립니다.

Pandas는 여전히 RAM을 소모하면서도 미안해하는 예의 바른 스프레드시트입니다.

Arrow는 제가 원한다고 해서 쿼리 엔진이 되지 않습니다.

저는 배선을 업그레이드했을 뿐이고, 기계 자체는 그대로였습니다.

파이프라인 3: DuckDB

Python에서 데이터 이동을 조정하는 대신, 전체 워크플로를 쿼리 엔진에 맡기고 실행 계획을 수립하도록 합니다.

conn = duckdb.connect(database=":memory:")
conn.execute("INSTALL postgres; LOAD postgres")
conn.execute(
 "ATTACH 'host=localhost port=5432 dbname=postgres user=postgres password=secret' AS pg (TYPE POSTGRES)"
)

그런 다음 전체 파이프라인을 하나의 SQL 쿼리로 작성합니다.

WITH users AS (
    SELECT *
    FROM read_csv_auto('users.csv')
),

filtered_events AS (
    SELECT
        user_id,
        session_id,
        duration_ms,
        interaction_count,
        CASE WHEN device = 'mobile' THEN 1 ELSE 0 END AS is_mobile,
        CASE WHEN interaction_count > 5 THEN 1 ELSE 0 END AS is_engaged
    FROM read_parquet('events.parquet')
    WHERE event_type IN ('click', 'view', 'add_to_cart')
      AND duration_ms > 200
),

session_agg AS (
    SELECT
        user_id,
        session_id,
        SUM(duration_ms) AS session_duration,
        COUNT(*) AS session_events,
        SUM(is_engaged) AS engaged_events,
        SUM(is_mobile) AS mobile_events
    FROM filtered_events
    GROUP BY user_id, session_id
),

user_event_features AS (
    SELECT
        user_id,
        AVG(session_duration) AS avg_session_duration,
        COUNT(session_id) AS total_sessions,
        SUM(session_events) AS total_events
    FROM session_agg
    GROUP BY user_id
),

orders_agg AS (
    SELECT
        user_id,
        COUNT(id) AS total_orders,
        AVG(amount) AS avg_order_amount
    FROM pg.orders
    GROUP BY user_id
),

user_enriched AS (
    SELECT
        u.id,
        u.country,
        COALESCE(ue.total_events, 0) AS total_events,
        COALESCE(o.total_orders, 0) AS total_orders,
        COALESCE(o.avg_order_amount, 0) AS avg_order_amount
    FROM users u
    LEFT JOIN user_event_features ue
        ON u.id = ue.user_id
    LEFT JOIN orders_agg o
        ON u.id = o.user_id
)

SELECT
    country,
    SUM(total_orders) AS total_orders,
    AVG(avg_order_amount) AS avg_order_amount,
    SUM(total_events) AS total_events
FROM user_enriched
GROUP BY country
ORDER BY country;

결과는 다음과 같습니다.

4.77초, 최대 메모리 3,105MB

동일한 연합 데이터 소스에서 동일한 1억 건의 이벤트를 처리했을 때 Pandas보다 28배 빠르고 메모리는 52% 적게 사용했습니다.

그 이유는 DuckDB가 단순히 더 빠르기 때문이 아닙니다. DuckDB는 ETL, 쿼리 계획, 실행 사이의 경계를 사실상 제거하기 때문입니다. 각 CTE를 독립적인 중간 결과로 물질화하는 대신, 전체 쿼리에 대해 하나의 통합 실행 계획을 생성합니다.

구체적으로 말하면 filtered_events 단계는 독립적인 데이터셋으로 완전히 다시 물질화되지 않습니다. 대신 영구적으로 저장되는 테이블이 아니라 집계 단계로 이어지는 스트리밍 파이프라인처럼 동작합니다.

Predicate Pushdown은 스캔 계층까지 내려가 적용됩니다. 따라서 event_type과 duration_ms에 대한 필터는 데이터가 파이프라인에 완전히 진입하기 전에 불필요한 행을 제거합니다. 집계 연산은 서로 융합됩니다. 또한 Postgres Attach 기능을 통해 DuckDB는 원격 데이터 소스를 전체 계획의 일부로 인식할 수 있으며, Python으로 전체 테이블을 먼저 가져온 뒤 처리할 필요가 없습니다.

사라지는 것은 바로 Pandas가 비용을 지불하던 부분들입니다. 단계 간 Python 수준의 오케스트레이션, 반복적인 중간 데이터셋 물질화, 그리고 계산 전체를 이해하지 못한 채 개별 단계만 실행하는 구조적 한계가 제거됩니다.

숫자가 실제로 말해주는 것

동일한 1억 건 이벤트 워크로드를 요약하면 다음과 같습니다.

• pandas — 133.96초, 6,480MB
• pandas + Arrow — 133.07초, 6,859MB
• DuckDB — 4.77초, 3,105MB

가장 오래 곱씹어 볼 만한 결과는 Arrow입니다.

이 결과는 중립적이지 않습니다. 실행 시간은 약간 빨라졌지만 최대 메모리 사용량은 오히려 의미 있게 증가했습니다. 이는 메모리 측면에서 가설이 예측한 것과 정반대의 결과입니다. 두 수치 모두 기준선과 비교하면 사실상 오차 범위 안에 있습니다.

Arrow는 IO 경계에서 실제로 의미 있는 일을 했습니다. 데이터 표현 방식도 바꾸었습니다. 하지만 이 워크로드에서는 그것이 중요하지 않았습니다. 비용의 원인은 데이터 표현 방식이 아니라 실행 전략이었기 때문입니다. 그리고 Arrow는 실행 전략을 바꾸지 않습니다.

Pandas와 Arrow 파이프라인이 느린 이유는 구현이 잘못되었기 때문이 아닙니다. Python이 모든 단계를 조정하고 각 단계 사이에서 중간 결과가 물질화되는 실행 모델에서 동작하기 때문입니다. 이 모델에는 구조적인 한계가 있으며, 백엔드를 교체한다고 해서 그 한계가 높아지지는 않습니다.

DuckDB는 완전히 다른 모델에서 동작합니다. 계산을 한 번 계획하고, Python의 개입 없이 실행하며, 독립적인 객체로 존재할 필요가 없는 결과는 물질화하지 않습니다.

28배의 차이는 우연이 아닙니다.

그것은 오케스트레이션과 실행 사이의 차이입니다.

일반화

데이터 엔지니어링에서는 성능 문제를 라이브러리 문제로 생각하려는 유혹이 있습니다. 백엔드를 교체하고, 포맷을 바꾸고, 데이터 표현 방식을 업그레이드합니다. 이러한 개입은 실제로 의미가 있으며 때로는 올바른 선택이기도 합니다.

하지만 백엔드 교체만으로 해결할 수 없는 성능 문제가 있습니다. 제약 조건이 데이터 포맷이 아니라 실행 모델에 있기 때문입니다. 워크로드가 충분히 크고 처리 단계가 충분히 많아지면, 물질화된 중간 결과 사이를 오가는 Python 수준의 오케스트레이션 비용이 다른 모든 비용을 압도하게 됩니다. 그 시점에 필요한 것은 더 나은 메모리 레이아웃이 아닙니다. 전체 계산을 한 번에 계획할 수 있는 시스템입니다.

데이터 레이아웃을 아무리 최적화해도 더 나은 실행 계획에는 결국 밀릴 수 있습니다.

Arrow는 분석 데이터를 위한 올바른 포맷입니다. 현대 데이터 스택의 공통 언어이며, 저는 앞으로도 컨퍼런스에서 그것에 대해 열정적으로 이야기할 것입니다. 하지만 Arrow는 실행 엔진이 아닙니다. 그리고 저는 이 실험에서 적지 않은 시간을 들여 Arrow가 실행 엔진처럼 동작하기를 기대했습니다.

데이터 표현 방식을 개선한다고 실행 계획이 생기는 것은 아닙니다.

실행 계획은 계산이 어떻게 수행될지를 결정하는 시스템 자체를 바꿀 때 얻을 수 있습니다.

그리고 데이터를 보면, 그렇게 하지 않았을 때 어떤 일이 일어나는지 매우 명확하게 드러납니다.

 

<출처: https://medium.com/@tfmv/arrow-pandas-and-duckdb-on-federated-data-8e9f22d1d49e>