KFServing InferenceService 배포와 예측 – PyTorch

PyTorch를 사용하는 InferenceService

학습이 완료된 PyTorch 모델을 저장 한 경우, KFServing에서 제공하는 PyTorch 서버를 사용하여 간단히 배포 할 수 있습니다.

전제 조건

  • state_dict() 사용하여 모델을 저장해야 합니다.

모델 생성

PyTorch 서버를 테스트하려면 먼저 파이썬을 사용하여 간단한 PyTorch 모델을 생성해야 합니다.

추론을 위해 모델을 저장할 때, 학습이 완료된 모델의 학습 매개 변수만 저장하면 됩니다. torch.save()함수를 사용하여 모델의 state_dict 를 저장하면 나중에 모델을 복원 할 때 유연성이 가장 높아 지므로 모델 저장에 권장되는 방법입니다. 현재 KFServing은 PyTorch가 추론을 위해 권장하는 모델 저장 방법인 state_dict() 메소드를 사용하여 저장된 PyTorch 모델을 지원합니다.

PyTorch의 KFServing 인터페이스는 사용자가 PyTorch 모델과 동일한 위치에 model_class_file을 업로드 할 것으로 예상하고 선택적 model_class_name을 런타임 입력으로 전달하도록 허용합니다. 모델 클래스 이름을 지정하지 않으면 ‘PyTorchModel’을 기본 클래스 이름으로 사용합니다. 다른 인터페이스를 사용하여 저장된 PyTorch 모델을 지원하기 위해, 이 인터페이스가 발전함에 따라 현재 인터페이스가 변경 될 수 있습니다.

PyTorch의 데이터셋 중의 하나인 cifar10 데이터를 분류하는 모델을 작성해 보겠습니다.

cifar10.py

import torch
import torchvision
import torchvision.transforms as transforms
import torch.nn as nn
import torch.nn.functional as F
import torch.optim as optim


class Net(nn.Module):
    def __init__(self):
        super(Net, self).__init__()
        self.conv1 = nn.Conv2d(3, 6, 5)
        self.pool = nn.MaxPool2d(2, 2)
        self.conv2 = nn.Conv2d(6, 16, 5)
        self.fc1 = nn.Linear(16 * 5 * 5, 120)
        self.fc2 = nn.Linear(120, 84)
        self.fc3 = nn.Linear(84, 10)

    def forward(self, x):
        x = self.pool(F.relu(self.conv1(x)))
        x = self.pool(F.relu(self.conv2(x)))
        x = x.view(-1, 16 * 5 * 5)
        x = F.relu(self.fc1(x))
        x = F.relu(self.fc2(x))
        x = self.fc3(x)
        return x


if __name__ == "__main__":

    transform = transforms.Compose(
        [transforms.ToTensor(),
         transforms.Normalize((0.5, 0.5, 0.5), (0.5, 0.5, 0.5))])

    trainset = torchvision.datasets.CIFAR10(root='./data', train=True,
                                            download=True, transform=transform)
    trainloader = torch.utils.data.DataLoader(trainset, batch_size=4,
                                              shuffle=True, num_workers=2)

    testset = torchvision.datasets.CIFAR10(root='./data', train=False,
                                           download=True, transform=transform)
    testloader = torch.utils.data.DataLoader(testset, batch_size=4,
                                             shuffle=False, num_workers=2)

    classes = ('plane', 'car', 'bird', 'cat',
               'deer', 'dog', 'frog', 'horse', 'ship', 'truck')

    net = Net()

    criterion = nn.CrossEntropyLoss()
    optimizer = optim.SGD(net.parameters(), lr=0.001, momentum=0.9)

    for epoch in range(2):  # loop over the dataset multiple times

        running_loss = 0.0
        for i, data in enumerate(trainloader, 0):
            # get the inputs; data is a list of [inputs, labels]
            inputs, labels = data

            # zero the parameter gradients
            optimizer.zero_grad()

            # forward + backward + optimize
            outputs = net(inputs)
            loss = criterion(outputs, labels)
            loss.backward()
            optimizer.step()

            # print statistics
            running_loss += loss.item()
            if i % 2000 == 1999:    # print every 2000 mini-batches
                print('[%d, %5d] loss: %.3f' %
                      (epoch + 1, i + 1, running_loss / 2000))
                running_loss = 0.0

    print('Finished Training')

    # Save model
    torch.save(net.state_dict(), "model.pt")

생성 된 모델을 사용하여 PyTorch 서버를 실행하고 예측을 수행 할 수 있습니다. 모델은 PV, S3 호환 가능 개체 저장소, Azure Blob 저장소 또는 Google Cloud Storage에 있을 수 있습니다.

모델 저장하기

쿠버네티스의 퍼시스턴스 볼륨에 모델을 저장해 보겠습니다. PVC 는 앞서 생성한 kfserving-models-pvc 을 사용하겠습니다. 모델을 학습시키기 위해서 쿠버네티스 잡(Job)을 사용하겠습니다. Job을 생성할 때 모델을 저장하기 위한 PVC를 마운트 해줍니다.

모델 코드 작성하기

cifa10 이미지를 분류하는 모델입니다. 모델을 저장할 위치를 --model_path 파라미터로 입력받게 하였습니다. state_dict() 메소드를 사용하여 모델을 저장하게 되면, 학습된 파라미터만 저장이 됩니다. 그래서 모델이 정의된 파이썬 파일이 별도로 필요합니다. 모델이 정의된 파이썬 파일도 같이 저장소에 복사해 줍니다.

ipytorch_cifar10.py

import argparse
import os
import shutil

import torch
import torch.nn as nn
import torch.nn.functional as F
import torch.optim as optim
import torchvision
import torchvision.transforms as transforms


class Net(nn.Module):
    def __init__(self):
        super(Net, self).__init__()
        self.conv1 = nn.Conv2d(3, 6, 5)
        self.pool = nn.MaxPool2d(2, 2)
        self.conv2 = nn.Conv2d(6, 16, 5)
        self.fc1 = nn.Linear(16 * 5 * 5, 120)
        self.fc2 = nn.Linear(120, 84)
        self.fc3 = nn.Linear(84, 10)

    def forward(self, x):
        x = self.pool(F.relu(self.conv1(x)))
        x = self.pool(F.relu(self.conv2(x)))
        x = x.view(-1, 16 * 5 * 5)
        x = F.relu(self.fc1(x))
        x = F.relu(self.fc2(x))
        x = self.fc3(x)
        return x


if __name__ == "__main__":

    parser = argparse.ArgumentParser()
    parser.add_argument('--model_path', default='/mnt/pv/models/pytorch/iris', type=str)
    args = parser.parse_args()

    model_path = args.model_path
    if not (os.path.isdir(model_path)):
        os.makedirs(model_path)

    model_file = os.path.join(model_path, 'model.pt')

    transform = transforms.Compose(
        [transforms.ToTensor(),
         transforms.Normalize((0.5, 0.5, 0.5), (0.5, 0.5, 0.5))])

    trainset = torchvision.datasets.CIFAR10(root='./data', train=True,
                                            download=True, transform=transform)
    trainloader = torch.utils.data.DataLoader(trainset, batch_size=4,
                                              shuffle=True, num_workers=2)

    testset = torchvision.datasets.CIFAR10(root='./data', train=False,
                                           download=True, transform=transform)
    testloader = torch.utils.data.DataLoader(testset, batch_size=4,
                                             shuffle=False, num_workers=2)

    classes = ('plane', 'car', 'bird', 'cat',
               'deer', 'dog', 'frog', 'horse', 'ship', 'truck')

    net = Net()
    if torch.cuda.is_available():
        print('Use GPU')
        net = net.cuda()

    criterion = nn.CrossEntropyLoss()
    optimizer = optim.SGD(net.parameters(), lr=0.001, momentum=0.9)

    for epoch in range(2):  # loop over the dataset multiple times

        running_loss = 0.0
        for i, data in enumerate(trainloader, 0):
            # get the inputs; data is a list of [inputs, labels]
            inputs, labels = data

            # zero the parameter gradients
            optimizer.zero_grad()

            # forward + backward + optimize
            outputs = net(inputs)
            loss = criterion(outputs, labels)
            loss.backward()
            optimizer.step()

            # print statistics
            running_loss += loss.item()
            if i % 2000 == 1999:  # print every 2000 mini-batches
                print('[%d, %5d] loss: %.3f' %
                      (epoch + 1, i + 1, running_loss / 2000))
                running_loss = 0.0

    print('Finished Training')

    # Save model
    torch.save(net.state_dict(), model_file)

    shutil.copy(os.path.abspath(__file__), os.path.join(model_path, __file__))

컨테이너 이미지를 만들기

컨테이너 이미지를 만들기 위한 Dockerfile 입니다. 파이썬을 기본 이미지로 사용하고, torchtorchvision 패키지를 추가로 설치합니다.

Dockerfile

FROM python:3.6-slim

RUN pip install torch torchvision

RUN mkdir -p /app
ADD pytorch_cifar10.py /app/

쿠버네티스 잡 실행하기

컨테이너 이미지를 빌드하고, 컨테이너 이미지 레지스트리에 푸시 한 다음, 쿠버네티스 잡(Job)을 생성하겠습니다.

Job을 생성할 때는 모델을 저장하기 위해서 PVC를 마운트 해줍니다. 이 일련의 작업들은 직접 실행 할 수 있습니다. 하지만 좀 더 편하게 하기 위해서 앞서 배운 Kubeflow Fairing을 사용하겠습니다.

다음은 로컬 개발 환경에서 Fairing을 사용하여 컨테이너 이미지를 만들고, 쿠버네티스 잡을 실행하는 예제입니다.

fairing-local-docker.py

import uuid
from kubeflow import fairing
from kubeflow.fairing.kubernetes import utils as k8s_utils

CONTAINER_REGISTRY = 'kangwoo'

namespace = 'admin'
job_name = f'sklean-iris-job-{uuid.uuid4().hex[:4]}'

command=["python", "pytorch_cifar10.py", "--model_path", "/mnt/pv/models/pytorch/cifar10"]
output_map = {
    "Dockerfile": "Dockerfile",
    "pytorch_cifar10.py": "pytorch_cifar10.py"
}

fairing.config.set_preprocessor('python', command=command, path_prefix="/app", output_map=output_map)

fairing.config.set_builder('docker', registry=CONTAINER_REGISTRY, image_name="pytorch-cifar10", dockerfile_path="Dockerfile")

fairing.config.set_deployer('job', namespace=namespace, job_name=job_name,
                            pod_spec_mutators=[k8s_utils.mounting_pvc(pvc_name='kfserving-models-pvc', pvc_mount_path='/mnt/pv')],
                            cleanup=True, stream_log=True)

fairing.config.run()

fairing을 실행하면 쿠버네티스 잡이 생성되고, 학습이 완료된 모델이 지정한 경로에 저장됩니다.

PyTorch을 사용하는 InferenceService 로 예측 하기

InferenceService 생성

InferenceService 매니페스트를 작성합니다. predictor로 pytorch 을 사용합니다. storageUri 필드로 모델 저장 위치를 지정해 줍니다. pvc 의 이름이 kfserving-models-pvc 이고 저장 위치가 models/pytorch/cifar10/ 이므로, pvc://kfserving-models-pvc/models/pytorch/cifar10/ 라고 지정해 줍니다.

앞서 학습 할때 사용한 모델 클래스의 이름이 Net 입니다. modelClassName 에 모델 클래스의 이름인 Net 를 지정해 줍니다. 만약 모델 클래스 이름을 별도로 지정해 주지 않으면, 기본값인 PyTorchModel 을 모델 클래스 이름으로 사용합니다.

pytorch.yaml

apiVersion: "serving.kubeflow.org/v1alpha2"
kind: "InferenceService"
metadata:
  name: "pytorch-cifar10"
spec:
  default:
    predictor:
      pytorch:
        storageUri: "pvc://kfserving-models-pvc/models/pytorch/cifar10/"
        modelClassName: "Net"

만약 GPU 리소스를 사용하려고 한다면, resource 필드에 GPU 를 할당해 주면됩니다.

pytorch_gpu.yaml

apiVersion: "serving.kubeflow.org/v1alpha2"
kind: "InferenceService"
metadata:
  name: "pytorch-cifar10-gpu"
spec:
  default:
    predictor:
      pytorch:
        storageUri: "pvc://kfserving-models-pvc/models/pytorch/cifar10/"
        modelClassName: "Net"
        resources:
          limits:
            cpu: 100m
            memory: 1Gi
            nvidia.com/gpu: "1"

InferenceService 를 생성합니다.

다음은 admin 네임스페이스 InferenceService 를 생성하는 예제입니다.

kubectl -n admin apply -f pytorch.yaml

생성한 InferenceService를 조회해 보겠습니다.

kubectl -n admin get inferenceservice

InferenceService 가 정상적으로 생성되면 다음과 같은 응답 결과를 확인할 수 있습니다.

NAME              URL                                                                  READY   DEFAULT TRAFFIC   CANARY TRAFFIC   AGE
pytorch-cifar10   <http://pytorch-cifar10.admin.example.com/v1/models/pytorch-cifar10>   True    100                                59s

예측 실행하기

예측을 요청하기 위해서는 모델 서버에 접근해야 합니다. 모델 서버는 ingressgateway 를 통해서 접근할 수 있습니다. ingressgateway 는 모델 서버들을 구분하기 위해서 호스트 이름을 사용합니다. ingressgateway에 접근하 기 위한 주소는 앞서 정의한 CLUSTER_IP 를 사용하겠습니다.

예측을 요청할 데이터를 json 파일로 작성합니다.

데이터의 크기가 크기 때문에 git 에 있는 파일을 다운받아서 사용해주세요.

cifar10-input.json

{
   "instances":[
      [
         [
            [
               0.23921573162078857,
               0.24705886840820312,
               0.29411768913269043,
               0.301960825920105,
               0.2549020051956177,
               0.22352945804595947,
               0.2705882787704468,
               0.24705886840820312,
               0.23921573162078857,
               0.24705886840820312,
               0.2627451419830322,
               0.2549020051956177,
               0.2627451419830322,
               0.301960825920105,
               0.32549023628234863,
               0.3333333730697632,
               0.30980396270751953,
               0.2705882787704468,
               0.2549020051956177,
               0.2549020051956177,
               0.22352945804595947,
               0.16862750053405762,
               0.17647063732147217,
               0.16078436374664307,
               0.16862750053405762,
               0.12156867980957031,
               0.09803926944732666,
               0.10588240623474121,
               0.12156867980957031,
               0.07450985908508301,
               -0.011764705181121826,
               -0.09019607305526733
            ]
....
         ]
      ]
   ]
}

다음은 admin 네임스페이스의 xgboost-iris InferenceService 에 예측을 요청하는 예제입니다.

MODEL_NAME=pytorch-cifar10
SERVICE_HOSTNAME=$(kubectl -n admin get inferenceservice pytorch-cifar10 -o jsonpath='{.status.url}' | cut -d "/" -f 3)

INPUT_PATH=@./cifar10-input.json
curl -v -H "Host: ${SERVICE_HOSTNAME}" http://$CLUSTER_IP/v1/models/$MODEL_NAME:predict -d $INPUT_PATH

정상적으로 실행되면 다음과 같은 응답 결과를 확인 할 수 있습니다.

*   Trying ::1...
* TCP_NODELAY set
* Connected to localhost (::1) port 8080 (#0)
> POST /v1/models/pytorch-cifar10:predict HTTP/1.1
> Host: pytorch-cifar10.admin.example.com
> User-Agent: curl/7.64.1
> Accept: */*
> Content-Length: 110681
> Content-Type: application/x-www-form-urlencoded
> Expect: 100-continue
> 
< HTTP/1.1 100 Continue
* We are completely uploaded and fine
< HTTP/1.1 200 OK
< content-length: 224
< content-type: text/html; charset=UTF-8
< date: Sat, 29 Mar 2020 13:36:23 GMT
< server: istio-envoy
< x-envoy-upstream-service-time: 8857
< 
* Connection #0 to host localhost left intact
{"predictions": [[-1.0642542839050293, -2.1011602878570557, 0.829179584980011, 2.3874664306640625, -0.8927481174468994, 1.6838929653167725, 1.1490685939788818, -0.8248656988143921, -0.6572933197021484, 0.05098217725753784]]}

댓글 남기기

이메일은 공개되지 않습니다. 필수 입력창은 * 로 표시되어 있습니다