Kubeflow Pipeline (1) - 데이터 준비

이번 글 부터는 본격적으로 Kubeflow 주요 리소스인 Pipeline의 개념과 사용법에 대해서 알아보겠습니다.
Kubeflow 공식 문서에 따르면 Kubeflow Pipeline의 구성과 목적은 다음과 같습니다.

구성

1. ML/DL의 실험, 작업, 실행을 관리하고 사용하기 위한 UI 제공
2. 다단계 ML 워크플로를 예약하기 위한 엔진
3. Pipeline과 구성요소를 정의하고 조작하기 위한 SDK

목적

1. End-to-End 오케스트레이션 : ML Pipeline의 오케스트레이션을 활성화하고 단순화
2. 쉬운 실험 : 다양한 아이디어와 기술을 쉽게 시도하고 다양한 실험을 관리
3. 간편한 재사용 : 구성요소와 파이프라인을 재사용하여 매번 다시 빌드할 필요 없이 End-to-End 솔루션을 빠르게 생성

실습 주제는 올해 7~8월 DACON에서 진행된 전력량 예측 AI 경진대회에서 사용한 데이터와 XGBoost Regressor 모델 학습 프로세스를 Kubeflow Pipeline으로 구축하는 것 입니다. (https://dacon.io/competitions/official/236125/overview/descripti)

실습 프로세스

1. 학습 데이터 DB에 저장 (전처리 완료된 학습 데이터)
2. 컴포넌트 구성요소 설계
   (1). Data Load
   (2). Data Split (Train, Test)
   (3). Model Train
   (4). Model Test & Evaluation
3. 각각의 컴포넌트 py코드 정의
4. 컨테이너 이미지로 빌드
5. Pipeline.py 코드 정의
6. Kubeflow에서 Pipeline 실행 및 확인

---

1. 학습 데이터 DB에 저장

• Kubernetes 내에 MySQL DB Pod를 생성하고 쉽게 접근할 수 있도록 서비스도 같이 생성해 줍니다. 데이터가 영구 저장될 수 있도록 PVC도 정의합니다. (Kubeflow Pipeline 사용법에 중점을 두었기 때문에 DB는 단순하게 구성하였습니다.)

(1) PVC 생성

#[control Node]
$ mkdir ~/db
$ cd db
$ vim mydb-pvc.yaml

# mydb-pvc.yaml
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
  name: mydb-pvc
spec:
  accessModes:
    - ReadWriteMany
  resources:
    requests:
      storage: 10Gi
      # Storage Class를 직접 정의하지 않으면 default SC로 자동 설정됩니다.

(2) DB Pod 생성

# mydb.yaml
apiVersion: v1
kind: Pod
metadata:
  name: mydb
  labels:
    app: mydb
spec:
  containers:
  - name: mydb
    image: mysql:latest
    env:
    - name: MYSQL_ROOT_PASSWORD
      value: qwer1234
    ports:
    - containerPort: 3306
    volumeMounts:
    - name: db
      mountPath: /var/lib/mysql
  volumes:
  - name: db
    persistentVolumeClaim:
      claimName: mydb-pvc

(3) Service 생성

# mydb-svc.yaml
apiVersion: v1
kind: Service
metadata:
  name: mydb-svc
spec:
  type: NodePort
  ports:
  - port: 3306
    targetPort: 3306
  selector:
    app: mydb

(4) 리소스 생성, 접속 및 데이터베이스 생성

# db 디렉토리에서 수행
$ kubectl create -f .
$ kubectl exec -it mydb -- mysql -u root -p  # ROOT PW 입력하여 접속
...
mysql> CREATE DATABASE trainset;   # "trainset"이라는 이름의 DB 생성

(5) Python DB 연결 및 데이터 업로드

• Kubeflow Notebook 환경에서 MySQL DB와 연결하고 전처리한 데이터를 업로드 합니다.
• 구체적인 전처리 방식은 생략하지만 Feature들은 숫자형 타입이어야 하고, 목표변수인 전력소비량(kWh) 컬럼은 power 라는 이름의 컬럼으로 지정하였습니다.
• sql과 python 코드와의 연결은 sqlaclchemy 패키지를 사용하였습니다.

# 필요 패키지
import pandas as pd
import os
import numpy as np
from sklearn.preprocessing import MinMaxScaler
from sqlalchemy import create_engine

# 전처리 클래스 정의
class Preprocessing(object):
    def __init__(self):
      # 원본 훈련 데이터 불러오기
      self.train=pd.read_csv('https://raw.githubusercontent.com/Parkjiwonha/prj_jwjk/main/train.csv')
        
    def preprocessing(self):
        train_df=self.train
        # 전처리 관련 코드
        ...
        return traindf
    
    # 데이터 업로드
    def data_upload(self, df):
        db_connection_str = 'mysql+pymysql://root:qwer1234@10.233.51.86/trainset' # mydb-svc의 CLUSTER-IP 및 데이터베이스 입력
        db_connection = create_engine(db_connection_str)
        try:
            conn = db_connection.connect()
            print('DB 연결에 성공하였습니다.')
            df.to_sql(name='traindf', con=conn, if_exists='replace',index=False) # traindf라는 이름의 테이블 생성
            conn.close()
            print('데이터 업로드를 완료하였습니다.')
        except:
            print('DB 연결/업로드에 실패하였습니다.')
            

• mydb의 CLUSTER-IP 확인

$ kubectl get svc
NAME         TYPE        CLUSTER-IP     EXTERNAL-IP   PORT(S)          AGE
kubernetes   ClusterIP   10.233.0.1     <none>        443/TCP          65m
mydb-svc     NodePort    10.233.51.86   <none>        3306:31078/TCP   20m

• 전처리 데이터프레임 -> DB 업로드

# Python
preproc = Preprocessing()
df = preproc.preprocessing()
preproc.data_upload(df)

• DB -> 데이터프레임 불러오기

# python
sql_statement = ''' SELECT * FROM traindf'''
new_df = pd.read_sql(sql=sql_statement, con=conn)
conn.close()

여기까지의 과정을 통해 원본 데이터를 전처리 한 후, Kubernetes 클러스터의 데이터베이스에 저장하였습니다.
이후부터 Kubeflow Pipeline을 통해서 데이터를 Load하고 Split, Train, Test & Evaluation을 컴포넌트로 하나씩 구성해 보도록 하겠습니다.