dask

dask은(는) 무엇을 하나요?

개요

Dask는 파이썬용 병렬 및 분산 컴퓨팅 라이브러리로, 세 가지 중요한 기능을 제공합니다:

• 메모리보다 큰 실행: 단일 머신에서 가용 RAM을 초과하는 데이터 처리
• 병렬 처리: 여러 코어를 활용한 계산 속도 향상
• 분산 컴퓨팅: 여러 머신에서 테라바이트 규모 데이터 지원

Dask는 랩톱(~100 GiB 처리)에서 클러스터(~100 TiB 처리)까지 확장 가능하며, 친숙한 Python API를 유지합니다.

현재 업스트림: dask 2026.3.0 (PyPI, 2026년 3월). 문서: docs.dask.org. 2025.1.0부터는 표현식 기반 DataFrame API가 유일한 구현체입니다. dask-expr를 별도로 설치하거나 dataframe.query-planning: False를 설정하지 마세요.

빠른 시작

설치

uv pip install "dask>=2025.1"

일반적인 pandas/NumPy 워크플로우(분산 스케줄러 및 대시보드 포함):

uv pip install "dask[complete]"

원격 객체 스토리지 (S3, GCS, Azure):

uv pip install s3fs    # s3:// 경로
uv pip install gcsfs   # gs:// 경로

Python 3.10+ 필요 (2024.12에서 3.9 지원 중단). DataFrame I/O는 PyArrow 16+ 필요 (dask 2026.1.2 기준).

이 스킬을 사용해야 하는 경우

다음과 같은 경우에 이 스킬을 사용하세요:

• 가용 RAM을 초과하는 데이터셋 처리
• pandas 또는 NumPy 연산을 더 큰 데이터셋으로 확장
• 성능 향상을 위한 계산 병렬화
• 여러 파일(CSV, Parquet, JSON, 텍스트 로그)을 효율적으로 처리
• 작업 종속성을 가진 커스텀 병렬 워크플로우 구축
• 여러 코어 또는 머신에 워크로드 분산

핵심 기능

Dask는 다섯 가지 주요 컴포넌트를 제공하며, 각각 다른 사용 사례에 적합합니다:

1. DataFrames - 병렬 Pandas 연산

목적: 병렬 처리를 통해 pandas 연산을 더 큰 데이터셋으로 확장합니다.

사용 시기:

• 테이블 형식 데이터가 가용 RAM을 초과할 때
• 여러 CSV/Parquet 파일을 함께 처리해야 할 때
• pandas 연산이 느리고 병렬화가 필요할 때
• pandas 프로토타입을 프로덕션으로 확장할 때

참조 문서: Dask DataFrames에 대한 포괄적인 안내는 references/dataframes.md를 참조하세요. 여기에는 다음이 포함됩니다:

• 데이터 읽기 (단일 파일, 여러 파일, glob 패턴)
• 일반적인 연산 (필터링, groupby, 조인, 집계)
• map_partitions을 사용한 커스텀 연산
• 성능 최적화 팁
• 일반적인 패턴 (ETL, 시계열, 다중 파일 처리)

빠른 예제:

import dask.dataframe as dd

# 여러 파일을 단일 DataFrame으로 읽기
ddf = dd.read_csv('data/2024-*.csv')

# 연산은 compute() 호출까지 지연됨
filtered = ddf[ddf['value'] > 100]
result = filtered.groupby('category').mean().compute()

핵심 사항:

• 연산은 .compute() 호출까지 지연됨 (태스크 그래프 구축)
• 효율적인 커스텀 연산을 위해 map_partitions 사용
• 다른 소스에서 구조화된 데이터를 다룰 때는 일찍 DataFrame으로 변환

2. Arrays - 병렬 NumPy 연산

목적: 블록 알고리즘을 사용하여 NumPy 기능을 메모리보다 큰 데이터셋으로 확장합니다.

사용 시기:

• 배열이 가용 RAM을 초과할 때
• NumPy 연산에 병렬화가 필요할 때
• 과학 데이터셋(HDF5, Zarr, NetCDF) 작업
• 병렬 선형 대수 또는 배열 연산이 필요할 때

참조 문서: Dask Arrays에 대한 포괄적인 안내는 references/arrays.md를 참조하세요. 여기에는 다음이 포함됩니다:

• 배열 생성 (NumPy로부터, 랜덤, 디스크에서)
• 청크 전략 및 최적화
• 일반적인 연산 (산술, 리덕션, 선형 대수)
• map_blocks를 사용한 커스텀 연산
• HDF5, Zarr, XArray와의 통합

빠른 예제:

import dask.array as da

# 청크로 큰 배열 생성
x = da.random.random((100000, 100000), chunks=(10000, 10000))

# 연산은 지연됨
y = x + 100
z = y.mean(axis=0)

# 결과 계산
result = z.compute()

핵심 사항:

• 청크 크기가 중요함 (청크당 ~100 MB 목표)
• 연산은 청크 단위로 병렬 처리됨
• 효율적인 연산을 위해 필요시 데이터 재청크화
• Dask에 없는 연산은 map_blocks 사용

3. Bags - 비정형 데이터의 병렬 처리

목적: 함수형 연산으로 비정형 또는 반정형 데이터(텍스트, JSON, 로그)를 처리합니다.

사용 시기:

• 텍스트 파일, 로그 또는 JSON 레코드 처리
• 구조화된 분석 전 데이터 정리 및 ETL
• 배열/데이터프레임 형식에 맞지 않는 Python 객체 작업
• 메모리 효율적인 스트리밍 처리 필요

참조 문서: Dask Bags에 대한 포괄적인 안내는 references/bags.md를 참조하세요. 여기에는 다음이 포함됩니다:

• 텍스트 및 JSON 파일 읽기
• 함수형 연산 (map, filter, fold, groupby)
• DataFrames로 변환
• 일반적인 패턴 (로그 분석, JSON 처리, 텍스트 처리)
• 성능 고려 사항

빠른 예제:

import dask.bag as db
import json

# JSON 파일 읽기 및 파싱
bag = db.read_text('logs/*.json').map(json.loads)

# 필터링 및 변환
valid = bag.filter(lambda x: x['status'] == 'valid')
processed = valid.map(lambda x: {'id': x['id'], 'value': x['value']})

# 분석을 위해 DataFrame으로 변환
ddf = processed.to_dataframe()

핵심 사항:

• 초기 데이터 정리에 사용한 후 DataFrame/Array로 변환
• 더 나은 성능을 위해 groupby 대신 foldby 사용
• 연산은 스트리밍 방식이며 메모리 효율적
• 복잡한 연산을 위해 구조화된 형식(DataFrame)으로 변환

4. Futures - 태스크 기반 병렬화

목적: 작업 실행 및 종속성을 세밀하게 제어하는 커스텀 병렬 워크플로우를 구축합니다.

사용 시기:

• 동적이고 진화하는 워크플로우 구축
• 즉시 태스크 실행 필요 (지연되지 않음)
• 런타임 조건에 따라 달라지는 계산
• 커스텀 병렬 알고리즘 구현
• 상태 저장 계산 필요

참조 문서: Dask Futures에 대한 포괄적인 안내는 references/futures.md를 참조하세요. 여기에는 다음이 포함됩니다:

• 분산 클라이언트 설정
• 태스크 제출 및 future 작업
• 태스크 종속성 및 데이터 이동
• 고급 조정 (큐, 락, 이벤트, 액터)
• 일반적인 패턴 (파라미터 스윕, 동적 태스크, 반복 알고리즘)

빠른 예제:

from dask.distributed import Client

client = Client()  # 로컬 클러스터 생성

# 태스크 제출 (즉시 실행)
def process(x):
    return x ** 2

futures = client.map(process, range(100))

# 결과 수집
results = client.gather(futures)

client.close()

핵심 사항:

• 분산 클라이언트 필요 (단일 머신의 경우에도)
• 제출 시 태스크가 즉시 실행됨
• 반복 전송을 피하기 위해 큰 데이터는 미리 scatter
• 태스크당 ~1ms 오버헤드 (수백만 개의 작은 태스크에는 부적합)
• 상태 저장 워크플로우에는 액터 사용

5. Schedulers - 실행 백엔드

목적: Dask 태스크가 실행되는 방식과 위치(스레드, 프로세스, 분산)를 제어합니다.

스케줄러 선택 기준:

• Threads (기본값): NumPy/Pandas 연산, GIL 해제 라이브러리, 공유 메모리 이점
• Processes: 순수 Python 코드, 텍스트 처리, GIL 바운드 연산
• Synchronous: pdb로 디버깅, 프로파일링, 오류 이해
• Distributed: 대시보드 필요, 다중 머신 클러스터, 고급 기능

참조 문서: Dask Schedulers에 대한 포괄적인 안내는 references/schedulers.md를 참조하세요. 여기에는 다음이 포함됩니다:

• 상세한 스케줄러 설명 및 특성
• 구성 방법 (전역, 컨텍스트 관리자, compute별)
• 성능 고려 사항 및 오버헤드
• 일반적인 패턴 및 문제 해결
• 최적 성능을 위한 스레드 구성

빠른 예제:

import dask
import dask.dataframe as dd

# DataFrame에는 스레드 사용 (기본값, 숫자 연산에 좋음)
ddf = dd.read_csv('data.csv')
result1 = ddf.mean().compute()  # 스레드 사용

# Python 중심 작업에는 프로세스 사용
import dask.bag as db
bag = db.read_text('logs/*.txt')
result2 = bag.map(python_function).compute(scheduler='processes')

# 디버깅에는 동기식 사용
dask.config.set(scheduler='synchronous')
result3 = problematic_computation.compute()  # pdb 사용 가능

# 모니터링 및 확장에는 분산 사용
from dask.distributed import Client
client = Client()
result4 = computation.compute()  # 대시보드와 함께 분산 사용

핵심 사항:

• Threads: 가장 낮은 오버헤드 (~10 µs/태스크), 숫자 연산에 최적
• Processes: GIL 회피 (~10 ms/태스크), Python 작업에 최적
• Distributed: 모니터링 대시보드 (~1 ms/태스크), 클러스터로 확장
• 계산별 또는 전역적으로 스케줄러 변경 가능

모범 사례

포괄적인 성능 최적화 안내, 메모리 관리 전략 및 피해야 할 일반적인 함정은 references/best-practices.md를 참조하세요. 핵심 원칙은 다음과 같습니다:

더 간단한 솔루션으로 시작

Dask를 사용하기 전에 다음을 고려하세요:

• 더 나은 알고리즘
• 효율적인 파일 형식 (CSV 대신 Parquet)
• 컴파일된 코드 (Numba, Cython)
• 데이터 샘플링

중요한 성능 규칙

1. 데이터를 로컬에 로드한 후 Dask에 전달하지 마세요

# 잘못된 예: 먼저 모든 데이터를 메모리에 로드
import pandas as pd
df = pd.read_csv('large.csv')
ddf = dd.from_pandas(df, npartitions=10)

# 올바른 예: Dask가 로딩을 처리하도록 함
import dask.dataframe as dd
ddf = dd.read_csv('large.csv')

2. compute() 호출을 반복하지 마세요

# 잘못된 예: 각 compute가 개별적
for item in items:
    result = dask_computation(item).compute()

# 올바른 예: 모든 계산을 한 번에
computations = [dask_computation(item) for item in items]
results = dask.compute(*computations)

3. 지나치게 큰 태스크 그래프를 만들지 마세요

• 수백만 개의 태스크가 있으면 청크 크기 증가
• map_partitions/map_blocks를 사용하여 연산 융합
• 태스크 그래프 크기 확인: len(ddf.__dask_graph__())

4. 적절한 청크 크기 선택

• 목표: 청크당 ~100 MB (또는 워커 메모리에서 코어당 10개 청크)
• 너무 큼: 메모리 오버플로우
• 너무 작음: 스케줄링 오버헤드

5. 대시보드 사용

from dask.distributed import Client
client = Client()
print(client.dashboard_link)  # 성능 모니터링, 병목 현상 식별

일반적인 워크플로우 패턴

ETL 파이프라인

import dask.dataframe as dd

# 추출: 데이터 읽기
ddf = dd.read_csv('raw_data/*.csv')

# 변환: 정리 및 처리
ddf = ddf[ddf['status'] == 'valid']
ddf['amount'] = ddf['amount'].astype('float64')
ddf = ddf.dropna(subset=['important_col'])

# 적재: 집계 및 저장
summary = ddf.groupby('category').agg({'amount': ['sum', 'mean']})
summary.to_parquet('output/summary.parquet')

비정형 → 정형 파이프라인

import dask.bag as db
import json

# 비정형 데이터에는 Bag으로 시작
bag = db.read_text('logs/*.json').map(json.loads)
bag = bag.filter(lambda x: x['status'] == 'valid')

# 정형 분석을 위해 DataFrame으로 변환
ddf = bag.to_dataframe()
result = ddf.groupby('category').mean().compute()

대규모 배열 계산

import dask.array as da

# 대규모 배열 로드 또는 생성
x = da.from_zarr('large_dataset.zarr')

# 청크 단위로 처리
normalized = (x - x.mean()) / x.std()

# 결과 저장 (덮어쓰기에는 mode=; 압축에는 zarr_array_kwargs 사용)
da.to_zarr(normalized, 'normalized.zarr', mode='w')

커스텀 병렬 워크플로우

from dask.distributed import Client

client = Client()

# 대규모 데이터셋을 한 번에 scatter
data = client.scatter(large_dataset)

# 종속성과 함께 병렬 처리
futures = []
for param in parameters:
    future = client.submit(process, data, param)
    futures.append(future)

# 결과 수집
results = client.gather(futures)

올바른 컴포넌트 선택

이 의사 결정 가이드를 사용하여 적절한 Dask 컴포넌트를 선택하세요:

데이터 유형:

• 테이블 형식 데이터 → DataFrames
• 숫자 배열 → Arrays
• 텍스트/JSON/로그 → Bags (그런 다음 DataFrame으로 변환)
• 커스텀 Python 객체 → Bags 또는 Futures

연산 유형:

• 표준 pandas 연산 → DataFrames
• 표준 NumPy 연산 → Arrays
• 커스텀 병렬 태스크 → Futures
• 텍스트 처리/ETL → Bags

제어 수준:

• 고수준, 자동 → DataFrames/Arrays
• 저수준, 수동 → Futures

워크플로우 유형:

• 정적 계산 그래프 → DataFrames/Arrays/Bags
• 동적, 진화 → Futures

통합 고려 사항

파일 형식

• 효율적: Parquet, HDF5, Zarr (컬럼형, 압축, 병렬 친화적)
• 호환되지만 느림: CSV (초기 수집에만 사용)
• Arrays용: HDF5, Zarr, NetCDF

컬렉션 간 변환

# Bag → DataFrame
ddf = bag.to_dataframe()

# DataFrame → Array (숫자 데이터용)
arr = ddf.to_dask_array(lengths=True)

# Array → DataFrame
ddf = dd.from_dask_array(arr, columns=['col1', 'col2'])

다른 라이브러리와 함께

• XArray: 레이블이 있는 차원으로 Dask 배열 래핑 (지리공간, 이미징)
• Dask-ML: scikit-learn 호환 API를 사용한 머신러닝
• Distributed: 고급 클러스터 관리 및 모니터링

디버깅 및 개발

반복적 개발 워크플로우

1. 작은 데이터로 동기식 스케줄러 테스트:

dask.config.set(scheduler='synchronous')
result = computation.compute()  # pdb 사용 가능, 쉬운 디버깅

2. 샘플에서 스레드로 검증:

sample = ddf.head(1000)  # 작은 샘플
# 로직 테스트 후 전체 데이터셋으로 확장

3. 분산으로 확장하여 모니터링:

from dask.distributed import Client
client = Client()
print(client.dashboard_link)  # 성능 모니터링
result = computation.compute()

일반적인 문제

메모리 오류:

• 청크 크기 줄이기
• persist()를 전략적으로 사용하고 사용 후 삭제
• 커스텀 함수의 메모리 누수 확인

느린 시작:

• 태스크 그래프가 너무 큼 (청크 크기 증가)
• 태스크를 줄이기 위해 map_partitions 또는 map_blocks 사용

병렬화 불량:

• 청크가 너무 큼 (파티션 수 증가)
• Python 코드와 함께 스레드 사용 (프로세스로 전환)
• 데이터 종속성 방지