Spark

MLlib 개요와 K-Means

MLlib은 전처리, 데이터 가공(munging), 모델 학습, 예측까지 대규모 데이터에서 수행할 수 있게 해주는 Spark 내장 머신러닝 라이브러리다. 분류, 회귀, 군집화 등 다양한 작업을 지원하며, MLlib으로 학습한 모델은 Structured Streaming 파이프라인 안에서도 예측에 사용할 수 있다.

예제로 쓰이는 K-Means는 군집화 알고리즘이다. K개의 중심점을 데이터 안에 무작위로 배치한 뒤, 각 점을 가장 가까운 중심점의 클래스에 배정하고, 배정된 점들의 평균 위치로 중심(centroid)을 다시 계산한다. 이 과정을 중심점이 더 이상 바뀌지 않을 때까지, 혹은 정해진 반복 횟수만큼 반복한다.

숫자가 아닌 데이터를 벡터로: StringIndexer → OneHotEncoder → VectorAssembler

MLlib 알고리즘은 대부분 입력이 숫자형 Vector일 것을 요구한다. 그런데 소매 데이터는 타임스탬프, 정수, 문자열이 뒤섞여 있으므로 수치형으로 변환하는 전처리가 필요하다. 예제는 na.fill(0)으로 결측치를 채우고, InvoiceDate에서 date_format으로 day_of_week(요일) 컬럼을 새로 만든 뒤, coalesce(5)로 파티션 수를 줄인다. 이후 2011-07-01을 기준으로 학습/테스트 세트를 수동으로 나누는데, MLlib이 제공하는 train validation split이나 cross validation 같은 유틸리티를 쓸 수도 있지만 이 예제에서는 날짜 기준 수동 분할을 택했다.

StringIndexer는 day_of_week 같은 문자열 카테고리를 정수 인덱스로 바꿔준다(예: Saturday는 6, Monday는 1). 문제는 이 숫자 값 자체가 크기 비교의 의미를 갖게 되어, 마치 Saturday가 Monday보다 '더 크다'는 잘못된 서열 관계를 암묵적으로 만들어낸다는 점이다. 이를 바로잡기 위해 OneHotEncoder를 사용해 각 요일을 독립된 boolean 컬럼으로 펼친다. 마지막으로 VectorAssembler는 UnitPrice, Quantity, day_of_week_encoded 같은 여러 피처 컬럼을 하나의 features Vector 컬럼으로 결합한다. Spark의 모든 머신러닝 알고리즘은 입력으로 이런 Vector 타입을 요구하기 때문이다.

val indexer = new StringIndexer()
  .setInputCol("day_of_week")
  .setOutputCol("day_of_week_index")

val encoder = new OneHotEncoder()
  .setInputCol("day_of_week_index")
  .setOutputCol("day_of_week_encoded")

val vectorAssembler = new VectorAssembler()
  .setInputCols(Array("UnitPrice", "Quantity", "day_of_week_encoded"))
  .setOutputCol("features")

Pipeline, fit/transform, 그리고 학습 데이터 캐싱

Pipeline은 indexer, encoder, vectorAssembler 같은 여러 변환 단계를 하나로 묶어, 앞으로 들어올 어떤 데이터든 정확히 같은 순서로 처리되도록 만든다. 학습 준비는 두 단계로 이뤄진다. 먼저 fit()으로 파이프라인을 학습 데이터에 맞춘다 — 예를 들어 StringIndexer는 인덱싱할 고유값이 몇 개인지 먼저 알아야 한다. 이렇게 만들어진 fittedPipeline을 transform()에 적용하면 어떤 데이터든 일관되고 반복 가능한 방식으로 변환할 수 있다.

이 예제에서는 모델 학습 자체를 파이프라인에 포함시키지 않고 별도로 분리했는데, 이는 캐싱의 효용을 보여주기 위해서다. 이후 하이퍼파라미터 튜닝을 하며 같은 변환 로직을 반복해서 태우고 싶지 않으므로, transformedTraining.cache()로 변환이 끝난 학습 데이터를 메모리에 저장해 둔다. 이렇게 하면 튜닝 과정에서 매번 파이프라인을 다시 실행하지 않고 이미 변환된 상태의 데이터에 효율적으로, 반복적으로 접근할 수 있다.

학습 자체는 KMeans(setK(20), setSeed(1L))를 정의한 뒤 fit(transformedTraining)을 호출해 학습된 KMeansModel을 얻는 과정이다. MLlib의 모든 알고리즘은 이렇게 학습 전 상태(추정기, 예: KMeans)와 학습 후 상태(변환기, 예: KMeansModel) 두 가지 타입을 가지며, 둘 다 앞서 본 StringIndexer 같은 전처리 변환기와 비슷한 인터페이스를 공유한다. computeCost로 학습/테스트 세트 각각의 비용을 계산해볼 수 있는데, 이 예제에서는 피처를 스케일링하지 않았기 때문에 비용이 상당히 높게 나온다.