[데이터 엔지니어링 부트캠프]10월 2주차 회고

좋았던 점

• 기획서 제출이 끝났습니다. 이번에 제출한 기획서 주제는 총 세 가지이고, 다음 주 중에 정규 수업 커리큘럼이 끝나면 바로 멘토님들과 매칭이 될 것 같습니다. 물론 우리 주제가 컨펌이 될 지는 미지수... ???: 이런 것도 기획서라고 써 왔어?!
• 파이널 프로젝트 관련 인원 변동이 있었습니다. 다행히도 타 팀에서 저희 팀으로 이적 옮겨오시는 분이 해보고 싶은 부분이 저희가 중점적으로 두는 데이터 분석과 관련된 파트라고 하니 그 부분으로 잘 조율해서 일을 맡겨보려고 합니다.

아쉬웠던 점

• 제주 버스 데이터를 이용한 미니 프로젝트 기간이 생겼습니다. 1주일을 주시기는 했지만, 사실 엄밀히 말해 프로젝트라기에도 뭣한 시각화 실습에 가까워서 주제 선정에 공을 들이지 않았다는 느낌만 강하게 받았습니다. 무엇보다 그 동안에도 프로젝트와 수업을 병행하겠다고 해서 부담이 상당합니다.
• 빅데이터 파트로 넘어오면서 간당간당하게 버텨오던 컴퓨터의 사양은 기어코 오늘 발목을 잡고 말았습니다. 다음 주 월요일 실습 건까지 생각하면 오늘 실습이 끝났을 떄 약 20GB 정도의 여유 공간이 남아 있어야 했는데, 실습 시작 전에 남아 있던 여유 공간은 무려 200MB(메가바이트). 오타 아닙니다 다행히도 도커의 특성을 이용해 이 여유 공간을 추가로 확보할 수 있는 방법을 찾았습니다. 하지만 여전히 16GB 램이 프로그램 실행 중에 발목을 잡는 경우가 생깁니다. 작업 관리자 들어가보면 CPU, RAM, 디스크가 전부 100%를 찍고 있으니...

배운 점

스파크(Spark)

인메모리 기반의 대용량 데이터 고속 처리 엔진입니다. 자바, 파이썬, scala, R을 지원한다고 하는데 우리는 주력이 파이썬이니만큼 그 쪽을 위주로 설명과 실습을 진행했습니다. 스파크는 자체적으로 머신러닝 알고리즘과 하이브(hive)를 내장하고 있기 때문에 각각 사이킷런과 판다스와 많이 비교되곤 합니다.

• SparkML과 사이킷런(Scikit-learn)의 비교: 사이킷런은 단일 서버에서만 구동 가능하지만, 스파크ML은 분산 서버에서 구동 가능합니다. (아, 참고로 사이킷런에도 멀티프로세싱 병렬 처리 수행 기능은 있습니다!) 이 때문에 사이킷런은 학습 데이터가 단일 서버의 메모리에 수용될 수 없을 정도로 거대하면 기능 수행이 불가합니다. 이럴 때는 MapReduce로 용량을 줄이는 등에 조치가 필요합니다.하지만 사이킷런은 스파크ML보다 더 많은 머신러닝 알고리즘을 지원하며, API의 유연성도 스파크ML에 비해 매우 뛰어납니다. 게다가 파이썬의 ML 생태계가 사이킷런에 맞춰져 있기 때문에 병렬 처리가 필요하지 않을 정도의 데이터셋은 사이킷런을 이용하는 게 오히려 이득입니다. 보면 알겠지만 일장일단이 매우 뚜렷하죠.
• 스파크에서 구동 가능한 머신러닝 라이브러리: 스파크에 자체적으로 내장된 MLlib가 있으며, SparkDL XGBoost, Synapse ML 등은 분산 병렬 처리가 가능하며, 그 중 일부는 GPU도 지원합니다. 그 외에 사이킷런 등 외부 머신러닝 라이브러리도 구동은 가능합니다. 하지만 아까 설명한 것과 같이, 구동만 가능하지 분산 병렬 처리가 지원되지 않습니다. 예외적으로 GridSearch와 같은 최적 하이퍼파라미터 탐색 시에는 분산 병렬 처리 기능이 지원됩니다.
• 데이터브릭스(Databricks): 웹 브라우저에서 스파크 개발을 할 수 있는 클라우드 기반의 개발환경으로, 무료 이용이 가능한 커뮤니티 버전에서 실습을 진행했습니다. 회원가입은 https://www.databricks.com/kr/try-databricks?itm_data=NavBar-TryDatabricks-Trial#account에서 진행했지만, 무료 버전인 계정 생성 시 '커뮤니티 에디션으로 시작하기'를 선택했기 때문에 실제 개발과 실습은 커뮤니티 버전 링크에서 진행했습니다.데이터브릭스 실습은 클러스터(cluster)라는 가상 컴퓨터를 생성하는 것에서 시작되며, 이는 주피터 노트북의 커널과 유사한 포지션을 담당합니다.
  • 커뮤니티 에디션에 대한 이모저모: 클라우드 서비스로 구동되는 데이터브릭스의 특성 상 한꺼번에 많은 작업과 트래픽이 중앙 서버에 집중되면 클러스터가 다운되는 경우가 있습니다. 유료 과금 버전에서는 이를 재시작(restart)할 수 있지만, 커뮤니티 버전에서는 그것이 안 되기 때문에 홈페이지 좌측 툴바 중에서 Compute 탭에 들어가 기존 클러스터를 삭제하고 클러스터를 새로 만들어야 진행을 할 수 있습니다. 심지어 새 클러스터 생성도 유료 과금러들은 빠르게 진행되지만, 커뮤니티 버전은 5~10분 가량 기다려야 하는 불편함이 있습니다. 돈으로 시간을 산다.
• 빅데이터 처리에 대한 고찰: 스파크는 빅데이터를 여러 개로 분할해서 계산하기 때문에 빅데이터를 빠르게 처리할 수 있는 장점이 있습니다. 문제는 빅데이터가 아닌 경우에 발생합니다. 예를 들어 충분히 크지 않은 데이터를 8개로 나눴다고 가정하면, 스파크에서는 데이터를 8개 집단으로 나누는 시간이 따로 걸리고, 결과를 모으는 시간도 따로 소모되기 때문에 오히려 스파크를 쓰지 않았을 때보다 시간이 더 소모되는 참사 경우도 발생합니다.아래 예시는 소수(prime number)를 계산하는 코드입니다. 각각 스파크를 사용했을 때와 사용하지 않았을 때의 결과입니다. 약 62,500개의 데이터를 8개의 집단으로 쪼개서 연산한 결과, 스파크를 쓰지 않은 케이스가 조금 더 연산이 빠른 것을 확인했습니다.

# 스파크 사용 코드
from pyspark.sql import SparkSession

sc = SparkSession.builder.master("yarn").appName("spark_prime01").getOrCreate()

# 소수를 계산할 최대 수
MAX = 500000

# 3 이상 MAX + 1 미만의 수를 저장한 리스트 생성
# 해당 리스트는 8개의 클러스터로 동시 실행
rdd = sc.sparkContext.parallelize([i for i in range(3, MAX + 1)], 8)

def get_prime(num):
    isPrime = True

    for index in range(2, num):
        if num % index == 0:
            isPrime = False
            break

    if isPrime == True:
        return num

import time
start = time.time() # 현재 시간을 시작 시간에 대입

# rdd.map(get_prime): rdd에 저장된 숫자들을 순서대로 get_prime 함수에 대입
# .collect(): 함수 실행 결과를 가져옴
prime_list = rdd.map(get_prime).collect()

end = time.time() # 현재 시간을 end에 대입

print("소요 시간:", end - start)

# 스파크를 사용하지 않은 코드
import time

MAX = 500000

lst01 = [i for i in range(3, MAX +1)

def get_prime(num):
    isPrime = True

    for index in range(2, num):
        if num % index == 0:
            isPrime = False
            break

    if isPrime == True:
        return num

start = time.time() # 현재 시간을 시작 시간에 대입

# rdd.map(get_prime): rdd에 저장된 숫자들을 순서대로 get_prime 함수에 대입
# .collect(): 함수 실행 결과를 가져옴
prime_list = rdd.map(get_prime).collect()

end = time.time() # 현재 시간을 end에 대입

print("소요 시간:", end - start)

스파크 데이터프레임

• RDD 개요: 위의 코드 중 스파크를 이용하는 부분을 보면 rdd라는 변수 이름을 볼수 있습니다. RDD는 Resilient Distributed Dataset의 약자로, 분산 데이터셋을 위한 스파크의 데이터 추상화 구조입니다. 스파크는 RDD를 이용해 데이터를 클러스터 내의 서로 다른 노드에 저장하고 액세스합니다.
• 판다스 데이터프레임과의 비교: 판다스의 데이터프레임은 넘파이의 API와 어느 정도 호환성을 유지합니다. 그래서 넘파이 유니버설 함수를 직접적으로 적용할 수 있었던 것이죠. 다만 병렬 CPU 처리가 불가능하고 단일 서버의 메모리 용량 이상의 데이터를 처리할 수 없습니다.반대로 스파크 데이터프레임은 SQL과 유사한 형태의 연산을 지원하는 API 구성으로 판다스 API와는 여러 모로 다릅니다. 스파크를 설치하면 하이브를 내장하고 있는데 이것과 같은 맥락이라고 볼 수 있습니다.

스파크를 이용한 타이타닉 데이터셋 EDA 및 생존자 예측

• 머신러닝 파트에서도 이미 한 번 건드린 바가 있는 타이타닉 데이터셋을 스파크로 분석하는 실습을 진행했습니다.
• EDA(탐색적 데이터 분석)에 대한 고찰: 분석 및 예측을 진행할 때는 그 문제에 대해서 개발자가 잘 모르거나 실제 사실과 다른 상식을 가지고 있는 경우가 많습니다. 도메인 지식이 부족하다는 뜻이죠. 따라서 데이터를 분석함으로써 문제에 대한 인사이트를 얻고 잘못된 상식을 올바르게 고치는 과정이 필요합니다. 데이터를 이해하는 것은 예측과 분석에 필요한 새로운 컬럼을 생성하는 데 밑바탕이 되어줍니다.이 실습의 경우 Survived 컬럼과 관련 있는 컬럼이 어떤 것인지 또한 관련이 없는 컬럼이 어떤 것인지를 알아내고 생존과 관련된 새로운 컬럼 생성 및 인사이트를 얻는 데 도움을 줄 것입니다.

from pyspark.sql.functions import col, isnan, when, count
from pyspark.sql.types import IntegerType, DoubleType
from pyspark import SparkFiles
import os
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns

plt.style.use('seaborn')

# 글자 크기 설정
sns.set(font_scale=2)

# 경고 메시지가 출력되지 않도록 설정
import warnings
warnings.filterwarnings('ignore')

url = "https://drive.google.com/uc?id=1hLyqMbl5foc96lDJk6_KWJ_RLVOtIzIJ"
spark.sparkContext.addFile(url)

df_train = spark.read.csv("file://" + SparkFiles.get("uc"), header=True)

column_name_list = df_train.schema.names

# int_column_list의 숫자 타입 컬럼이 String 타입으로 설정되어 있으므로 숫자로 변환
int_column_list = column_name_list[1:2] + column_name_list[5:7] + column_name_list[11:12]
for column_name in int_column_list:
    print("column_name =", column_name)
    df_train = df_train.withColumn(column_name, df_train[column_name].cast('int'))
    print('='* 100)

# 같은 방식으로 실수 타입의 컬럼 전환
double_column_list = column_name_list[4:5] + column_name_list[8:9]
for column_name in double_column_list:
    print("column_name =", column_name)
    df_train = df_train.withColumn(column_name, df_train[column_name].cast('double'))
    print('='* 100)

import missingno as msno
# msno.bar는 각 컬럼의 null이 아닌 데이터의 비율을 그래프로 그려주는 패키지
msno.bar(df=df_train.toPandas())

# SQL 사용을 위해서 Spark DataFrame을 View 형태로 변환
# View: 데이터를 추가 또는 수정할 수 없는 테이블
df_train.createOrReplaceTempView('titanic_view')

# Sex와 Survived 칸을 교차해서 빈도수 출력
df_train.crosstab('Sex', 'Survived').show()

# 각 성별별로 Pclass 컬럼과 Survived 컬럼을 교차해서 그래프를 그림
# sns.factorplot(컬럼1, 컬럼2, hue=범주에 사용할 컬럼, data=데이터프레임)
sns.factorplot('Pclass', 'Survived', hue='Sex', data=df_train.toPandas())

# df_train['Survived']==1인 행 대입
df_train.filter(df_train["Survived"]==1).show()

# Age 컬럼에서 각 나이별 사망자 생존자 비율을 그래프로 그림
plt.figure(figsize=(15, 8))
sns.kdeplot(data=df_train.toPandas(), x="Age", hue="Survived")
plt.legend(['Survived_age', 'Dead_age'])
plt.show()

numeric_column_list = []

for column in df_train.schema:
    # 컬럼명 조회
    print("column.name =", column.name)
    # 컬럼 타입 조회
    print("column.dataType =", column.dataType)

    # 컬럼 타입이 IntegerType(): 정수 또는
    # 컬럼 타입이 DoubleType(): 실수일 경우
    if (column.dataType == IntegerType()) or (column.dataType == DoubleType()):
        print("numeric type")
        numeric_column_list.append(column.name)

    print("=" * 100)

for column_name in numeric_column_list:
    print("column_name =", column_name)

    # column_name 컬럼과 Survived 컬럼의 연관 관계
    print(df_train.corr(column_name, "Survived"))
    print("=" * 100)

 

 

from pyspark.sql.functions import col, isnan, when, count
from pyspark.sql.types import IntegerType, DoubleType
from pyspark import SparkFiles
import os
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns

plt.style.use('seaborn')
sns.set(font_scale=2)

import warnings
warnings.filterwarnings('ignore')

url = "https://drive.google.com/uc?id=1hLyqMbl5foc96lDJk6_KWJ_RLVOtIzIJ"

spark.sparkContext.addFile(url)

df_train = spark.read.csv("file://" + SparkFiles.get("uc"), header=True)

column_name_list = df_train.schema.names

int_column_list = column_name_list[1:2] + column_name_list[5:7] + column_name_list[11:12]
for column_name in int_column_list:
    print("column_name =", column_name)
    df_train = df_train.withColumn(column_name, df_train[column_name].cast("int"))
    print("=" * 100)

double_column_list = column_name_list[4:5] + column_name_list[8:9]
for column_name in double_column_list:
    print("column_name =", column_name)
    df_train = df_train.withColumn(column_name, df_train[column_name].cast("double"))
    print("=" * 100)

# df_train를 Pandas DataFrame으로 변환해서 psample_df에 대입
psample_df = df_train.toPandas()

# 가장 기울어진 컬럼 시각화
plt.hist(psample_df["Fare"])
plt.show()

from pyspark.sql.functions import log1p
# log1p("Fare"): Fare 컬럼에 log(1 + 데이터)를 취한 값을 리턴
df_train = df_train.withColumn("Fare", log1p("Fare"))

psample_df = df_train.toPandas()

plt.hist(psample_df["Fare"])
plt.show()

# 배우자 수와 부모 아이 수를 합해서 FamilySize 컬럼 생성
df_train = df_train.withColumn("FamilySize", df_train["SibSp"] + df_train["Parch"] + 1)
display(df_train)

df_train.crosstab("FamilySize", "Survived").show()

import re
# 다음 단어 중 하나를 포함하는 문자열을 찾기 위한 객체 p 생성
p = re.compile("Mrs|Mr|Miss|Master|Don|Dr|Ms|Major|Mlle|Mme|Countess|Lady|Don|Countess")

# 이름에서 Mr, Mrs, Miss를 리턴하는 함수
# 매개변수 name_row: Name 컬럼의 이름 하나씩 순서대로 대입되는 매개변수
def get_sir(name_row):
    if p.search(name_row) != None:
        if p.search(name_row)[0] in ["Countess", "Lady"]:
            return "Mrs"
        elif p.search(name_row)[0] in ["Mme", "Ms", "Mlle"]:
            return "Miss"
        elif p.search(name_row)[0] in ["Don", "Dr", "Major"]:
            return "Mr"
        else:
            return p.search(name_row)[0]
    else:
        return "Other"

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType
# get_sir 함수를 Spark DataFrame의 컬럼 값들을 수정해서 리턴할 수 있는 함수로 변환
# udf(lambda 컬럼의 데이터 하나씩 저장할 변수: 함수명, 리턴받을 결과의 타입)

udf_get_sir = udf(lambda x: get_sir(x), StringType())

df_train = df_train.withColumn("Sir", udf_get_sir(df_train["Name"]))
display(df_train)

# Mrs, Miss, Master의 생존률이 높음
df_train.crosstab("Sir", "Survived").show()

# SQL 쿼리 활용을 위한 View 변환
df_train.createOrReplaceTempView('titanic_view')

 

 

• 여기까지 진행한 후 RandomForest 모델의 시각화를 위해 graphviz와 dtreeviz를 설치했습니다. 기존 노트북을 clone한 후 맨 위쪽에 블록 2개를 추가하고 아래와 같은 명령어를 실행시켰습니다.

  %sh
  apt-get install -y graphviz

%sh
pip install dtreeviz[pyspark]

이후 클러스터를 재실행한 뒤, 맨 아래 블록으로 다시 이동해서 실습합니다.

df_train = df_train.drop("PassengerId") # drop 여부를 확인하기 위해 먼저 한 개의 컬럼만 제거해봤습니다.
df_train = df_train.drop("Name", "Sex", "Age", "Ticket", "Cabin", "Embarked", "Sir")
columns = df_train.columns
columns.remove("Survived")

from pyspark.ml.feature import VectorAssembler
assembler = VectorAssembler(inputCols=columns, outputCol="features")
assembler_df = assembler.transform(df_train)
display(assembler_df)

# assembler_df의 데이터를 trainingData:testData = 7:3으로 분할
(trainingData, testData) = assembler_df.randomSplit([0.7, 0.3])

from pyspark.ml.classification import RandomForestClassifier

rf = RandomForestClassifier(featuresCol="features", labelCol="Survived", numTrees=100)
rfModel = rf.fit(trainingData)

rf_prediction = rfModel.transform(testData)
right_predict = rf_prediction.filter(rf_prediction.Survived == rf_prediction.prediction)

from pyspark.ml.evaluation import MulticlassClassificationEvaluator
evaluator_accuracy = MulticlassClassificationEvaluator(
    labelCol='Survived', predictionCol='prediction', metricName='accuracy'
)
evaluator_accuracy.evaluate(rf_prediction)

dataset = trainingData.toPandas()
dataset = dataset[columns + ["Survived"]]

import dtreeviz

viz_model = dtreeviz.model(
    rfModel.trees[99], # RandomForest의 99번째 Decision Tree 객체

    X_train=dataset[columns], y_train=dataset["Survived"],
    feature_names=columns, target_name="Survived",
    class_names=["Dead", "Survived"]
)

v = viz_model.view().svg()

# Decision Tree의 내용 출력
displayHTML(v)

같은 방식으로 XGBoost 분류기를 사용했을 때와, 최적의 파라미터를 찾는 실습을 진행했습니다.

from sparkdl.xgboost import XgboostClassifier

xgb_classifier = XgboostClassifier(featuresCol="features",
                                   labelCol="Survived",
                                   n_estimators=500, # Decision Tree의 개수
                                   missing=0.0) # 결측치를 치환할 값

xgb_model = xgb_classifier.fit(trainingData)
xgb_prediction = xgb_model.transform(testData)

# 기존 랜덤 포레스트보다 낮은 정확도. 튜닝 필요.
acc_eval = MulticlassClassificationEvaluator(
    labelCol="Survived", predictionCol="prediction", metricName="accuracy"
)
acc_eval.evaluate(xgb_prediction)

from hyperopt import fmin, tpe, hp, STATUS_OK, Trials

# hp.quniform: 시작값, 종료값, 간격 순으로 값을 입력받아 그 범위에서 최적값을 찾음
# hp.uniform: 시작값, 종료값만 받아서 최적값을 찾음
search_space = {
    'max_depth': hp.quniform('max_depth', 3, 20, 1),
    'min_child_weight':hp.quniform('min_child_weight', 3, 20, 1),
    'colsample_bytree':hp.uniform('colsample_bytree', 0.5, 1.0),
    'learning_rate':hp.uniform('learning_rate', 0.01, 0.5)
}

def objective_func(space):
    xgb_classifier = XgboostClassifier(featuresCol="features",
                                       labelCol="Survived",
                                       n_estimators=500,
                                       missing=0.0,
                                       num_works=3,
                                       max_depth=int(space["max_depth"]),
                                       min_child_weight=int(space["min_child_weight"]),
                                       colsample_bytree=space["colsample_bytree"],
                                       learning_rate=space["learning_rate"])
    xgb_model = xgb_classifier.fit(trainingData)
    accuracy = acc_eval.evaluate(xgb_prediction)
    print(f"space = {space}, accuracy = {accuracy}")
    print("=" * 100)
    return {'loss': -1 * accuracy, 'status':STATUS_OK}

algo = tpe.suggest
trials = Trials()
# 파라미터들의 조합 중 가장 정확도가 높은 파라미터 반환
best = fmin(fn=objective_func, space=search_space, max_evals=100,
            algo=algo, trials=trials)

# 가장 정확도가 높은 파라미터로 XGboost 객체 생성
xgb_best = XgboostClassifier(featuresCol="features",
                             labelCol="Survived",
                             n_estimators=500,
                             missing=0.0,
                             num_works=3,
                             max_depth=int(best["max_depth"]),
                             min_child_weight=int(best["min_child_weight"]),
                             colsample_bytree=best["colsample_bytree"],
                             learning_rate=best["learning_rate"])

xgb_model = xgb_best.fit(trainingData)
xgb_predictions = xgb_model.transform(testData)
accuracy = acc_eval.evaluate(xgb_predictions)
print(f"accuracy = {accuracy}")

 

스파크를 이용한 실시간 데이터 처리

• 도커 가상환경과 윈도우 환경 사이에 파일을 공유할 공유 폴더를 하나 생성합니다. C:\bigdata_system\share라는 디렉터리를 생성합니다. 이후 도커 허브로부터 강사님이 제공해주신 도커 이미지를 pull 해와서 컨테이너를 만들고 docker run --hostname=localhost --privileged=true -v C:\bigdata_system\share:/root/share -it -p 8089:8088 -p 8084:8083 -p 9871:9870 -p 8043:8042 -p 9094:9093 -p 9202:9201 -p 9302:9301 -p 5602:5601 -p 8889:8888 -p 8081:8080 -p 8877:8877 -p 9092:9092 -p 8083:8083 encore0906/spark_env01 /bin/bash로 실행했습니다.
• 한글 설정을 한 뒤, ./spark_start.sh 명령어로 스파크를 시작합니다. 이 파일에는 하이브, 스파크, 그리고 데이터브릭스 페이지에서 타이타닉 머신러닝을 구현했던 프로그램이기도 한 스파크 개발 환경 Zeppeline(제펠린)이 포함되어 있습니다.
• wget -O /root/bus_train.csv https://drive.google.com/uc?id=1Yg5pcV3QRlGv2oSP4kEEqx0H10vrBdDF로 실습용 데이터를 받아옵니다. 다운이 됐으면 확인을 위해 cat /root/bus_train.csv 명령어로 내용을 출력합니다.
• 학습데이터를 HDFS에 업로드합니다. hdfs dfs -mkdir /data, hdfs dfs -put /root/bus_train.csv /data로 업로드 한 뒤, hdfs dfs -ls /data로 데이터와 그 권한을 확인합니다.
• 제펠린의 실행은 엣지(Edge) 브라우저에서 진행합니다. 주소는 http://localhost:8877/입니다. 이후 새 노트북을 생성하고 제주 버스 데이터의 실시간 모델을 만드는 실습을 진행했습니다. 하지만 용량 부족에 대처하기 위한 트러블슈팅으로 인해 시간이 부족해서 데이터 특성 확인과 전처리까지만 진행하고 실제 학습된 모델은 월요일에 pre-trained model을 받아와서 진행하기로 했습니다.

앞으로 바라는 점

• 제펠린 특징: 데이터브릭스 홈페이지와는 다르게 자동완성이 지원되지 않는 것 같습니다. 다른 설정이나 플러그인이 없는지 찾아봐야겠습니다.
• 머신러닝/딥러닝 때와 마찬가지로 수업 내용과 가르쳐주시는 태도가 너무 부실해서 이 내용들을 조금 더 실무 지향적이고 철저하게 한 번 더 가르쳐주실 수 없을까 싶습니다.