Spark DataFrame·PySpark MLlib·Dask로 설계하는 분산 데이터 파이프라인

분산 처리 도구를 고를 때 먼저 볼 기준

Spark DataFrame과 PySpark MLlib은 분산 처리와 학습을 위한 견고한 엔진을 제공한다. Dask는 Python 생태계에 밀착한 병렬화 환경으로, Pandas·NumPy·Scikit-learn 흐름을 이어가기 좋다.

세 도구는 모두 대규모 데이터를 다룰 수 있지만 실행 모델, 머신러닝 파이프라인의 범위, 클러스터 운영 방식이 다르다. 데이터 규모와 지연 요구, 기존 Python 코드와의 호환성, 테이블 포맷을 통한 일관성 요구를 함께 기준으로 삼아야 한다.

Spark와 Dask가 작업을 실행하는 방식

Spark DataFrame은 Spark SQL 기반의 컬럼 지향 데이터 모델이다. Catalyst 옵티마이저와 Tungsten 실행 엔진이 코드 생성과 메모리 최적화를 담당하며, 배치와 Structured Streaming을 하나의 API로 다룬다. HDFS, S3, Kafka, JDBC 등 다양한 소스와 싱크에 연결할 수 있다.

Spark는 DAG 기반 지연 평가와 스테이지 전체 최적화를 통해 셔플과 조인 비용을 낮추는 물리 계획을 만든다. 컬럼형 메모리 레이아웃과 오프히프 관리는 GC 부담을 완화하는 데 쓰인다.

Dask는 Python 네이티브 병렬·분산 컴퓨팅 프레임워크다. DataFrame, Array, Bag, Delayed 같은 고수준 API를 제공하며, 작업 그래프를 바탕으로 동적으로 스케줄링한다. 파티션별 메모리 해제와 스필링이 유연하고 스케일 업·아웃도 가능하지만, Python 객체를 다루는 과정에서는 serialization 오버헤드를 고려해야 한다.

Spark DataFrame은 SQL과 DSL 중심이라 최적화 효과를 얻기 좋다. 다만 UDF를 사용하면 JVM과 Python 경계를 넘는 비용이 생길 수 있다. Dask DataFrame은 Pandas와 유사해 진입 장벽이 낮지만, 일부 Pandas 연산은 지원되지 않거나 부분적으로만 지원된다.

학습 파이프라인과 클러스터 운영의 차이

PySpark MLlib은 Spark 기반 대규모 머신러닝 라이브러리다. Pipeline, Transformer, Estimator 추상화를 제공하고 분산 데이터셋으로 스케일 아웃 학습과 평가를 수행한다. 모델과 메타데이터를 저장해 재현성을 관리할 수 있으며, 그리드서치와 교차검증도 내장한다. 대규모 데이터에는 적합하지만 알고리즘 범위는 Scikit-learn보다 제한적이다.

Dask-ML은 Scikit-learn 호환 API를 바탕으로 병렬 학습과 하이퍼파라미터 탐색을 지원한다. 초대용량 데이터나 고난도 모델을 다룰 때는 별도의 튜닝이 필요하다.

Spark는 YARN, Kubernetes, Standalone 환경을 지원하며 체크포인트, 장애 복구, 쿼리 계획 캐시 등 운영 기능이 성숙해 있다. Dask는 Scheduler와 Workers 구조를 사용하고, 경량 배포와 Jupyter 환경에 잘 맞는다. 장기 배치나 대규모 셔플 워크로드에서는 Dask 쪽의 튜닝 비중이 커진다.

데이터 일관성이 중요하다면 Spark에 Delta, Apache Iceberg, Hudi를 결합해 ACID 트랜잭션, 스키마 진화, 타임트래블을 제공할 수 있다. 이 조합은 옵티미스틱 락 기반의 커밋과 충돌 해결을 사용한다. Dask는 파일 기반 멱등 설계를 권장하며, 메타스토어와 테이블 포맷을 연계하면 ACID를 보완할 수 있다.

항목	Spark DataFrame	PySpark MLlib	Dask 병렬 처리
성능	대규모 배치/스트리밍 고성능, 코드 생성 최적화	분산 학습 스케일 아웃, 데이터 병렬 최적	중소~대규모 파이썬 워크로드 효율, 직렬화 비용 영향
확장성	선형에 가까운 수평 확장, 대규모 셔플 강점	클러스터 크기에 비례 확장	유연한 스케일 업/아웃, 초대용량 셔플 시 튜닝 필요
일관성	Delta/Iceberg/Hudi와 ACID 보장	모델/메타데이터 일관 관리	파일 기반 멱등 처리 중심, ACID는 외부 포맷 연계
안정성	체크포인트/리니지/장애 복구 성숙	파이프라인 기반 재현성 우수	작업 그래프 재시도/스필, 워커 안정성 튜닝 요구
운영 편의	SQL 친화, 성숙한 모니터링/툴링	ML 파이프라인 일원화	Python 생태계 친화, Jupyter/랩탑-클러스터 전환 용이

데이터 규모와 지연 요구에 따른 처리 흐름

⤢✕

처리 경로는 데이터 규모와 지연 요구를 기준으로 Spark 또는 Dask에서 시작한다. 이후 품질검사와 스키마 적용을 거쳐 특징 생성과 학습 여부를 결정하고, Delta 또는 Parquet로 결과를 저장한다.

실패한 태스크나 스테이지는 재시도 경로를 두고, 스필과 백프레셔를 적용한다. 저장 단계에서는 ACID 커밋과 충돌 해결 절차까지 포함해야 한다.

파이프라인에서 자주 쓰이는 구성

데이터 웨어하우스 ETL/ELT에서는 Kafka CDC와 S3 Raw를 원천으로 두고, Spark Structured Streaming으로 브론즈→실버→골드 레이어를 정제할 수 있다. Delta Lake 트랜잭션은 스키마 진화와 타임트래블 운영에 사용된다.

대규모 추천·분류 모델은 PySpark MLlib Pipeline으로 피쳐 엔지니어링, 분산 학습, 교차검증을 한 흐름으로 처리할 수 있다. 모델과 파이프라인의 버전 관리, 셰도우 배치 검증도 이 흐름에 포함된다.

Python 기반 데이터 과학에서는 Dask DataFrame으로 10~200GB CSV/Parquet를 병렬 처리하고, Scikit-learn 또는 Dask-ML로 하이퍼파라미터를 탐색할 수 있다. JupyterLab과 Dask Dashboard는 셔플과 메모리 병목을 실시간으로 관찰하는 데 활용된다.

Spark DataFrame으로 ETL 구성하기

전제 조건은 다음과 같다.

Python 3.9~3.11, Java 8+/11
Apache Spark 3.4+ (버전별 기본 설정 상이, 최신 정보 확인 필요)
PySpark, delta-spark(선택), Dask 2024.x, dask[distributed], dask-ml(선택)
로컬 실행: 단일 노드 테스트 후 YARN/Kubernetes/Standalone 확장

# pip install pyspark delta-spark
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, to_timestamp

spark = (
    SparkSession.builder
    .appName("etl-basic")
    .config("spark.sql.shuffle.partitions", "200")
    .getOrCreate()
)

# 입력: Parquet
df = spark.read.parquet("s3://bucket/raw/events/")

# 정제/스키마 적용
clean = (
    df.withColumn("ts", to_timestamp(col("event_time")))
      .filter(col("ts").isNotNull())
      .select("user_id", "event_type", "ts")
)

# 파티셔닝 저장(Parquet/Delta)
(
    clean.repartition(200, col("event_type"))
         .write.mode("overwrite")
         .partitionBy("event_type")
         .parquet("s3://bucket/silver/events/")
)

spark.stop()

event_type을 기준으로 재파티셔닝하면 다운스트림 스캔 비용을 줄일 수 있다. 셔플 파티션 수는 클러스터 코어 규모에 맞춰 조정한다.

PySpark MLlib에서 변환과 학습을 묶는 방법

# pip install pyspark
from pyspark.sql import SparkSession
from pyspark.ml import Pipeline
from pyspark.ml.feature import StringIndexer, OneHotEncoder, VectorAssembler
from pyspark.ml.classification import LogisticRegression
from pyspark.ml.evaluation import BinaryClassificationEvaluator

spark = SparkSession.builder.appName("mllib-pipeline").getOrCreate()

train = spark.read.parquet("s3://bucket/silver/events/")

si = StringIndexer(inputCol="event_type", outputCol="event_type_idx", handleInvalid="keep")
ohe = OneHotEncoder(inputCols=["event_type_idx"], outputCols=["event_type_ohe"])
va = VectorAssembler(
    inputCols=["event_type_ohe"],  # 예시: 실제로는 수치 피쳐 포함 권장
    outputCol="features"
)
lr = LogisticRegression(featuresCol="features", labelCol="label", maxIter=50)

pipeline = Pipeline(stages=[si, ohe, va, lr])
model = pipeline.fit(train)

pred = model.transform(train)
auc = BinaryClassificationEvaluator(labelCol="label").evaluate(pred)
print("AUC =", auc)

model.write().overwrite().save("s3://bucket/models/lr_pipeline/")
spark.stop()

Pipeline은 피쳐 변환과 학습을 동일한 흐름으로 유지하므로 학습과 서빙에서 같은 변환을 재사용할 수 있다. 그리드서치와 교차검증은 TrainValidationSplit 또는 CrossValidator로 확장한다.

Dask DataFrame으로 병렬 저장하기

# pip install "dask[distributed]" fsspec s3fs dask-ml
import dask.dataframe as dd
from dask.distributed import Client

client = Client()  # 로컬 스레드/프로세스 기반 실행

ddf = dd.read_csv("s3://bucket/raw/*.csv", blocksize="256MB", assume_missing=True)

# 간단 전처리
ddf = ddf.dropna(subset=["event_time"]).astype({"user_id": "int64"})

# 파티션 수 튜닝(코어 수의 2~4배 권장)
ddf = ddf.repartition(npartitions=64)

# 병렬 저장
ddf.to_parquet("s3://bucket/silver/events_dask/", write_index=False)

blocksize는 입력 병렬성을 제어한다. 메모리를 고려해 파티션 크기는 128MB~512MB 범위로 잡는 방식을 권장한다. Dask는 지연 평가를 사용하므로 저장 시점에 compute가 트리거된다.

운영 설계에서 놓치기 쉬운 지점

Delta, Iceberg, Hudi 같은 테이블 포맷은 ACID, 스키마 진화, 타임트래블을 확보하는 기반이다. merge·upsert, VACUUM, RETENTION 설정은 운영 표준으로 관리할 필요가 있다.

셔플과 조인은 파티션 설계에 크게 좌우된다. 소스 특성에 따라 디멘전과 시간을 분할하고, 데이터 스큐가 있으면 솔팅 또는 솔트 키를 검토한다. 브로드캐스트 조인도 활용할 수 있으나 임계값은 버전별로 다르므로 최신 정보를 확인해야 한다.

Spark에서는 동적 할당, 스필 튜닝, 적절한 캐시 전략인 persist level을 함께 조정한다. Dask는 워커 메모리 한도와 spill-to-disk, 코어×2~4 수준의 파티션 수, cloudpickle과 msgpack 같은 직렬화 프로토콜 선택을 검토한다.

신뢰성 측면에서 Spark Structured Streaming과 Delta sink 조합은 체크포인트 및 정확히 한 번 처리 보장에 활용된다. Dask는 출력 경로를 분리한 뒤 원자적으로 스왑하는 멱등 작업 설계와 재시도 정책이 필요하다.

전송과 저장 구간은 TLS/KMS로 암호화하고 IAM/Kerberos, Vault/Secret Manager 기반 시크릿 관리를 연계한다. 스키마·제약·기대치 기반 데이터 품질 규칙을 자동화하고 데이터 카탈로그와 연결하는 방식도 함께 고려한다.

Spark UI, History Server, Ganglia/Prometheus, Structured Streaming 지표 경보는 Spark 운영 상태를 확인하는 수단이다. Dask Dashboard에서는 태스크 대기와 셔플 병목을 보고 백프레셔를 조정한다.

성능 개선과 운영 표준화에서 얻는 효과

파티션과 조인을 최적화하고 컬럼형 포맷을 적용하면 ETL 처리 시간은 30~~70% 단축될 수 있다. 동적 할당, 스필·캐시 튜닝, 적절한 파이프라인 분리는 인프라 비용 20~~40% 절감으로 이어질 수 있다. 분산 학습과 파이프라인 캐시 재사용은 모델 학습 시간을 2~10배 단축하는 효과가 있다.

ACID 테이블 포맷과 자동 검증은 데이터 품질 및 일관성을 높인다. 개발·운영 표준화는 변경 용이성과 재현성을 강화하고, Python과 Spark 생태계를 함께 쓰는 구조는 팀 생산성 향상에 도움이 된다.

대규모 ETL, 스트리밍, 머신러닝 파이프라인에는 Spark DataFrame과 PySpark MLlib의 안정성과 확장성이 맞는다. Python 중심 데이터 과학과 중간 규모 워크로드에는 Dask가 높은 개발 생산성을 제공한다. Delta 또는 Iceberg/Hudi 기반 ACID 테이블, 멱등 출력 설계, 파티션·셔플·메모리 최적화 절차를 조합해 선택 기준과 복구 가능성을 함께 마련해야 한다.