📒 Spark(10) - ML(1)

📌 Spark ML

Spark ML 소개

• 머신러닝 관련 다양한 알고리즘, 유틸리티로 구성된 라이브러리

  • Classification, Regression, Clustering, Collaborative Filtering, Dimensionality Reduction - 참고
  • 아직 딥러닝의 대한 지원은 미약하다.

• RDD 기반과 DataFrame 기반의 두 버진이 존재한다.

  • spark.mllib vs. spark.ml
    • spark.mllib -> RDD 기반
    • spark.ml -> DataFrame 기반
  • spark.mllib는 RDD위에서 동작하는 이전 라이브러리로 더 이상 업데이트가 안된다.
  • spark.ml을 사용하는 것이 좋다
    -> import pyspark.ml

Spark ML 장점

• DataFrame과 SparkSQL등을 이용하여 전처리를 진행한다.
• Spakr MLlib를 사용하여 모델을 빌드한다.
• ML Pipeline을 통해 모델 빌딩을 자동화한다.
• MLflow로 모델을 관리하고 서빙한다.
• 대용량 데이터도 처리 가능하다.

MLflow

• 모델의 관리와 서빙을 위한 Ops 관련 기능도 제공한다.
• MLflow
  • 모델 개발, 테스트, 관리 그리고 서빙까지 제공해주는 End-to-End Framwork이다.
  • MLflow는 Python, Java, R, API를 지원한다.
  • MLflow는 Tracking, Models, Projects를 지원한다.

Spark ML에서 제공하는 알고리즘

• Classification(분류)
  -> Logistic regression, Decision tree, Random forest, Gradient-boosted tree, ...
  -> 레이블(Lable)을 사용하여 데이터셋을 가져와 해당 정보를 기반으로 새 데이터에 레이블을 지정하는 방법을 학습하는 감독 학습 기법이다. Yes & No 와 같이 분류에 대한 문제를 다룬다.

• Collaborative Filtering(협업 필터링)

  • 권장 사항을 만드는 기술이다. 일반적으로 쇼핑 웹 사이트에 "좋아할 수 도 있는.." 등등의 형태로 표기되는것이 이것이다.
  • 다수의 데이터 관측을 처리하여 유사한 특성이나 특징을 가진 엔티티를 찾은다음 이전 관측에 따라 새로 관찰된 엔티티에 권장 사항 또는 제안을 하는 것이다.
  • 분류와 달리 비감독 학습 기법이다. 이는 레이블(Lable)없이 데이터의 패턴을 도출할 수 있음을 의미한다.

• Clustering(클러스터링)
  -> K-means, LDA, GMM...

  • 데이터 관측 수집 내에서 구조를 발견하는 프로세스이다.
  • 형식 & 구조가 명확하지 않은 경우 더욱 유용하다.
  • 제공된 데이터에서 자연스럽게 발생하는 그룹을 발견한다.

📌 모델 빌딩의 기본 구조

• 여느 라이브러리를 사용한 모델 빌빙과 크게 다르지 않다.

  • Train dataset 전처리
  • Model 빌드
  • Model 검증(confusion matrix)

• Scikit-Learn(사이킷 런)과 비교했을 때 장점

  • 데이터 크기

    • Scikit-Learn은 하나의 컴퓨터에서 돌아가는 모델을 빌딩한다.
    • Spark MLlib는 여러 서버 위에서 모델을 빌딩한다.

  • Training Set의 크기가 크면 전처리와 모델 빌딩에 있어서 Spark이 큰 장점을 갖는다.

  • Spark은 ML PipeLine을 통해 모델 개발의 반복을 쉽게 해준다.

📌 Spark ML 피쳐 변환

Feature Extractor & Transformer

• Feature값들을 모델 훈련에 적합한 형태로 바꾸는 것을 지칭한다.

• Feature Transformer가 하는 일

  • 먼저 Feature값들은 숫자 필드여야 한다.
    -> 텍스트 필드(카테고리 값들)를 숫자 필드로 변환해야한다.
  • 숫자 필드 값의 범위를 표준화한다.
    • 숫자필드라고 해도 가능한 값의 범위를 특정 범위(0~1)로 변환해야한다.
    • 이를 Feature Scaling & Normalization이라고 한다.
  • 비어있는 필드들의 값은 어떻게 채울 것인가?
    -> 여러 방식으로 채울 수 있다. 평균값, 최대값, 최솟값 등등

• Feature Extractor가 하는 일

  • 기존 Feature에서 새로운 Feature를 추출한다.
  • TF-IDF, Word2Vec, ...
    -> 많은 경우 텍스트 데이터를 어떤 형태로 인코딩하는 것이 여기에 해당한다.

Feature Transformer - StringIndexer

• 텍스트 카테고리를 숫자로 변환
• 아래 왼쪽과 같은 값을 갖는 Color라는 이름의 Feature가 존재한다면 이를 숫자로 변환해주는 것이 Feature Trasformer의 목적이다.

변환 전	변환 후
Red	1
Blue	2
Orange	3
White	4
Black	5
Gray	6
Yellow	7

• Scikit-Learn은 sklearn.preprocessing 모듈 아래 여러 인코더가 존재한다.
  -> OneHotEncoder, Labelencoder, OrdianlEncoder, ...

• Spark MLlib의 경우 pyspark.ml.feature 모듈 아래 두 개의 인코더가 존재한다.

  • StringIndexer, OneHotEncoder
  • 사용법은 Indexer 모델을 만들고 (Fit), Indexer 모델로 데이터프레임을 변환(Transform)

from pyspark.ml.feature import StringIndexer

gender_indexer = StringIndexer(inputCol='Gender', outputCol='GenderIndexed')
gender_indexer_model = gender_indexer.fit(final_data)

final_data_with_transformed_gender_gender = gender_indexer_model.transform(final_data)

Feature Transformer - Scaler

• 숫자 필드값의 범위를 표준화
• 숫자 필드 값의 범위를 특정 범위(0~1)로 변환 하는 것이다.
• Feature Scaling & Normalization이라고 부른다.

변환 전	변환 후
-20	0
100	1
40	0.5
25	0.375
15	0.125

• Scikit-Learn은 sklearn.preprocessing 모듈 아래 두 개의 스케일러가 존재한다.
  -> StandardScaler, MinMaxScaler

• Spark MLlib의 경우 pyspark.ml.feature 모듈 아래 두 개의 스케일러가 존재한다.

  • StandardScaler, MinMaxScaler
  • 사용법은 Scaler 모델을 만든 후 (fit), Scaler 모델로 DataFrame을 변환(Transform)

• StandardScler
  -> 각 값에서 평균을 빼고 이를 표준편차로 나눈다. 값의 분포가 정규분포를 따르는 경우에 사용한다.

• MinMaxScaler
  -> 모든 값을 0과 1사이로 스케일한다. 각 값에서 최소값을 빼고(최대값-최소값)으로 나눈다.

Feature Transformer - Imputer

• 값이 없는 필드 채우기
• 값이 존재하지 않는 레코드들이 존재하는 필드들의 경우 기본값을 정해서 채우는 것이다. -> Impute한다. 라고 부른다.

변환 전	변환 후
10	10
	25
20	20
30	30
40	40

• Scikit-Learn은 sklearn.preprocessing 모듈 아래 존재한다. -> Imputer

• Spark MLlib의 경우 pyspark.ml.feature 모듈 아래에 존재한다.

  • Imputer
  • 사용법은 Imputer 모델을 만든 후(fit), Imputer 모델로 DataFrame을 변환(Transform)

from pyspark.ml.feature import Imputer

imputer = Imputer(strategy='mean', imputCols=['Age'], outputCols=['Agelmputed'])
imputer_model = imputer.fit(final_data)
final_data_age_transformed = imputer_model.transform(final_data)