[SparkML] MLlib 개념 및 실습 코드

 

MLlib

Machine Learning Library

스파크의 컴포넌트 중 하나로, 머신러닝 파이프라인 개발을 쉽게 하기 위해 개발되었다.

MLlib에는 다시 알고리즘(Classification 등) + 파이프라인(Training, Evaluation 등) + FE + Utils(Statistics 등)의 여러 컴포넌트가 있다. MLlib은 DataFrame 위에서 동작하며, MLlib API를 Spark ML이라고 부른다.


이번 포스팅에선 MLlib의 활용을 알고리즘, 파이프라인, FE, Utils에 걸쳐서 차근차근 살펴보도록 하겠다.

MLlib 포스팅에서 사용할 데이터는 이전 포스팅에서 다뤘던 '택시 데이터'이다.

 

https://mengu.tistory.com/50?category=932924

[SparkSQL] 택시 데이터 다운/전처리/분석 feat. TLC

 

위의 데이터를 받아놓지 않아도 충분히 MLlib은 견학할 수 있으니 무리하지 않아도 된다.

가보자.

 

 

"본 포스팅은 패스트캠퍼스의 강의를 듣고, 정리한 자료임을 밝힙니다."

 

 

 

 

Basic Settings

기본 세팅이다. 임포트 해줘야 할 것들은 미리 해놨고, SparkSession을 열어두었다.

 

# 폰트 설정
from matplotlib import font_manager, rc
font_path = 'C:\\WINDOWS\\Fonts\\HBATANG.TTF'
font = font_manager.FontProperties(fname=font_path).get_name()
rc('font', family=font)

# basic settings
import os
import findspark
findspark.init(os.environ.get("SPARK_HOME"))
import pyspark
from pyspark import SparkConf, SparkContext
import pandas as pd
import faulthandler
faulthandler.enable()
from pyspark.sql import SparkSession

MAX_MEMORY = "5g"
spark = SparkSession.builder.master('local').appName("taxi-fare-prediction")\
.config("spark.executor.memory", MAX_MEMORY)\
.config("spark.driver.memory", MAX_MEMORY).getOrCreate()

# 데이터가 있는 파일
zone_data = "C:/DE study/data-engineering/01-spark/data/taxi_zone_lookup.csv"
trip_files = "C:/DE study/data-engineering/01-spark/data/trips/*"

# 데이터 로드
trips_df = spark.read.csv(f"file:///{trip_files}", inferSchema = True, header = True)
zone_df = spark.read.csv(f"file:///{zone_data}", inferSchema = True, header = True)

# 데이터 스키마
trips_df.printSchema()
zone_df.printSchema()

# 데이터 createOrReplaceTempView()
trips_df.createOrReplaceTempView("trips")
zone_df.createOrReplaceTempView("zone")


root
 |-- VendorID: integer (nullable = true)
 |-- tpep_pickup_datetime: string (nullable = true)
 |-- tpep_dropoff_datetime: string (nullable = true)
 |-- passenger_count: integer (nullable = true)
 |-- trip_distance: double (nullable = true)
 |-- RatecodeID: integer (nullable = true)
 |-- store_and_fwd_flag: string (nullable = true)
 |-- PULocationID: integer (nullable = true)
 |-- DOLocationID: integer (nullable = true)
 |-- payment_type: integer (nullable = true)
 |-- fare_amount: double (nullable = true)
 |-- extra: double (nullable = true)
 |-- mta_tax: double (nullable = true)
 |-- tip_amount: double (nullable = true)
 |-- tolls_amount: double (nullable = true)
 |-- improvement_surcharge: double (nullable = true)
 |-- total_amount: double (nullable = true)
 |-- congestion_surcharge: double (nullable = true)

root
 |-- LocationID: integer (nullable = true)
 |-- Borough: string (nullable = true)
 |-- Zone: string (nullable = true)
 |-- service_zone: string (nullable = true)

 

 

 

 

 

DataFrame

---

데이터를 일단 DataFrame 형식으로 가져와야 한다. 또한 원하는 Input과 Output을 남기고는 제거해둬야 한다.

 

 

query = '''
SELECT
    trip_distance,
    total_amount
FROM
    trips
WHERE
    total_amount < 5000
    AND total_amount > 0
    AND trip_distance > 0
    AND trip_distance < 500
    AND passenger_count < 4
    AND TO_DATE(tpep_pickup_datetime) >= '2021-01-01'
    AND TO_DATE(tpep_pickup_datetime) < '2021-08-01'
'''

# 원하는 data만 남겨두기
data_df = spark.sql(query)
data_df.createOrReplaceTempView('data')


data_df.show


+-------------+------------+
|trip_distance|total_amount|
+-------------+------------+
|         16.5|       70.07|
|         1.13|       11.16|
|         2.68|       18.59|
|         12.4|        43.8|
|          9.7|        32.3|
|          9.3|       43.67|
|         9.58|        46.1|
|         16.2|        45.3|
|         3.58|        19.3|
|         0.91|        14.8|
|         2.57|        12.8|
|          0.4|         5.3|
|         3.26|        17.3|
|        13.41|       47.25|
|         18.3|       61.42|
|         1.53|       14.16|
|          2.0|        11.8|
|         16.6|       54.96|
|         15.5|       56.25|
|          1.3|        16.8|
+-------------+------------+
only showing top 20 rows

 

 

 

훈련 데이터셋과 테스트 데이터셋을 미리 나눈다.

 

# 8:2 비율로 데이터를 나누기
train_df, test_df = data_df.randomSplit([0.8, 0.2], seed=1)


print(f'---> {train_df.count()}')
print(f'---> {test_df.count()}')


---> 10500253
---> 2625787

 

 

 

 

 

Transformer

---

데이터를 학습이 가능한 포멧으로 바꾸는 역할을 한다. DataFrame을 받아 새로운 DataFrame을 만드는데, 보통 하나 이상의 column을 더하게 된다.

- Data Normalization

- Tokenization

- one-hot encoding ... etc

 

 

 

모델에 적용하기 위해 VectorAssembler를 사용한다. 모델에 들어갈 features들을 Vector로 압축해서 쉽게 러닝할 수 있도록 한다. vtrain_df를 보면, 새로운 칼럼이 등장했고, features는 trip_distance를 벡터화한 것임을 확인할 수 있다.

 

from pyspark.ml.feature import VectorAssembler
vassembler = VectorAssembler(inputCols=["trip_distance"], outputCol="features")
vtrain_df = vassembler.transform(train_df)
vtrain_df.show()


+-------------+------------+--------+
|trip_distance|total_amount|features|
+-------------+------------+--------+
|         0.01|        3.05|  [0.01]|
|         0.01|         3.3|  [0.01]|
|         0.01|         3.3|  [0.01]|
|         0.01|         3.3|  [0.01]|
|         0.01|         3.3|  [0.01]|
|         0.01|         3.3|  [0.01]|
|         0.01|         3.3|  [0.01]|
|         0.01|         3.3|  [0.01]|
|         0.01|         3.3|  [0.01]|
|         0.01|         3.3|  [0.01]|
|         0.01|         3.3|  [0.01]|
|         0.01|         3.3|  [0.01]|
|         0.01|         3.3|  [0.01]|
|         0.01|         3.3|  [0.01]|
|         0.01|         3.3|  [0.01]|
|         0.01|         3.3|  [0.01]|
|         0.01|         3.3|  [0.01]|
|         0.01|         3.3|  [0.01]|
|         0.01|         3.3|  [0.01]|
|         0.01|         3.3|  [0.01]|
+-------------+------------+--------+
only showing top 20 rows

 

 

 

 

 

Estimator

---

모델의 학습과정을 추상화한 컴포넌트다. fit() 함수를 가지고 있으며, 회귀/분류/군집 등 다양한 머신러닝 알고리즘을 사용할 수 있도록 한다. fit()은 DataFrame을 받아 Model을 반환하고, Model은 transform() 함수를 통해 예측한다.

 

# 컴포넌트 불러오기
from pyspark.ml.regression import LinearRegression
lr = LinearRegression(
    maxIter=50,
    labelCol = 'total_amount',
    featuresCol = 'features')
    
    
# fit() / 학습
model = lr.fit(vtrain_df)

 

 

 

테스트 데이터셋을 벡터화하고, transform() 함수를 이용하여 추론하기

 

# 테스트 데이터셋 벡터화 시키기
vectorassembler = VectorAssembler(inputCols=['trip_distance'], outputCol='features')
vtest_df = vectorassembler.transform(test_df)


# 추론
prediction = model.transform(vtest_df)
prediction.show()


+-------------+------------+--------+-----------------+
|trip_distance|total_amount|features|       prediction|
+-------------+------------+--------+-----------------+
|         0.01|         3.3|  [0.01]|9.430440745312902|
|         0.01|         3.3|  [0.01]|9.430440745312902|
|         0.01|         3.3|  [0.01]|9.430440745312902|
|         0.01|         3.3|  [0.01]|9.430440745312902|
|         0.01|         3.3|  [0.01]|9.430440745312902|
|         0.01|         3.3|  [0.01]|9.430440745312902|
|         0.01|         3.3|  [0.01]|9.430440745312902|
|         0.01|         3.3|  [0.01]|9.430440745312902|
|         0.01|         3.3|  [0.01]|9.430440745312902|
|         0.01|         3.3|  [0.01]|9.430440745312902|
|         0.01|         3.3|  [0.01]|9.430440745312902|
|         0.01|         3.3|  [0.01]|9.430440745312902|
|         0.01|         3.3|  [0.01]|9.430440745312902|
|         0.01|         3.3|  [0.01]|9.430440745312902|
|         0.01|         3.8|  [0.01]|9.430440745312902|
|         0.01|         3.8|  [0.01]|9.430440745312902|
|         0.01|         3.8|  [0.01]|9.430440745312902|
|         0.01|         3.8|  [0.01]|9.430440745312902|
|         0.01|         3.8|  [0.01]|9.430440745312902|
|         0.01|         3.8|  [0.01]|9.430440745312902|
+-------------+------------+--------+-----------------+
only showing top 20 rows

 

 

 

Model 안에 내장되어 있는 함수들을 이용해서, 모델 성능을 간단하게 확인할 수 있다.

 

# 모델의 성능을 평가하자.
print(model.summary.r2)
print(model.summary.rootMeanSquaredError)


0.7648633777017714
6.30781413196623

 

 

 

머신러닝을 다뤄본 사람이라면, Tensorflow나 Spark나 활용 방식은 아주 비슷하다고 느낄 것이다.

다만, 일반 Pandas를 쓰는 것과 달리 Spark를 썼을 때의 분산적/병렬적 접근이 더 효용성이 높기에 '다름'을 인지하길 바란다.

그럼 다음 포스팅에선 Transform을 좀 더 심화하고, Pipeline을 자세히 다뤄보도록 하겠다.