Parquet에서 Snowflake로: 스마트하게 쿼리하고 빠르게 로드

Source: Dev.to

문제

대용량 금융 데이터를 다룰 때, 효율적인 쿼리와 결과를 Snowflake와 같은 데이터 웨어하우스에 로드하는 것이 매우 중요합니다. 목표는 AWS S3에 저장된 3 TB 규모의 Parquet 데이터에서 일일 메트릭(총 거래량, 활성 고객 수, 평균 잔액)을 생성하는 것입니다. 데이터는 transaction_date 기준으로 파티션되어 있지만, 오래된 파티션에서는 컬럼 이름이 일관되지 않습니다. 결과는 추가 분석을 위해 Snowflake에 로드되어야 합니다.

접근 방식

데이터를 효율적으로 쿼리하기

파티션 프루닝을 사용해 최근 30일치 데이터만 읽어들임으로써 시간과 비용을 절감합니다.

스키마 진화 처리

COALESCE와 같은 SQL 함수를 활용해 누락되었거나 이름이 다른 컬럼을 처리하고, 파티션 간에 일관된 스키마를 유지합니다.

메트릭 집계

지역별로 데이터를 집계하여 총 거래량, 활성 고객 수, 평균 계좌 잔액을 계산합니다.

Snowflake에 데이터 로드

처리 후, Snowflake의 COPY INTO 메서드를 사용해 CSV 파일을 웨어하우스로 효율적으로 대량 적재합니다.

왜 이 방법이 효과적인가

• 파티션 프루닝은 쿼리를 필요한 데이터에만 제한하므로 빠르고 비용 효율적입니다.
• COALESCE를 이용한 스키마 처리는 진화하는 파티션 간의 원활한 통합을 가능하게 합니다.
• Snowflake의 최적화된 로딩 메커니즘은 빠르고 신뢰성 있는 데이터 전송을 지원합니다.

이 접근 방식은 클라우드 스토리지에 저장된 대규모 파티션 데이터셋을 관리하기 쉽게 만들며, 효율적인 처리와 Snowflake로의 로딩을 보장합니다.

PySQL 솔루션

파티션 프루닝을 사용해 최근 30일치 Parquet 읽기

import duckdb
import datetime

end = datetime.date.today()
start = end - datetime.timedelta(days=30)

con = duckdb.connect()
df = con.execute(f"""
    SELECT *
    FROM read_parquet('s3://bank-lake/transactions/transaction_date>= {start} AND transaction_date <= {end}/*.parquet')
""").fetchdf()

스키마 차이를 처리하면서 메트릭 집계

SELECT
    region,
    SUM(transaction_amount) AS total_tx,
    COUNT(DISTINCT customer_id) AS active_customers,
    AVG(COALESCE(account_balance, acct_balance)) AS avg_balance
FROM df
GROUP BY region

결과를 Snowflake에 로드

import snowflake.connector

# Export the aggregated DataFrame to CSV
result.to_csv("daily.csv", index=False)

# Connect to Snowflake
conn = snowflake.connector.connect(
    user='YOUR_USER',
    password='YOUR_PASSWORD',
    account='YOUR_ACCOUNT'
)

# Upload and copy the CSV into Snowflake
conn.cursor().execute("""
    PUT file://daily.csv @%DAILY_REGION_METRICS;
    COPY INTO DAILY_REGION_METRICS
    FROM @%DAILY_REGION_METRICS
    FILE_FORMAT=(TYPE=CSV FIELD_OPTIONALLY_ENCLOSED_BY='"');
""")