Spark MLlib으로 시작하는 머신러닝: 실전 가이드

MLlib은 Apache Spark의 머신러닝 컴포넌트로, 분산 환경에서의 확장 가능한 ML 솔루션을 제공합니다.

• 주요 특징:
• 분산 처리 기반 머신러닝 - 대규모 데이터셋 처리
• DataFrame 기반 고수준 API - 직관적인 인터페이스
• 종합적인 ML 파이프라인 - 전처리부터 배포까지
• 자동화된 모델 튜닝 - 하이퍼파라미터 최적화
• 고성능 네이티브 구현 - BLAS/LAPACK 활용

• 지원하는 알고리즘:
• 1. 분류 알고리즘
• 로지스틱 회귀 - 이진/다중 분류
• 랜덤 포레스트 - 앙상블 기법
• 결정 트리 - 해석 가능한 모델
• GBT 분류기 - 그래디언트 부스팅

• 2. 회귀 알고리즘
• 선형 회귀 - 기본 회귀 분석
• GBT 회귀 - 비선형 패턴
• 생존 회귀 - 생존 분석
• AFT 생존 회귀 - 가속 실패 시간

• 3. 클러스터링
• K-means - 벡터 양자화
• GMM - 확률적 클러스터링
• 계층적 클러스터링 - 트리 기반
• Power Iteration - 스펙트럴 군집화

• 4. 차원 축소
• PCA - 주성분 분석
• SVD - 특이값 분해
• 요인 분석 - 잠재 변수 모델

MLlib을 활용한 효율적인 머신러닝 파이프라인 구축 방법을 알아봅니다.

1. 개발 환경 설정

from pyspark.ml import Pipeline
from pyspark.ml.feature import *
from pyspark.ml.classification import *
from pyspark.ml.regression import *
from pyspark.ml.evaluation import *

# 최적화된 SparkSession 설정 spark = SparkSession.builder \ .appName("MLPipeline") \ .config("spark.executor.memory", "4g") \ .config("spark.executor.cores", "4") \ .config("spark.driver.memory", "2g") \ .config("spark.sql.shuffle.partitions", "100") \ .config("spark.default.parallelism", "100") \ .getOrCreate() ```

2. 데이터 로드 및 전처리

data = spark.read \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .option("mode", "DROPMALFORMED") \
    .option("cacheSize", "1024") \

# 데이터 캐싱 및 파티셔닝 data = data.repartition(100).cache()

# 학습/검증/테스트 분할 train, val, test = data.randomSplit([0.7, 0.15, 0.15], seed=42) ```

3. 특징 엔지니어링 파이프라인

feature_pipeline = Pipeline(stages=[
    # 범주형 변수 처리
    StringIndexer(
        inputCols=categorical_cols,
        outputCols=[f"{col}_idx" for col in categorical_cols],
        handleInvalid="keep"
    ),
    
    # 원-핫 인코딩
    OneHotEncoder(
        inputCols=[f"{col}_idx" for col in categorical_cols],
        outputCols=[f"{col}_vec" for col in categorical_cols]
    ),
    
    # 수치형 변수 스케일링
    StandardScaler(
        inputCol="numeric_features",
        outputCol="scaled_features",
        withStd=True,
        withMean=True
    ),
    
    # 특징 선택
    ChiSqSelector(
        numTopFeatures=50,
        featuresCol="features",
        outputCol="selected_features",
        labelCol="label"
    )
])

모델의 성능을 극대화하기 위한 다양한 학습 및 최적화 기법을 살펴봅니다.

1. 복잡한 모델 아키텍처

from pyspark.ml.classification import GBTClassifier, RandomForestClassifier

# 기본 모델 정의 gbt = GBTClassifier( featuresCol="selected_features", labelCol="label", maxIter=100, maxDepth=8, stepSize=0.1, maxBins=64, minInstancesPerNode=10, minInfoGain=0.01 )

rf = RandomForestClassifier( featuresCol="selected_features", labelCol="label", numTrees=200, maxDepth=10, maxBins=64, minInstancesPerNode=10, minInfoGain=0.01 )

# 메타 특징 생성기 def create_meta_features(predictions_df): return predictions_df.select( "label", "probability_gbt", "probability_rf", "raw_prediction_gbt", "raw_prediction_rf" ) ```

2. 하이퍼파라미터 최적화

param_grid = ParamGridBuilder() \
    .addGrid(gbt.maxDepth, [6, 8, 10]) \
    .addGrid(gbt.maxIter, [100, 200]) \
    .addGrid(gbt.stepSize, [0.05, 0.1]) \
    .addGrid(rf.numTrees, [150, 200]) \
    .addGrid(rf.maxDepth, [8, 10]) \

# 교차 검증 설정 crossval = CrossValidator( estimator=pipeline, estimatorParamMaps=param_grid, evaluator=MulticlassClassificationEvaluator(), numFolds=5, parallelism=4 ) ```

3. 모델 평가 및 해석

from pyspark.sql.functions import udf

# 다양한 메트릭 평가 evaluators = { "areaUnderROC": BinaryClassificationEvaluator( metricName="areaUnderROC" ), "areaUnderPR": BinaryClassificationEvaluator( metricName="areaUnderPR" ) }

# 특징 중요도 계산 def calculate_feature_importance(model, feature_names): importances = model.featureImportances.toArray() return sorted( zip(feature_names, importances), key=lambda x: x[1], reverse=True ) ```