Dask
Dask는 Python을 확장합니다. Pandas/NumPy처럼 보이지만 클러스터에서 실행됩니다. 2025년 업데이트는 고성능 셔플 및 GPU 통합에 중점을 둡니다.
사용 시기
- 메모리보다 큰 테이블형 연산: 단일 머신 또는 멀티노드 클러스터에서 10GB~1TB CSV/Parquet 데이터셋 처리.
- 사용자 정의 Python 워크플로우 병렬화:
dask.delayed를 사용하여 임의의 Python 루프 및 태스크 그래프 병렬화. - 분산 머신러닝: 분산 작업자 노드에 걸쳐 Scikit-learn, XGBoost, LightGBM과 통합.
- NumPy 및 Pandas 확장: 대용량 배열 및 DataFrame 계산을 위한 익숙한 API를 즉시 사용.
빠른 시작
import dask.dataframe as dd
# 멀티 기가바이트 파티션된 CSV 파일을 느긋하게 읽기
df = dd.read_csv('data/transactions_*.csv')
# 모든 CPU 코어에서 병렬로 집계 통계 계산
result = df.groupby('category').amount.mean().compute()
print(result)
핵심 개념
#Dask DataFrame을 이용한 대용량 처리
파티션된 Parquet 파일을 로드하고 집계를 느긋하게 계산:
import dask.dataframe as dd
from dask.distributed import Client
# 로컬 분산 클러스터 초기화
client = Client(n_workers=4, threads_per_worker=2, memory_limit='4GB')
print(f"Dask Dashboard 실행 중: {client.dashboard_link}")
# 다중 파일 데이터셋을 느긋하게 로드
df = dd.read_parquet(
's3://analytics-bucket/transactions/*.parquet',
columns=['customer_id', 'amount', 'status', 'country'],
engine='pyarrow'
)
# 계산 그래프를 느긋하게 구축
filtered = df[df['status'] == 'completed']
metrics = filtered.groupby('country')['amount'].agg(['mean', 'sum', 'count'])
# 그래프를 실행하고 구체적인 Pandas DataFrame 반환
result_df = metrics.compute()
print(result_df)
#dask.delayed를 이용한 사용자 정의 태스크 그래프
독립적인 함수 실행 병렬화:
from dask import delayed, compute
import time
@delayed
def fetch_api_data(endpoint: str) -> dict:
time.sleep(0.5) # I/O 시뮬레이션
return {'endpoint': endpoint, 'status': 200, 'items': [1, 2, 3]}
@delayed
def transform_payload(data: dict) -> int:
return sum(data['items'])
@delayed
def aggregate_totals(totals: list[int]) -> int:
return sum(totals)
# 종속성 DAG 구축
endpoints = ['users', 'orders', 'inventory', 'payments']
fetched = [fetch_api_data(ep) for ep in endpoints]
transformed = [transform_payload(f) for f in fetched]
final_total = aggregate_totals(transformed)
# 클러스터에서 모든 태스크를 병렬로 실행
total_result = final_total.compute()
print("총합:", total_result)
#분산 선형 대수를 위한 Dask Array
대규모 다차원 배열 조작:
import dask.array as da
# 50,000 x 50,000 배열을 5,000 x 5,000 청크로 분할하여 생성
x = da.random.normal(10, 0.1, size=(50000, 50000), chunks=(5000, 5000))
# 느긋하게 계산 수행
y = x + x.T
z = y[::2, ::2].mean(axis=0)
# 결과 계산
result = z.compute()
print("계산된 배열 평균 형태:", result.shape)
일반적인 패턴
비정규 병렬 태스크 그래프를 위한 분산 Futures
문제: DataFrame/Array의 행/열 구조에 깔끔하게 맞지 않는 사용자 정의 병렬 계산.
해결책: 비동기 futures와 함께 Dask Distributed Client 사용:
from dask.distributed import Client
client = Client(n_workers=4, threads_per_worker=2)
def train_partition(part_id: int):
# 로컬 하위 모델 학습
return f"Model {part_id} trained"
futures = [client.submit(train_partition, i) for i in range(10)]
results = client.gather(futures)
print(results)
모범 사례 (2026)
- 해야 할 일: 메모리 압력, 태스크 스트림, 병목 현상을 모니터링하려면 Dask Web Dashboard(일반적으로 포트 8787)를 확인하세요.
- 해야 할 일: 최적의 병렬 효율성을 위해 메모리에서 청크 크기를 100MB~300MB 사이로 선택하세요.
- 해야 할 일: 동일한 변환 데이터를 반복적으로 쿼리할 때 중간 DataFrame을 유지하세요(
df = df.persist()). - 해야 할 일: 열 프로젝션 및 조건자 푸시다운을 사용하여 초기에 열과 행을 필터링하세요.
- 하지 말아야 할 일: 루프 내에서
.compute()를 호출하지 마세요. 전체 태스크 그래프를 구축하고compute()를 한 번만 호출하세요. - 하지 말아야 할 일: 데이터가 RAM에 쉽게 맞는다면 Dask를 사용하지 마세요. 메모리 내 작업에는 네이티브 Pandas 및 Polars가 훨씬 빠릅니다.
- 하지 말아야 할 일: 수백만 개의 작은 지연 태스크를 만들지 마세요. 과도한 태스크 오버헤드는 스케줄러 성능을 저하시킵니다.
문제 해결
| 오류 | 원인 | 해결책 |
|---|---|---|
KilledWorker: Worker exceeded 95% memory budget | 계산 중 태스크 파티션이 작업자 RAM에 비해 너무 큼. | 파티션 수 늘리기: df.repartition(npartitions=100). |
UserWarning: Sending large object to workers | 직접 함수 인수로 전달된 큰 NumPy 배열 또는 DataFrame. | 태스크 제출 전에 client.scatter(data)로 데이터를 한 번 분산. |
Slow compute: Too many tiny tasks | 수천 개의 마이크로 파티션으로 인한 태스크 스케줄링 오버헤드. | repartition(partition_size="100MB")를 사용하여 작은 파티션 병합. |