Kubeflow에서 tf.Keras training

 

 

Kubeflow에서 Training을 하기 위해 쓰는 방법은 크게 2종류가 있다.

 

1. Kubeflow에서 Jupyter 를 사용하기

Kubeflow에서 Jupyter를 만들때, Jupyter에 사용할 jupyter server image를 선택하는 부분에서 tensorflow-cpu? gpu? 선택 부분이 있고, resource할당에 cpu, gpu 코어수를 어떻게 할당할지 정할 수 가 있다.

 

이 부분에서 GPU를 충분히 할당해주고 실행하면 EMR 클러스터 등에서 사용하듯이 분산학습도 수행 할 수 가 있다.

 

장점 : 쓰기 너무 쉽다. <- 중요포인트다. Training code짜고, 논문 읽고 모델과 데이터 품질에 대한 아이디어 짤 시간도 없는 AI Researcher한테 파이프라인 맞춰서 코드 짜오라고 하는건 좋지 않다.

 

단점 :

- pipeline 과 함께 실행할 수 없어 continuous learning이나, 여러 데이터 조건에 따른 모델 차이를 보기 위해 학습을 여러개를 동시에 돌린다던지 하는 작업을 전혀 수행 할 수 없다.

- jupyter를 생성할때 정해둔 resource만 사용 할 수 있기 때문에 중간에 scale out 하려고 하면 jupyter를 다시 만들어야 한다.

- 이게 내가 경험적으로 느낀거라 정확하지 않을 수 있는데, jupyter notebook으로 할당 가능한 리소스 양이 kubeflow의 각 component들이 붙잡고 사용하지 않는 리소스를 쓰지 못하는 것 같아보인다.

(kubeflow의 각 componenet들은 실행은 되고 있지만 대부분의 시간이 뭔가 처리를 하지는 않는 idle 상태인거 같다.

그래서 kubectl top node를 쳐보면 항상 cpu usage는 10퍼센트 이하, memory usage는 30퍼센트 근처이다.)

여기의 안쓰고 대기하는 리소스를 쓰지 못하는 것 같다.

최소 사양만 딱 갖춘 kubeflow에서 위처럼 리소스가 남아도는데, jupyter core 수를 아주 조금만 늘려도 juptyer가 제대로 생성이 되지 않는다. 그래서 t3a.large * 2 개 사이즈의 리소스를 그냥 놀도록 둬야 하더라.

그래서 jupyter는 core 0.5개만 써서 단순 mnist 학습의 epoch 당 시간이 39초씩 걸렸는데(위 스크린샷), 밑에 기술하는 TFJob은 epoch 당 시간이 8~9초 정도 밖에 걸리지 않았다.

 

 

아래는 TFJob으로 kubectl apply -f 를 통해 학습을 수행한 로그이다.

Epoch 1/5
1875/1875 [==============================] - 8s 4ms/step - loss: 0.2944 - accuracy: 0.9151
Epoch 2/5
1875/1875 [==============================] - 7s 4ms/step - loss: 0.1440 - accuracy: 0.9568
Epoch 3/5
1875/1875 [==============================] - 9s 5ms/step - loss: 0.1086 - accuracy: 0.9668
Epoch 4/5
1875/1875 [==============================] - 7s 4ms/step - loss: 0.0890 - accuracy: 0.9727
Epoch 5/5
1875/1875 [==============================] - 7s 4ms/step - loss: 0.0765 - accuracy: 0.9763
2023-07-16 15:10:22.712567: W tensorflow/tsl/framework/cpu_allocator_impl.cc:83] Allocation of 31360000 exceeds 10% of free system memory.
313/313 - 1s - loss: 0.0780 - accuracy: 0.9755 - 1s/epoch - 3ms/step

 

 

 

 

 

 

 

2. TFJob을 사용하기

 

Jupyter가 아니라 Kubernetes 클러스터 자체를 사용하는 학습은 TFJob을 사용해야 한다.

 

설명은 공식 홈페이지를 참고 해도 좋다.

https://www.kubeflow.org/docs/components/training/tftraining/

 

이걸 도대체 어떻게 쓰는건가 하고 한참동안 헤맸었는데

 

사용법은 이렇다. (내가 짠 예제 코드를 올려보겠다.)

(사용하기 무지 귀찮고 복잡한데 쉽게 하는 법 있으면 댓글로 가르쳐주시면 감사하겠습니다...

kubeflow가 이거 저거 쓰기 쉽게 모아놨다고 해서 기대했는데 생각보다 각자 따로 알아서 써야 되서

불편한거 같기도 하고...)

 

1) training code를 작성한다. (.py파일)

Jupyter 상에서 돌리는게 아니니까 hyper parameter만 바꾸고 싶으면 class화 해서 argument를 받는게 좋다.

import tensorflow as tf
import tensorflow.keras as keras

from tensorflow.keras.layers import Flatten
from tensorflow.keras.layers import Dense
from tensorflow.keras.layers import Dropout

from tensorflow.keras.models import Model


print(tf.__version__)
print(keras.__version__)

(x_train, y_train) , (x_test, y_test) = keras.datasets.mnist.load_data()
x_train, x_test = x_train / 255.0, x_test / 255.0


input_x = keras.layers.Input(shape=(28,28))

x = Flatten()(input_x)
x = Dense(128, activation='relu')(x)
x = Dropout(0.2)(x)
output_y = Dense(10, activation='softmax')(x)

model = Model(inputs = input_x, outputs = output_y)


model.summary()


model.compile(optimizer='adam', loss='sparse_categorical_crossentropy', metrics=['accuracy'])

#짜놓고 보니까 valid set을 안넣었네?
model.fit(x_train, y_train, epochs=5)

model.evaluate(x_test, y_test, verbose=2)

 

2) 이 .py 파일을 실행하는 docker container를 만든다. (dockerize)

#escs34/mnist에 push해뒀습니다.
From tensorflow/tensorflow:2.12.0

ADD mnist_jupyter_test.py /var/source/mnist_jupyter_test.py

#코드 실행은 yaml파일에서 지정할거라 안해도 됨 (여기서 지정하거나, yaml 파일에서 지정하거나 한번만.)
#예제코드 보니까 single은 yaml에서 distributed는 여기서 지정하길래 따라해봄

#ENTRYPOINT ["python", "/var/source/mnist_jupyter_test.py"]

 

3) 이 docker file을 build해서 registry에 올려둔다. (ecr이건 gcr이건 docker hub건)

docker build -t escs34/mnist:latest .

docker push escs34/mnist:latest

4) 이 container image를 kubernetes에서 사용하기 위해 yaml파일을 작성합니다.

apiVersion: "kubeflow.org/v1"
kind: "TFJob"
metadata:
  name: "mnist"
  namespace: "kubeflow"
spec:
  runPolicy:
    cleanPodPolicy: None
  tfReplicaSpecs:
    Worker:
      replicas: 1
      restartPolicy: Never
      template:
        spec:
          containers:
            - name: tensorflow
              #컨테이너 직접 따로 만들어놔야 함.
              image: escs34/mnist:latest
              command:
                - "python"
                - "/var/source/mnist_jupyter_test.py"
              #데이터 다운로드 등을 위해 pv가 필요할때 아래처럼
              #하거나 pipeline에서 minio를 연결해주면 됨
          #    volumeMounts:
          #      - mountPath: "/train"
          #        name: "training"
          #volumes:
          #  - name: "training"
          #    persistentVolumeClaim:
          #      claimName: "tfevent-volume"

 

5) 이걸 이제 kubeflow apply -f single_tf_job.yaml 로 실행해야 한다.

 

6) 이걸 pipe라인으로 쓰고 싶으면 

jupyter에서 comp.load_component_from_file("yaml파일이름") 을 써야 한다고 한다... <-여긴 아직 안해봤다.

 

 

 

#예제 학습 코드나 pipeline은 공식 github를 참고하면 Tensorflow, Pytorch, Keras 유저인가에 따라 찾아 볼 수 있다.

https://github.com/kubeflow/training-operator/tree/master/examples

 

 

 

 

 

 

위와 같은 선택과, 특성때문에 필연적으로 학습 코드가 pipeline까지 도착하는 과정은 다음과 같을것으로 추정된다.

 

 

1. jupyter에서 학습 및 가설 검증

2. py 코드로 만듦

3. dockerize 시킴 => container가 실행될때 파이썬 코드 실행시키게 만듦

4. 이미지를 registry에 띄움

5. tfjob yaml 파일 작성함

6. kubectl apply -f 로 실행해서 동작 테스트

7. yaml파일을 jupyter에서 읽어서 pipeline으로 만듦

 

너~~~무 길고 복잡한데?

 

한번 짜 놓으면 좀 자동화 된다거나 하면 모르겠는데 다 일일이 손으로 만들어야 되는 느낌이고...

 

내가 뭘 잘못하고 있는 걸까? 

 

아무튼 분산 학습 코드도 간단히 실행해 봤다.

 

import tensorflow as tf
import tensorflow.keras as keras

from tensorflow.keras.layers import Flatten
from tensorflow.keras.layers import Dense
from tensorflow.keras.layers import Dropout

from tensorflow.keras.models import Model

print(tf.__version__)
print(keras.__version__)

(x_train, y_train) , (x_test, y_test) = keras.datasets.mnist.load_data()
x_train, x_test = x_train / 255.0, x_test / 255.0

strategy = tf.distribute.MirroredStrategy()


print('num of machine: {}'.format(strategy.num_replicas_in_sync))


with strategy.scope():
    input_x = keras.layers.Input(shape=(28,28))

    x = Flatten()(input_x)
    x = Dense(128, activation='relu')(x)
    x = Dropout(0.2)(x)
    output_y = Dense(10, activation='softmax')(x)

    model = Model(inputs = input_x, outputs = output_y)
    model.compile(optimizer='adam', loss='sparse_categorical_crossentropy', metrics=['accuracy'])


model.summary()

model.fit(x_train, y_train, epochs=5)

model.evaluate(x_test, y_test, verbose=2)

그냥 같은 코드에 mirrored strategy 넣은게 전부다 원하면 parameter server도 쓸 수 있을거다.

 

#https://github.com/kubeflow/training-operator/blob/master/examples/tensorflow/distribution_strategy/keras-API/multi_worker_tfjob.yaml
#mirrored strategy는 paramter server를 쓰지 않음
apiVersion: "kubeflow.org/v1"
kind: "TFJob"
metadata:
  name: "dist-mnist"
  namespace: "kubeflow"
spec:
  runPolicy:
    cleanPodPolicy: None
  tfReplicaSpecs:
    Worker:
      replicas: 2
      restartPolicy: Never
      template:
        spec:
          containers:
            - name: tensorflow
              #컨테이너 직접 따로 만들어놔야 함.
              image: escs34/dist_mnist:latest
              #command:
              #  - "python"
              #  - "/var/source/mnist_jupyter_test.py"
              #데이터 다운로드 등을 위해 pv가 필요할때 아래처럼
              #하거나 pipeline에서 minio를 연결해주면 됨
          #    volumeMounts:
          #      - mountPath: "/train"
          #        name: "training"
          #volumes:
          #  - name: "training"
          #    persistentVolumeClaim:
          #      claimName: "tfevent-volume"

 

Worker 2대를 써서 했는데...

 

아무리 봐도... 

 

그냥 워커 2대에서 각자 모델 1개씩 만든거 같은데...

 

저 코드 자체는 전에도  v100 gpu 2대가 달린 single node에서 써본 적 있어서 아는데 확실히 데이터와 모델 크기가 커지면 학습 시간이 절반으로 줄어든다.

 

그러니까... worker도 2대 됐고, 코드도 잘 짰는데... 중간 과정에 문제가 있는거 같기도 하고... 추가 테스트 되면 또 내용을 보태 넣어야 겠다.