KFServing InferenceService 배포와 예측

InferenceService를 사용하여 모델 서버를 제공하려면, 사용할 네임스페이스가 다음과 같은지 확인해야합니다.

  • [serving.kubeflow.org/inferenceservice=enabled](<http://serving.kubeflow.org/inferenceservice=enabled>) 레이블이 네임스페이스 추가 되어 있어야 합니다.
  • 쿠버네티스 클러스터의 Istio IngressGateway에 접근할 수 있어야 합니다.

레이블 추가

Kubeflow의 대시보드나 프로필 컨트롤러(Profile Controller)를 사용하여, 사용자 네임스페이스를 만드는 경우에는 KFServing에서 모델을 배포할 수 있도록 serving.kubeflow.org/inferenceservice: enabled 레이블이 자동으로 추가됩니다. 만약 네임스페이스를 직접 생성하는 경우에는 해당 네임스페이스에 serving.kubeflow.org/inferenceservice: enabled 레이블을 추가해야만 합니다.

다음은 my-namespace 라는 네임스페이스에 레이블을 추가하는 예제입니다.

kubectl label namespace my-namespace serving.kubeflow.org/inferenceservice=enabled

Istio IngressGateway에 접근하기

InferenceService 가 정상적으로 생성되면, istio의 ingressgateway 를 모델 서버에 접속할 수 있습니다. KFServing에서 사용하는 ingressgateway의 이름을 알려면, config-istio 라는 ConfigMap을 조회하면 됩니다.

다음은 knative-serving 네임스페이스에 있는 config-istio 을 조회하는 예제입니다.

kubectl -n knative-serving get cm config-istio -o yaml

정상적으로 조회 되면 다음과 같은 결과를 얻을 수 있습니다.

apiVersion: v1
data:
  gateway.knative-serving.knative-ingress-gateway: kfserving-ingressgateway.istio-system.svc.cluster.local
  local-gateway.knative-serving.cluster-local-gateway: cluster-local-gateway.istio-system.svc.cluster.local
  local-gateway.mesh: mesh
  reconcileExternalGateway: "false"
kind: ConfigMap
metadata:
  ...
  name: config-istio
  namespace: knative-serving

data 섹션의 gateway.knative-serving.knative-ingress-gateway 필드가 현재 KFServing에서 사용하는 ingressgateway 를 설정하는 부분입니다. 위의 예제에서는 kfserving-ingressgateway를 사용하고 있습니다.

kfserving-ingressgateway를 조회해 보겠습니다.

다음은 istio-system 네임스페이스에 있는 kfserving-ingressgateway을 조회하는 예제입니다.

kubectl -n istio-system get service kfserving-ingressgateway 

KFServing이 설치된 쿠버네티스 클러스터에 따라 결과가 다르게 나옵니다. 응답 결과에 따른 크게 세가지 방법으로 접근 할 수 있습니다.

  • LoadBalancer 를 통해서 접근하기
  • NodePort를 통해서 접근하기
  • kubectl port-forward를 통해서 접근하기

LoadBalancer

쿠버네티스 클러스터가 LoadBalancer 를 지원하면 다음과 같은 결과를 얻을 수 있습니다. 서비스의 타입이 LoadBalancer 이고, EXTERNAL-IP 에 IP가 할당되어 있습니다. 이럴 경우에는 EXTERNAL-IP 를 통해서 ingressgateway에 접근할 수 있습니다.

NAME                       TYPE           CLUSTER-IP      EXTERNAL-IP   PORT(S)                                                                                                                                                                                   AGE
kfserving-ingressgateway   LoadBalancer   10.101.141.37   10.201.121.4  15020:30543/TCP,80:32380/TCP,443:32390/TCP,31400:32400/TCP,15011:30263/TCP,8060:32119/TCP,853:32180/TCP,15029:32156/TCP,15030:30674/TCP,15031:30230/TCP,15032:32563/TCP,15443:30995/TCP   2d23h

앞으로 만들 예제에서 사용하기 위해서 ingressgateway 의 접근 주소를 다음과 같이 정의하겠습니다. EXTERNAL-IP 주소를 사용합니다.

CLUSTER_IP=10.201.121.4

NodePort

쿠버네티스 클러스터가 LoadBalancer 를 지원하지 않거나, 서비스의 타입이 NodePort 인 경우 EXTERNAL-IP 의 값이 비어 있습니다. 이럴 경우에는 클러스터의 노드 IP 와 NodePort를 통해서 접근할 수 있습니다.

NAME                       TYPE           CLUSTER-IP      EXTERNAL-IP   PORT(S)                                                                                                                                                                                   AGE
kfserving-ingressgateway   LoadBalancer   10.101.141.37   <pending>     15020:30543/TCP,80:32380/TCP,443:32390/TCP,31400:32400/TCP,15011:30263/TCP,8060:32119/TCP,853:32180/TCP,15029:32156/TCP,15030:30674/TCP,15031:30230/TCP,15032:32563/TCP,15443:30995/TCP   2d23h

노드 IP는 노드를 조회하면 알 수 있습니다.

다음은 노드를 조회 하는 예제입니다.

kubectl get node -o wide

정상적으로 조회되면 다음과 같은 응답 결과가 나옵니다.

NAME     STATUS   ROLES    AGE   VERSION    INTERNAL-IP     EXTERNAL-IP   OS-IMAGE             KERNEL-VERSION      CONTAINER-RUNTIME
mortar   Ready    master   13d   v1.15.10   192.168.21.38   <none>        Ubuntu 18.04.3 LTS   4.15.0-91-generic   docker://18.9.9

노드가 한 개가 아닌 경우에는 여러개의 노드 정보가 출력됩니다. 해당 노드들 중에서 아무 노드의 INTERNAL-IP 를 사용하면 됩니다.

앞으로 만들 예제에서 사용하기 위해서 ingressgateway 의 접근 주소를 다음과 같이 정의하겠습니다. 노드의 IP 와 80 PORT(80:32380/TCP)의 노드 포트인 32380을 포트로 사용합니다.

CLUSTER_IP=192.168.21.38:32380

port-forward

외부에서 쿠버네티스 클러스터의 서비스에 접근할 수 없는 경우, kubectl 의 port-forward를 사용할 수 있습니다. 접근 하려는 외부 시스템에서 다음 명령어 실행하면 로컬 포트를 경유 해서 쿠버네티스 서비스에 접근할 수 있습니다.

kubectl -n istio-system port-forward svc/kfserving-ingressgateway 8080:80

포트 포워딩이 정상적으로 실행되면, 로컬포트로 ingressgateay 서비스로 접근할 수 있습니다. http://localhost:8080 처럼 선언한 로컬 포트의 주소로 접근하면, 쿠버네티스 ingressgateway 의 80 포트로 포워딩 됩니다.

앞으로 만들 예제에서 사용하기 위해서 ingressgateway 의 접근 주소를 다음과 같이 정의하겠습니다.

CLUSTER_IP=localhost:8080

PVC 생성하기

InferenceService 에 사용할 모델은 PVC에 저장하겠습니다. 만약 클라우드 스토리지와 같은 다른 저장소를 사용하려면, “클라우드 저장소를 이용하여 InfeerneceService 배포와 예측”을 참조하시기 바랍니다.

kfserving-models-pvc라는 PVC 매니페스트를 작성합니다.

kfserving-models-pvc.yaml

kind: PersistentVolumeClaim
apiVersion: v1
metadata:
  name: kfserving-models-pvc
spec:
  accessModes:
    - ReadWriteOnce
  resources:
    requests:
      storage: 1Gi

다음 명령어를 실행하여, admin 네임스페이스에 kfserving-models-pvc라는 PVC를 생성하겠습니다.

kubectl -n admin apply kfserving-models-pvc.yaml

KFServing 설정

KFServing 에서 사용하는 여러 설정 정보들은 inferenceservice-config 라는 쿠버네티스 ConfigMap에 정의되어 있습니다.

이 설정 정보에는 다음과 같은 내용이 정의되어 있습니다.

  • credentials : S3나 GCS를 사용할 때 참조할 값들.
  • explainers : explainer를 실행할 때 사용할 컨테이너의 이미지 정보.
  • ingress : KFServing에서 사용할 Istio ingress 정보.
  • predictors : predictor를 실행할 때 사용할 컨테이너 이미지의 정보.

다음 명령어를 실행하면, 설정 정보를 조회할 수 있습니다.

kubectl -n kubeflow get cm inferenceservice-config -o yaml

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

apiVersion: v1
data:
  credentials: |-
    {
       "gcs": {
           "gcsCredentialFileName": "gcloud-application-credentials.json"
       },
       "s3": {
           "s3AccessKeyIDName": "awsAccessKeyID",
           "s3SecretAccessKeyName": "awsSecretAccessKey"
       }
    }
  explainers: |-
    {
        "alibi": {
            "image" : "gcr.io/kfserving/alibi-explainer",
            "defaultImageVersion": "0.2.2",
            "allowedImageVersions": [
               "0.2.2"
            ]
        }
    }
  ingress: |-
    {
        "ingressGateway" : "knative-ingress-gateway.knative-serving",
        "ingressService" : "kfserving-ingressgateway.istio-system.svc.cluster.local"
    }
  logger: |-
    {
        "image" : "gcr.io/kfserving/logger:0.2.2",
        "memoryRequest": "100Mi",
        "memoryLimit": "1Gi",
        "cpuRequest": "100m",
        "cpuLimit": "1"
    }
  predictors: |-
    {
        "tensorflow": {
            "image": "tensorflow/serving",
            "defaultImageVersion": "1.14.0",
            "defaultGpuImageVersion": "1.14.0-gpu",
            "allowedImageVersions": [
               "1.11.0",
               "1.11.0-gpu",
               "1.12.0",
               "1.12.0-gpu",
               "1.13.0",
               "1.13.0-gpu",
               "1.14.0",
               "1.14.0-gpu"
            ]
        },
        "onnx": {
            "image": "mcr.microsoft.com/onnxruntime/server",
            "defaultImageVersion": "v0.5.1",
            "allowedImageVersions": [
               "v0.5.1"
            ]
        },
        "sklearn": {
            "image": "gcr.io/kfserving/sklearnserver",
            "defaultImageVersion": "0.2.2",
            "allowedImageVersions": [
               "0.2.2"
            ]
        },
        "xgboost": {
            "image": "gcr.io/kfserving/xgbserver",
            "defaultImageVersion": "0.2.2",
            "allowedImageVersions": [
               "0.2.2"
            ]
        },
        "pytorch": {
            "image": "gcr.io/kfserving/pytorchserver",
            "defaultImageVersion": "0.2.2",
            "allowedImageVersions": [
               "0.2.2"
            ]
        },
        "tensorrt": {
            "image": "nvcr.io/nvidia/tensorrtserver",
            "defaultImageVersion": "19.05-py3",
            "allowedImageVersions": [
               "19.05-py3"
            ]
        }
    }
  storageInitializer: |-
    {
        "image" : "gcr.io/kfserving/storage-initializer:0.2.2",
        "memoryRequest": "100Mi",
        "memoryLimit": "1Gi",
        "cpuRequest": "100m",
        "cpuLimit": "1"
    }
  transformers: |-
    {
    }
kind: ConfigMap
metadata:
...
  name: inferenceservice-config
  namespace: kubeflow

Kubeflow – KFServing 설치

KFServing는 Kubeflow의 구성 요소로 포함되어 있습니다. 별도로 설치가 필요 없이 사용할 수 있습니다. 물론 Kubeflow 없이 독립적으로 설치해서 사용할 수도 있습니다.

전제 조건

KFServing을 사용하려면, 쿠버네티스 클러스터에 Knative Serving 및 Istio가 설치되어 있어야 합니다. Knative는 Istio Ingress Gateway를 사용하여 요청을 Knative 서비스로 라우팅합니다. Kubeflow 및 KFServing 팀이 테스트 한 정확한 버전을 사용하려면 개발자 안내서의 전제 조건을 참조하십시오

Knative를 빠르게 실행하거나 서비스 메시가 필요하지 않은 경우, 서비스 메시(sidecar injection 없이 Istio를 설치하는 것이 좋습니다.

현재는 Knative Serving 만 필요합니다. cluster-local-gateway 는 클러스터 내부 트래픽을 위한 통로로 사용합니다. cluster-local-gateway를 설치하려면 여기의 지침을 따르십시오

KFServing 웹훅 인증서를 제공합니다.

KFServing 설치

Kubeflow와 함께 KFServing 설치

KFServing 은 Kubeflow를 설치할때 기본적으로 설치됩니다. Kubeflow 매니페스트에 KFServing을 설치하는 부분이 포함되어 있습니다. Kubeflow와 함께 설치되는 KFServing의 경우 KFServing 컨트롤러는 kubeflow  네임스페이스에 배포됩니다. Kubeflow의 쿠버네티스 최소 요주 버전이 1.14이므로 개체 선택기(object selector)를 지원하지 않을 수 있습니다. 그래서 Kubeflow 설치시 ENABLE_WEBHOOK_NAMESPACE_SELECTOR 가 기본적으로 활성화 되어 있어야합니다.

Kubeflow의 대시보드나 프로필 컨트롤러(Profile Controller)를 사용하여, 사용자 네임스페이스를 만드는 경우에는 KFServing에서 모델을 배포할 수 있도록 serving.kubeflow.org/inferenceservice: enabled 레이블이 자동으로 추가됩니다. 만약 네임스페이스를 직접 생성하는 경우에는 해당 네임스페이스에 serving.kubeflow.org/inferenceservice: enabled 레이블을 추가해야만, KFServing의  InferenceService 를 사용할 수 있습니다.

독립형 KFServing 설치

쿠버네티스 클러스터에 KFServing을 독립적으로 설치하면, 우선 위의 전제 조건을 충족시켜야 합니다. 전제 조건이 충족되면 다음 명령어를 사용하여 KFServing을 설치할 수 있습니다. 다음 명령어는 GitHub 리포지토리의 yaml 파일을 사용하여 KFServing 0.3.0을 설치합니다.

TAG=v0.3.0
CONFIG_URI=https://raw.githubusercontent.com/kubeflow/kfserving/master/install/$TAG/kfserving.yaml

kubectl apply -f ${CONFIG_URI}

KFServing을 독립형으로 설치했을 경우에는 KFServing 컨트롤러는 kfserving-system 네임스페이스에 배포됩니다.

KFServing은 pod mutator와 mutating admission webhooks 을 사용하여 KFServing의 스토리지 이니셜라이저(storage initializer) 컴포넌트를 주입합니다. 기본적으론 네임스페이스에 control-plane 레이블이 지정되어 있지 않으면, 해당 네임스페이스의 포드들은 pod mutator를 통과합니다. 그렇기 때문에 KFServing의 pod mutator의 웹훅이 필요 없는 포드가 실행될때 문제가 발생할 수 있습니다.

쿠버네티스 1.14 사용자의 경우 serving.kubeflow.org/inferenceservice: enabled 레이블이 추가된 네임스페이스의 포드에 ENABLE_WEBHOOK_NAMESPACE_SELECTOR 환경변수를 추가하여, KFServing pod mutator를 통과하도록 하는게 좋습니다.

env:
- name: ENABLE_WEBHOOK_NAMESPACE_SELECTOR
  value: enabled

쿠버네티스 1.15+ 사용자의 경우 KFServing InferenceService 포드만 pod mutator 를 통과 할 수 있도록 객체 선택기(object selector)를 켜는 것이 좋습니다.

kubectl patch mutatingwebhookconfiguration inferenceservice.serving.kubeflow.org --patch '{"webhooks":[{"name": "inferenceservice.kfserving-webhook-server.pod-mutator","objectSelector":{"matchExpressions":[{"key":"serving.kubeflow.org/inferenceservice", "operator": "Exists"}]}}]}'

Kubeflow – KFServing 개요

KFServing은 쿠버네티스에서 서버리스 추론을 가능하게 하며, TensorFlow, XGBoost, scikit-learn, PyTorch 및 ONNX와 같은 일반적인 머신러닝 프레임워크를 위한 고성능의 추상화 인터페이스를 제공합니다. 그래서 프로덕션에서 다양한 프레임워크의 모델을 서빙하기에 적합합니다.

KFServing을 이용하여 모델을 서빙하려면, InferenceService 라는 쿠버네티스 사용자 리소스를 생성하면 됩니다.

KFServing의 장점은 다음과 같습니다.

  • 다양한 머신 러닝 프레임워크 제공하기 위한, 쿠버네티스 사용자 리소스의 추상화가 잘 되어 있습니다. 그래서 쉽고 간편한게 추론 서비스를 생성할 수 있습니다.
  • 자동 확장, 네트워킹, 상태 확인 및 서버 구성의 복잡성을 캡슐화하여 GPU 자동 확장 및 카나리아 롤아웃과 같은 최첨단 서비스 기능을 머신러닝 배포에 사용할 수 있습니다.
  • 기본적으로 예측, 전처리, 후처리 및 설명 기능을 제공하여 프로덕션 머신러닝 추론 서버에 대해 간단하고 플러그 가능하며 완벽한 제품을 만들 수 있습니다.

KFServing는 Kubeflow의 함께 설치됩니다. 그래서 별도의 설치 없이 사용할 수 있습니다. 물론 Kubeflow 없이 독립적으로 설치해서 사용할 수도 있습니다.

모델 서버

KFServing 은 다음과 같은 머신러닝 프레임워크를 지원하는 모델 서버를 제공하고 있습니다.

  • Tensorflow
  • NVIDIA Triton Inference Server
  • PyTorch
  • XGBoost
  • scikit-learn
  • ONNX

이러한 머신러닝 프레임워크를 사용하여 모델을 저장한 경우에는 Google 버킷, S3 버킷, Azure 또는 minio에 저장된 모델의 위치만 있으면 쉽게 추론 서비스를 생성할 수 있습니다.

다음은 scikit-learn과 tensorflow의 매니페스트 예제입니다.

scikit-learn

apiVersion: "serving.kubeflow.org/v1alpha2"
kind: "InferenceService"
metadata:
  name: "sklearn-iris"
spec:
  default:
    predictor:
      sklearn:
        storageUri: "gs://kfserving-samples/models/sklearn/iris"

pytorch

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

tensorflow

apiVersion: "serving.kubeflow.org/v1alpha2"
kind: "InferenceService"
metadata:
  name: "flowers-sample"
spec:
  default:
    predictor:
      tensorflow:
        storageUri: "gs://kfserving-samples/models/tensorflow/flowers"

storageUri는 학습한 모델의 저장 경로입니다.

모델 저장소

storageUri에서 사용할 수 있는, 지원하는 스토리는 다음과 같습니다.

  • Google Cloud Storage : 접두사가 “gs://” 로 시작합니다.
    • 기본적으로 사용자 인증에 GOOGLE_APPLICATION_CREDENTIALS 환경 변수를 사용합니다.
    • GOOGLE_APPLICATION_CREDENTIALS 가 제공되지 않으면, 익명 클라이언트가 아티팩트를 다운로드합니다.
  • S3 Compatible Object Storage : 접두사가 “s3://” 로 시작합니다.
    • 기본적으로 사용자 인증을 위해 S3_ENDPOINT, AWS_ACCESS_KEY_IDAWS_SECRET_ACCESS_KEY 환경 변수를 사용합니다.
  • Azure Blob Storage : “https://{$STORAGE_ACCOUNT_NAME}.blob.core.windows.net/{$CONTAINER}/{$PATH}”
    • 기본적으로 익명 클라이언트를 사용하여 아티팩트를 다운로드합니다.
  • 로컬 파일 시스템 : 접두사가 없거나 접두사가 “file://” 로 시작합니다.
    • 절대 경로: /absolute/path or file:///absolute/path
    • 상대 경로 : relative/path or file://relative/path
    • 로컬 파일 시스템의 경우 접두사 없이 상대 경로를 사용하는 것이 좋습니다.
  • Persistent Volume Claim (PVC) : 접두사가 “pvc://” 로 시작합니다.
    • 경로 형태는 “pvc://{$pvcname}/[path]” 입니다.
    • pvcname은 모델을 저장하고 있는 PVC의 이름입니다.
    • [path]“는 PVC의 모델에 대한 상대 경로입니다.
    • For e.g. pvc://mypvcname/model/path/on/pvc

KFServing 스택

다음은 KFServing 스택을 나타내는 그림입니다.

출처 : kfsrving

KFServing 은 쿠버네티스 위에서 동작합니다. 그리고 istio와 Knative를 사용하고 있습니다.

Istio

서비스를 연결(Connect), 보안(secure), 제어(control) 그리고 관찰(observe) 하기 위한 서비스 메쉬(Service Mesh) 플랫폼입니다.

  • Connect : 서비스 간의 트래픽 및 API 호출 흐름을 지능적으로 제어하고 다양한 테스트를 수행하며 Red/Black 배포를 제공합니다.
  • Secure : 관리 인증, 권한 부여 및 서비스 사이의 통신 암호화를 통해 서비스를 자동으로 보호합니다.
  • Control : 정책을 적용하고 시행해서 자원이 소비자에게 공정하게 분배되도록 합니다.
  • Observe : 모든 서비스의 트레이싱, 모니터링 및 로깅을 관찰하여 발생하는 상황을 확인합니다.

Knative

Knative는 선언적인 컨테이너 기반 애플리케이션을 구축하는데 필수적인 미들웨어 구성 요소 세트를 제공합니다 Knative는 서빙(Serving) 컴포넌트과 이벤트(Eventing) 컴포넌트로 구성되어 있습니다.

Knative Eventing은 클라우드 네이티브 개발에 대한 일반적인 요구를 해결하도록 설계된 시스템이며 바인딩 가능한 이벤트 소스 및 이벤트 소비자를 가능하게하는 구성 요소를 제공합니다.

Knative Serving은 애플리케이션이나 함수들을 서버리스 컨테이너를 사용하여 배포와 서빙 할 수 있는 기능을 지원합니다. Knative Serving은 사용 하기 쉽고, 여러 고급 시나리오를 지원하도록 확장할 수 있습니다.

Knative Serving 프로젝트는 다음을 가능하게 하는 미들웨어 기본 요소를 제공합니다.

  • 서버리스 컨테이너의 빠른 배포
  • 자동 스케일링
  • Istio 구성 요소의 라우팅 및 네트워크 프로그래밍
  • 배포 된 코드 및 구성의 특정 시점 스냅 샷

KFServing

KFServing은 모델의 호스팅 측면을 관리합니다. KFServing 은 추론을 서비스하기위 해서, 쿠버네티스의 사용자 리소스인 InferenceService 를 제공하고 있습니다. InferenceService 를 생성하게 되면, 모델 서버가 실행되어 추론 요청을 처리할 수 있습니다.

  • InferenceService : 모델의 생명주기를 관리합니다
  • Configuration : 모델 배포 기록을 관리합니다. 기본(Default) 및 카나리아(Canary)의 두 가지 구성이 존재합니다.
  • Revision : 모델 버전의 스냅샷 입니다. 설정 정보와 이미지 정보를 가지고 있습니다.
  • Route : 네트워크 트래픽을 관리 하는 엔드 포인트 입니다.


KFServing Data Plane

InferenceService 의 데이터 플레인은 Predictor, Explainer, Transformer 로 구성되어 있습니다. 이중에서 실제 예측을 수행하는 모델 서버인 Predictor가 핵심 컴포넌트 입니다. 그리고 모델을 안전하게 변경할 수 있도록, Default”와 “Canary” 엔드포인트를 가지고 있습니다.

다음은 InferenceService의 데이터 플레인을 그래프로 나타낸것입니다.

출처 : kfserving

구조

Endpoint: “Default”와 “Canary” 엔드포인트를 가집니다. 이 엔드포인트 덕분에 사용자는 카나리아 배포 전략을 사용해서 모델을 안전한게 변경 할 수 있습니다. 구 버전의 모델 서버와 새 버전의 모델 서버들을 구성하고 일부 트래픽을 새 버전으로 분산하여 오류 여부를 판단합니다. 분산 후 결과에 따라 새 버전이 운영 환경을 대체할 수도 있고, 다시 구 버전으로 돌아갈 수도 있습니다.

Component: 각 엔드 포인트는 “예측자(predictor)”, “설명자(explainer)”및 “변환기(transformer)”와 같은 여러 컴포넌트로 구성됩니다. 꼭 필요한 컴포넌트는 시스템의 핵심인 “예측자(predictor)”입니다. KFServing은 Outlier Detection과 같은 사용 사례를 지원하기 위해 지원되는 컴포넌트의 수를 늘릴 계획을 가지고 있습니다.

  • Predictor: Predictor는 InferenceService의 핵심 컴포넌트입니다. 네트워크 엔드포인트에서 사용 가능하게하는 모델 및 모델 서버입니다.
  • Explainer: Explainer는 선택 가능한 컴포넌트로서 모델이 어떻게 예측을 했는지에 대한 설명을 제공합니다. 사용자는 자신들이 가지고 있는 자체 설명 컨테이너를 정의할 수 있습니다. 일반적인 사용 사례의 경우 KFServing은 Alibi와 같은 기본 Explainer를 제공합니다.
  • Transformer: Transformer는 예측 및 설명 워크 플로우 전에 사전 및 사후 처리 단계를 정의 할 수 있는 컴포넌트 입니다. Explainer 과 마찬가지로 관련 환경 변수로 구성됩니다. 일반적인 사용 사례의 경우 KFServing은 Feast와 같은 기본 Transformer를 제공합니다.

Data Plane (V1)

KFServing에는 제공하는 모든 모델 서버에는 표준화 된 API를 지원합니다.

데이터 플레인 프로토콜의 V1 에서는 다음과 같은 HTTP/REST API를 제공하고 있습니다.

APIVERBPATH
ListGET/v1/models
ReadGET/v1/models/
PredictPOST/v1/models/:predict
ExplainPOST/v1/models/:explain

Predict

모든 InferenceServices는 Tensorflow V1 HTTP API (https://www.tensorflow.org/tfx/serving/api_rest#predict_api)와 호환되는 API를 사용합니다.

URL

POST <http://host>:port/v1/models/${MODEL_NAME}:predict

Request format

예측 API의 요청 본문은 다음과 같은 형식의 JSON 객체여야 합니다.

{
  "instances": <value>|<(nested)list>|<list-of-objects>
}

예측을 요청할 데이터의 값은 JSON 객체의 instances 필드에 입력합니다.

{
  // List of 3 scalar tensors.
  "instances": [ "foo", "bar", "baz" ]
}

{
  // List of 2 tensors each of [1, 2] shape
  "instances": [ [[1, 2]], [[3, 4]] ]
}

Response format

예측을 요청하면, 예측 결과를 응답합니다. 응답 본문에는 JSON 객체가 포함되어 있습니다. 행 형식의 요청에는 다음과 같은 형식의 응답이 있습니다.

{
  "predictions": <value>|<(nested)list>|<list-of-objects>
}

Explain

Explainer와 함께 배치 된 모든 InferenceService는 표준화 된 설명 API를 지원합니다. 이 인터페이스는 “: explain”동사가 추가 된 Tensorflow V1 HTTP API와 동일합니다.

URL

POST <http://host>:port/v1/models/${MODEL_NAME}:explain

Request format

예측 API의 요청 본문은 다음과 같은 형식의 JSON 객체여야 합니다.

{
  "instances": <value>|<(nested)list>|<list-of-objects>
}

예측을 요청할 데이터의 값은 JSON 객체의 instances 필드에 입력합니다.

{
  // List of 3 scalar tensors.
  "instances": [ "foo", "bar", "baz" ]
}

{
  // List of 2 tensors each of [1, 2] shape
  "instances": [ [[1, 2]], [[3, 4]] ]
}

Response format

예측을 요청하면, 예측 결과를 응답합니다. 응답 본문에는 JSON 객체가 포함되어 있습니다. 행 형식의 요청에는 다음과 같은 형식의 응답이 있습니다.

{
  "predictions": <value>|<(nested)list>|<list-of-objects>,
	"explanations": <value>|<(nested)list>|<list-of-objects>
}

Data Plane (V2)

데이터 플레인 프로토콜의 두 번째 버전은 V1 데이터 플레인 프로토콜에서 발견 된 여러 가지 문제를 해결하기 위해서 만들어지고 있습니다. 여기에는 수많은 모델 프레임워크의 일반화와 서버 성능 문제등을 포함하고 있습니다.

Predict

V2 프로토콜은 HTTP/REST 및 GRPC API를 모두 제안하고 있습니다. 자세한 내용은 전체 제안서를 참조하십시오.

Kubeflow Pipelines – 경량 파이썬 컴포넌트에서의 시각화 사용하기

경량 컴포넌트에서 시각화를 사용할 경우 구문이 약간 다릅니다.

텐서플로우의 mnist 이미지 식별 모델 학습을 이용하여 파이프라인 메트릭과 출력 뷰어의 한 종류인 텐서보드를 사용해 보도록 하겠습니다.

파이썬 함수를 시각화를 할 경우 typing.NamedTuple 타입 힌트를 사용하여 메타데이터와 메트릭을 선언하여야합니다.

def train(tb_log_dir: str) -> NamedTuple('Outputs', [('mlpipeline_ui_metadata', 'ui_metadata'),
                                                     ('mlpipeline_metrics', 'metrics')]):

그리고 함수의 마지막 부분에서 메타데이터와 메트릭의 값을 JSON 형식으로 변환하여 반환해야 합니다.

    from collections import namedtuple
    outputs = namedtuple('Outputs', ['mlpipeline_ui_metadata', 'mlpipeline_metrics'])
    return outputs(json.dumps(metadata), json.dumps(metrics))

다음은 경량 컴포넌트에서 시각화를 사용할 전체 코드 입니다.

tensorboard_pipeline.py

from typing import NamedTuple

import kfp
from kubernetes.client.models import V1EnvVar


def train(tb_log_dir: str) -> NamedTuple('Outputs', [('mlpipeline_ui_metadata', 'ui_metadata'),
                                                     ('mlpipeline_metrics', 'metrics')]):
    import tensorflow as tf
    import json

    print("TensorFlow version: ", tf.__version__)

    mnist = tf.keras.datasets.mnist

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

    model = tf.keras.models.Sequential([
        tf.keras.layers.Flatten(input_shape=(28, 28)),
        tf.keras.layers.Dense(128, activation='relu'),
        tf.keras.layers.Dropout(0.2),
        tf.keras.layers.Dense(10, activation='softmax')
    ])

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

    model.summary()

    callbacks = [tf.keras.callbacks.TensorBoard(log_dir=tb_log_dir)]

    print("Training...")
    model.fit(x_train, y_train, epochs=5, validation_split=0.2, callbacks=callbacks)

    score = model.evaluate(x_test, y_test, batch_size=128)
    print('Test accuracy: ', score[1])

    metadata = {
        'outputs': [{
            'type': 'tensorboard',
            'source': tb_log_dir,
        }]
    }

    loss = score[0]
    accuracy = score[1]
    metrics = {
        'metrics': [{
            'name': 'accuracy',
            'numberValue': float(accuracy),
            'format': "PERCENTAGE",
        }, {
            'name': 'loss',
            'numberValue': float(loss),
            'format': "RAW",
        }]
    }

    from collections import namedtuple
    outputs = namedtuple('Outputs', ['mlpipeline_ui_metadata', 'mlpipeline_metrics'])
    return outputs(json.dumps(metadata), json.dumps(metrics))


train_op = kfp.components.func_to_container_op(train, base_image='tensorflow/tensorflow:2.1.0-py3')


def lightweight_tensorboard_pipeline(tb_log_dir):
    s3_endpoint = 'minio-service.kubeflow.svc.cluster.local:9000'
    minio_endpoint = "http://" + s3_endpoint
    minio_username = "minio"
    minio_key = "minio123"
    minio_region = "us-east-1"

    train_op(tb_log_dir).add_env_variable(V1EnvVar(name='S3_ENDPOINT', value=s3_endpoint)) \\
        .add_env_variable(V1EnvVar(name='AWS_ENDPOINT_URL', value=minio_endpoint)) \\
        .add_env_variable(V1EnvVar(name='AWS_ACCESS_KEY_ID', value=minio_username)) \\
        .add_env_variable(V1EnvVar(name='AWS_SECRET_ACCESS_KEY', value=minio_key)) \\
        .add_env_variable(V1EnvVar(name='AWS_REGION', value=minio_region)) \\
        .add_env_variable(V1EnvVar(name='S3_USE_HTTPS', value='0')) \\
        .add_env_variable(V1EnvVar(name='S3_VERIFY_SSL', value='0'))


if __name__ == '__main__':
    arguments = {'tb_log_dir': 's3://tensorboard/lightweight'}
    my_run = kfp.Client().create_run_from_pipeline_func(lightweight_tensorboard_pipeline,
                                                        arguments=arguments,
                                                        experiment_name='Sample Experiment')

다음은 Kubeflow 파이프 라인 UI의 Run Output 화면입니다.

Kubeflow Pipelines – 파이프라인에서 외부 저장소를 이용하기

파이프라인 컴포넌트에서 PVC 사용하기

파이프라인 컴포넌트에서 모델을 학습하고 PV에 저장하려면, PVC를 사용하면 됩니다. PV이 마운트 되면, 내부 파일시스템처럼 접근하여 데이터를 저장하고 읽을 수 있습니다.

PVC 를 생성하기

PV를 사용하기 위하여, 쿠버네티스 리소스인 PVC 를 생성해 보겠습니다.

다음은 100Mi의 용량을 가지는 PersistentVolumeClaim 매니페스트입니다.

kfp-pvc.yaml

kind: PersistentVolumeClaim
apiVersion: v1
metadata:
  name: kfp-pvc
spec:
  accessModes:
    - ReadWriteOnce
  resources:
    requests:
      storage: 100Mi

kubectl 을 사용하여 kubeflow 네임스페이스 PVC 를 생성하겠습니다.

kubectl -n kubeflow apply -f kfp-pvc.yaml

텐서플로우 mnist 모델을 학습하고 S3에 저장하기

mnist 모델을 S3에 저장하겠습니다. 코드는 앞서 사용한 Tensorflow 코드와 동일하기 때문에 자세한 설명은 생략하겠습니다.

tensorflow_mnist.py

from __future__ import absolute_import, division, print_function, unicode_literals

import argparse
import os

import tensorflow as tf


def train():
    print("TensorFlow version: ", tf.__version__)

    parser = argparse.ArgumentParser()
    parser.add_argument('--model_path', default='./model', type=str)
    args = parser.parse_args()

    version = 1
    export_path = os.path.join(args.model_path, str(version))

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

    model = tf.keras.models.Sequential([
        tf.keras.layers.Flatten(input_shape=(28, 28)),
        tf.keras.layers.Dense(128, activation='relu'),
        tf.keras.layers.Dropout(0.2),
        tf.keras.layers.Dense(10, activation='softmax')
    ])

    model.compile(optimizer=tf.keras.optimizers.SGD(learning_rate=0.01),
                  loss='sparse_categorical_crossentropy',
                  metrics=['accuracy'])

    print("Training...")
    training_history = model.fit(x_train, y_train, batch_size=64, epochs=10,
                                 validation_split=0.2)

    print('\\nEvaluate on test data')
    results = model.evaluate(x_test, y_test, batch_size=128)
    print('test loss, test acc:', results)

    model.save(export_path)
    print('"Saved model to {}'.format(export_path))


if __name__ == '__main__':
    train()

컨테이너 이미지를 만들기

컨테이너 이미지를 만들기 위한 Dockerfile 입니다. 텐서플로우를 기본 이미지로 사용합니다.

Dockerfile

FROM tensorflow/tensorflow:2.1.0-py3

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

컨테이너 이미지를 빌드하겠습니다.

docker build -t kangwoo/kfp-mnist-storage:0.0.1 .

빌드한 컨테이너 이미지를 컨테이너 이미지 레지스트리에 업로드 합니다.

docker push kangwoo/kfp-mnist-storage:0.0.1

컴포넌트 작성

Kubeflow Pipelines DSL을 사용하여 컴포넌트를 작성합니다. 컴포넌트에서 사용하는 컨테이너 이미지를 정의합니다. 텐서플로우가 실행되는 컴포트넌트에서 PV에 접근할 수 있도록, PVC를 마운트 합니다.

pvc_name = "kfp-pvc"
    volume_name = 'pipeline'
    volume_mount_path = '/mnt/pipeline'

    dsl.ContainerOp(
        name='mnist_pvc',
        image='kangwoo/kfp-mnist-storage:0.0.1',
        arguments=['--model', '/mnt/pipeline/kfp/mnist/model']
    ).add_volume(V1Volume(name=volume_name, persistent_volume_claim=V1PersistentVolumeClaimVolumeSource(claim_name=pvc_name))) \\
        .add_volume_mount(V1VolumeMount(mount_path=volume_mount_path, name=volume_name))

파이프라인 작성

Kubeflow Pipelines DSL을 사용하여 파이프 라인 함수를 작성합니다. 파이프 라인을 정의하고 사용하는 컴포넌트들을 추가합니다. Kubeflow Pipelines SDK 를 사용하여 파이프라인을 빌드 한 후, 업로드하고 실행합니다.

import kfp
from kfp import dsl

from kubernetes.client.models import V1PersistentVolumeClaimVolumeSource, \\
    V1Volume, V1VolumeMount


def pipeline_pvc():
    pvc_name = "kfp-pvc"
    volume_name = 'pipeline'
    volume_mount_path = '/mnt/pipeline'

    dsl.ContainerOp(
        name='mnist_pvc',
        image='kangwoo/kfp-mnist-storage:0.0.1',
        arguments=['--model', '/mnt/pipeline/kfp/mnist/model']
    ).add_volume(V1Volume(name=volume_name, persistent_volume_claim=V1PersistentVolumeClaimVolumeSource(claim_name=pvc_name))) \\
        .add_volume_mount(V1VolumeMount(mount_path=volume_mount_path, name=volume_name))


if __name__ == '__main__':
    my_run = kfp.Client().create_run_from_pipeline_func(pipeline_pvc, arguments={},
                                                        experiment_name='Sample Experiment')

파이프라인을 실행하면 학습된 모델이 지정한 위치에 저장됩니다.

onprem.mount_pvc() 사용하기

앞 예제에서는 PVC를 직접 마운트해 주었습니다. kfp 에서 제공하는 onprem.mount_pvc() 메소드를 사용하면 보다 간단히 사용할 수 있습니다.

다음은 onprem.mount_pvc() 를 사용하는 컴포넌트 예제입니다.

import kfp
from kfp import dsl
from kfp import onprem


def pipeline_pmount_pvc():
    pvc_name = "kfp-pvc"
    volume_name = 'pipeline'
    volume_mount_path = '/mnt/pipeline'

    dsl.ContainerOp(
        name='mnist_mount_pvc',
        image='kangwoo/kfp-mnist-storage:0.0.1',
        arguments=['--model', '/mnt/pipeline/kfp/mnist/model']
    ).apply(onprem.mount_pvc(pvc_name, volume_name=volume_name, volume_mount_path=volume_mount_path))


if __name__ == '__main__':
    my_run = kfp.Client().create_run_from_pipeline_func(pipeline_pmount_pvc, arguments={},
                                                        experiment_name='Sample Experiment')

파이프라인 컴포넌트에서 S3 사용하기

파이프라인 컴포넌트에서 모델을 학습하고 S3에 저장하려면, S3에 접속할 수 있는 정보가 필요합니다. 환경 변수를 이용하여 접속 정보를 컴포넌트에 전달할 수 있습니다.

S3 Secret 을 생성하기

S3에 접속하여 데이터를 저장하거나 가져오려면 접속 정보가 필요합니다. 중요한 정보인 AWS_ACCESS_KEY_IDAWS_SECRET_ACCESS_KEY 는 쿠버네티스트 Secret 리소스에 저장하겠습니다.

다음은 Kubeflow에서 제공하는 minio를 s3 저장소로 사용하는 Secret 매니페스트 입니다. data 섹션에 있는 AWS_ACCESS_KEY_IDAWS_SECRET_ACCESS_KEY필드의 값은 BASE64 로 인코딩 된 값을 사용해야합니다.

export AWS_ACCESS_KEY_ID=minio
export AWS_SECRET_ACCESS_KEY=minio123

kubectl -n kubeflow create secret generic kfp-aws-secret \\
    --from-literal=AWS_ACCESS_KEY_ID=${AWS_ACCESS_KEY_ID} \\
    --from-literal=AWS_SECRET_ACCESS_KEY=${AWS_SECRET_ACCESS_KEY}

생성한 kfp-aws-secret 을 조회해 보겠습니다.

kubectl -n kubeflow get secret kfp-aws-secret

정상적으로 조회되면, 다음과 같은 결과를 확인할 수 있습니다.

apiVersion: v1
kind: Secret
metadata:
  name: kfp-aws-secret
type: Opaque
data:
  AWS_ACCESS_KEY_ID: bWluaW8=
  AWS_SECRET_ACCESS_KEY: bWluaW8xMjM=

텐서플로우 mnist 모델을 학습하고 S3에 저장하기

mnist 모델을 S3에 저장하겠습니다. 코드와 컨테이너 이미지는 앞서 사용한 PVC 저장 코드와 동일하기 때문에, “파이프라인 작성” 단계로 건너 뛰어도 됩니다.

tensorflow_mnist.py

from __future__ import absolute_import, division, print_function, unicode_literals

import argparse
import os

import tensorflow as tf


def train():
    print("TensorFlow version: ", tf.__version__)

    parser = argparse.ArgumentParser()
    parser.add_argument('--model_path', default='./model', type=str)
    args = parser.parse_args()

    version = 1
    export_path = os.path.join(args.model_path, str(version))

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

    model = tf.keras.models.Sequential([
        tf.keras.layers.Flatten(input_shape=(28, 28)),
        tf.keras.layers.Dense(128, activation='relu'),
        tf.keras.layers.Dropout(0.2),
        tf.keras.layers.Dense(10, activation='softmax')
    ])

    model.compile(optimizer=tf.keras.optimizers.SGD(learning_rate=0.01),
                  loss='sparse_categorical_crossentropy',
                  metrics=['accuracy'])

    print("Training...")
    training_history = model.fit(x_train, y_train, batch_size=64, epochs=10,
                                 validation_split=0.2)

    print('\\nEvaluate on test data')
    results = model.evaluate(x_test, y_test, batch_size=128)
    print('test loss, test acc:', results)

    model.save(export_path)
    print('"Saved model to {}'.format(export_path))


if __name__ == '__main__':
    train()

컨테이너 이미지를 만들기

컨테이너 이미지를 만들기 위한 Dockerfile 입니다. 텐서플로우를 기본 이미지로 사용합니다.

Dockerfile

FROM tensorflow/tensorflow:2.1.0-py3

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

컨테이너 이미지를 빌드하겠습니다.

docker build -t kangwoo/kfp-mnist-storage:0.0.1 .

빌드한 컨테이너 이미지를 컨테이너 이미지 레지스트리에 업로드 합니다.

docker push kangwoo/kfp-mnist-storage:0.0.1

컴포넌트 작성

Kubeflow Pipelines DSL을 사용하여 컴포넌트를 작성합니다. 컴포넌트에서 사용하는 컨테이너 이미지를 정의합니다. 텐서플로우가 실행되는 컴포트넌트에서 s3에 접근할 수 있도록, s3 설정 정보를 환경 변수로 넘겨 주었습니다.

		secret_name = "kfp-aws-secret"

    s3_endpoint = 'minio-service.kubeflow.svc.cluster.local:9000'
    minio_endpoint = "http://" + s3_endpoint
    minio_region = "us-east-1"

    dsl.ContainerOp(
        name='tensorboard',
        image='kangwoo/kfp-mnist-s3:0.0.1',
        arguments=['--model', 's3://tensorflow/kfp/mnist/model']
    ).add_env_variable(V1EnvVar(name='S3_ENDPOINT', value=s3_endpoint)) \\
        .add_env_variable(V1EnvVar(name='AWS_ENDPOINT_URL', value=minio_endpoint)) \\
        .add_env_variable(V1EnvVar(name='AWS_ACCESS_KEY_ID',
                                   value_from=V1EnvVarSource(
                                       secret_key_ref=V1SecretKeySelector(name=secret_name, key='AWS_ACCESS_KEY_ID')))) \\
        .add_env_variable(V1EnvVar(name='AWS_SECRET_ACCESS_KEY',
                                   value_from=V1EnvVarSource(secret_key_ref=V1SecretKeySelector(name=secret_name,
                                                                                                key='AWS_SECRET_ACCESS_KEY')))) \\
        .add_env_variable(V1EnvVar(name='AWS_REGION', value=minio_region)) \\
        .add_env_variable(V1EnvVar(name='S3_USE_HTTPS', value='0')) \\
        .add_env_variable(V1EnvVar(name='S3_VERIFY_SSL', value='0'))

파이프라인 작성

Kubeflow Pipelines DSL을 사용하여 파이프 라인 함수를 작성합니다. 파이프 라인을 정의하고 사용하는 컴포넌트들을 추가합니다. Kubeflow Pipelines SDK 를 사용하여 파이프라인을 빌드 한 후, 업로드하고 실행합니다.

import kfp
from kfp import dsl

from kubernetes.client.models import V1EnvVar, V1EnvVarSource, V1SecretKeySelector


def pipeline_s3():
    secret_name = "kfp-aws-secret"

    s3_endpoint = 'minio-service.kubeflow.svc.cluster.local:9000'
    minio_endpoint = "http://" + s3_endpoint
    minio_region = "us-east-1"

    dsl.ContainerOp(
        name='mnist-s3',
        image='kangwoo/kfp-mnist-storage:0.0.1',
        arguments=['--model', 's3://tensorflow/kfp/mnist/model']
    ).add_env_variable(V1EnvVar(name='S3_ENDPOINT', value=s3_endpoint)) \\
        .add_env_variable(V1EnvVar(name='AWS_ENDPOINT_URL', value=minio_endpoint)) \\
        .add_env_variable(V1EnvVar(name='AWS_ACCESS_KEY_ID',
                                   value_from=V1EnvVarSource(
                                       secret_key_ref=V1SecretKeySelector(name=secret_name, key='AWS_ACCESS_KEY_ID')))) \\
        .add_env_variable(V1EnvVar(name='AWS_SECRET_ACCESS_KEY',
                                   value_from=V1EnvVarSource(secret_key_ref=V1SecretKeySelector(name=secret_name,
                                                                                                key='AWS_SECRET_ACCESS_KEY')))) \\
        .add_env_variable(V1EnvVar(name='AWS_REGION', value=minio_region)) \\
        .add_env_variable(V1EnvVar(name='S3_USE_HTTPS', value='0')) \\
        .add_env_variable(V1EnvVar(name='S3_VERIFY_SSL', value='0'))


if __name__ == '__main__':
    my_run = kfp.Client().create_run_from_pipeline_func(pipeline_s3, arguments={},
                                                        experiment_name='Sample Experiment')

파이프라인을 실행하면 학습된 모델이 지정한 위치에 저장됩니다.

aws.use_aws_secret() 사용하기

앞 예제에서는 환경 변수에 직접 값들을 설정해 주었습니다. kfp 에서 제공하는 aws.use_aws_secret() 메소드를 사용하면 보다 간단히 인증 정보를 설정할 수 있습니다. 하지만 현재 버전에서는 AWS_ACCESS_KEY_IDAWS_SECRET_ACCESS_KEY 값 만을 변경할 수 있기 때문에 aws 에서 제공하는 s3만 사용할 수 있습니다. 다시 말해서 minio는 사용할 수 없습니다.

다음은 aws.use_aws_secret() 를 사용하는 컴포넌트 예제입니다.

import kfp
from kfp import aws
from kfp import dsl


def pipeline_use_aws_secret():
    secret_name = "kfp-aws-secret"

    dsl.ContainerOp(
        name='mnist_use_aws_secret',
        image='kangwoo/kfp-mnist-storage:0.0.1',
        arguments=['--model', 's3://tensorflow/kfp/mnist/model']
    ).apply(aws.use_aws_secret(secret_name,
                               aws_access_key_id_name='AWS_ACCESS_KEY_ID',
                               aws_secret_access_key_name='AWS_SECRET_ACCESS_KEY'))


if __name__ == '__main__':
    my_run = kfp.Client().create_run_from_pipeline_func(pipeline_use_aws_secret, arguments={},
                                                        experiment_name='Sample Experiment')

파이프라인 컴포넌트에서 GCS 사용하기

파이프라인 컴포넌트에서 모델을 학습하고 GCS에 저장하려면, GCS에 접속할 수 있는 정보가 필요합니다. 환경 변수를 이용하여 접속 정보를 컴포넌트에 전달할 수 있습니다.

GCS Secret 을 생성하고 서비스 계정에 추가하기

컴포넌트에서 GCS에 접속하여 데이터를 저장하거나 가져오려면 접속 정보가 필요합니다.

구글 클라우드의 JSON 서비스 계정 키를 다운로드 합니다. 이 서비스 계정은 사용할 GCS에 접근할 권한이 부여되어 있어야합니다.

다음은 서비스 계정 키를 gcp-sa-credentials.json 파일로 다운로드 하는 명령어 입니다. [SA-NAME] 에는 서비스 계정의 아이디를, [PROJECT-ID] 에는 프로젝트 아이디를 입력하면 됩니다.

gcloud iam service-accounts keys create gcp-sa-credentials.json \\
  --iam-account [SA-NAME]@[PROJECT-ID].iam.gserviceaccount.com

다운로드 받은 구글 클라우드의 서비스 계정 키를, 쿠버네티스 Secret 에 등록합니다.

다음은 kubeflow 네임스페이스 Secret 을 생성하는 명령어 입니다.

kubectl -n kubeflow create secret generic kfp-gcp-sa \\
  --from-file=gcloud-application-credentials.json=gcp-sa-credentials.json

텐서플로우 mnist 모델을 학습하고 GCS에 저장하기

mnist 모델을 GCS에 저장하겠습니다. 코드와 컨테이너 이미지는 앞서 사용한 S3 저장 코드와 동일하기 때문에, “파이프라인 작성” 단계로 건너 뛰어도 됩니다.

tensorflow_mnist.py

from __future__ import absolute_import, division, print_function, unicode_literals

import argparse
import os

import tensorflow as tf


def train():
    print("TensorFlow version: ", tf.__version__)

    parser = argparse.ArgumentParser()
    parser.add_argument('--model_path', default='./model', type=str)
    args = parser.parse_args()

    version = 1
    export_path = os.path.join(args.model_path, str(version))

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

    model = tf.keras.models.Sequential([
        tf.keras.layers.Flatten(input_shape=(28, 28)),
        tf.keras.layers.Dense(128, activation='relu'),
        tf.keras.layers.Dropout(0.2),
        tf.keras.layers.Dense(10, activation='softmax')
    ])

    model.compile(optimizer=tf.keras.optimizers.SGD(learning_rate=0.01),
                  loss='sparse_categorical_crossentropy',
                  metrics=['accuracy'])

    print("Training...")
    training_history = model.fit(x_train, y_train, batch_size=64, epochs=10,
                                 validation_split=0.2)

    print('\\nEvaluate on test data')
    results = model.evaluate(x_test, y_test, batch_size=128)
    print('test loss, test acc:', results)

    model.save(export_path)
    print('"Saved model to {}'.format(export_path))


if __name__ == '__main__':
    train()

컨테이너 이미지를 만들기

컨테이너 이미지를 만들기 위한 Dockerfile 입니다. 텐서플로우를 기본 이미지로 사용합니다.

Dockerfile

FROM tensorflow/tensorflow:2.1.0-py3

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

컨테이너 이미지를 빌드하겠습니다.

docker build -t kangwoo/kfp-mnist-storage:0.0.1 .

빌드한 컨테이너 이미지를 컨테이너 이미지 레지스트리에 업로드 합니다.

docker push kangwoo/kfp-mnist-storage:0.0.1

컴포넌트 작성

Kubeflow Pipelines DSL을 사용하여 컴포넌트를 작성합니다. 컴포넌트에서 사용하는 컨테이너 이미지를 정의합니다. 텐서플로우가 실행되는 컴포트넌트에서 gcs에 접근할 수 있도록, gcs 설정 정보를 환경 변수로 넘겨 주었습니다. kfp-gcp-sa 를 볼륨 마운트 한 다음, GOOGLE_APPLICATION_CREDENTIALS 라는 환경 변수에 인증 정보가 들어 있는 json 파일의 위치를 설정해 줍니다.

		GCSCredentialFileName = "gcloud-application-credentials.json"
    GCSCredentialVolumeName = "user-gcp-sa"
    GCSCredentialVolumeMountPath = "/var/secrets/"
    GCSCredentialEnvKey = "GOOGLE_APPLICATION_CREDENTIALS"
    GCSCredentialFilePath = os.path.join(GCSCredentialVolumeMountPath, GCSCredentialFileName)

    secret_name = 'kfp-gcp-sa'

    dsl.ContainerOp(
        name='mnist-gcs',
        image='kangwoo/kfp-mnist-storage:0.0.1',
        arguments=['--model', 'gcs://kfp-bucket/kfp/mnist/model']
    ).add_volume(V1Volume(name=GCSCredentialVolumeName, secret=V1SecretVolumeSource(secret_name=secret_name))) \\
        .add_volume_mount(V1VolumeMount(name=GCSCredentialVolumeName, mount_path=GCSCredentialVolumeMountPath)) \\
        .add_env_variable(V1EnvVar(name=GCSCredentialEnvKey, value=GCSCredentialFilePath))

파이프라인 작성

Kubeflow Pipelines DSL을 사용하여 파이프 라인 함수를 작성합니다. 파이프 라인을 정의하고 사용하는 컴포넌트들을 추가합니다. Kubeflow Pipelines SDK 를 사용하여 파이프라인을 빌드 한 후, 업로드하고 실행합니다.

import os

import kfp
from kfp import dsl
from kubernetes.client.models import V1EnvVar, V1VolumeMount, V1Volume, \\
    V1SecretVolumeSource


def pipeline_gcs():
    GCSCredentialFileName = "gcloud-application-credentials.json"
    GCSCredentialVolumeName = "user-gcp-sa"
    GCSCredentialVolumeMountPath = "/var/secrets/"
    GCSCredentialEnvKey = "GOOGLE_APPLICATION_CREDENTIALS"
    GCSCredentialFilePath = os.path.join(GCSCredentialVolumeMountPath, GCSCredentialFileName)

    secret_name = 'kfp-gcp-sa'

    dsl.ContainerOp(
        name='mnist-gcs',
        image='kangwoo/kfp-mnist-storage:0.0.1',
        arguments=['--model', 'gs://kfp-bucket/kfp/mnist/model']
    ).add_volume(V1Volume(name=GCSCredentialVolumeName, secret=V1SecretVolumeSource(secret_name=secret_name))) \\
        .add_volume_mount(V1VolumeMount(name=GCSCredentialVolumeName, mount_path=GCSCredentialVolumeMountPath)) \\
        .add_env_variable(V1EnvVar(name=GCSCredentialEnvKey, value=GCSCredentialFilePath))


if __name__ == '__main__':
    my_run = kfp.Client().create_run_from_pipeline_func(pipeline_gcs, arguments={},
                                                        experiment_name='Sample Experiment')

gcp.use_gcp_secret() 사용하기

앞 예제에서는 환경 변수에 직접 값들을 설정해 주었습니다. kfp 에서 제공하는 gcp.use_gcp_secret() 메소드를 사용하면 보다 간단히 인증 정보를 설정할 수 있습니다.

다음은 gcp.use_gcp_secret() 를 사용하는 컴포넌트 예제입니다.

import kfp
from kfp import dsl
from kfp import gcp


def pipeline_use_gcp_secret():
    secret_name = 'kfp-gcp-sa'
    secret_file_path_in_volume = '/gcloud-application-credentials.json'

    dsl.ContainerOp(
        name='mnist_use_gcp_secret',
        image='kangwoo/kfp-mnist-storage:0.0.1',
        arguments=['--model', 'gs://kfp-bucket/kfp/mnist/model']
    ).apply(gcp.use_gcp_secret(secret_name, secret_file_path_in_volume=secret_file_path_in_volume))


if __name__ == '__main__':
    my_run = kfp.Client().create_run_from_pipeline_func(pipeline_use_gcp_secret, arguments={},
                                                        experiment_name='Sample Experiment')

한꺼번에 외부 저장소 추가하기

파이프라인의 단계가 많아지면 일일히 외부 저장소를 추가해 주는 작업이 번거로울 수 있습니다. 그럴 경우에는 다음 예제처럼 반복문을 이용해서 저장소를 추가해 줄 수 있습니다.

ingest = dsl.ContainerOp(...)
transformation = dsl.ContainerOp(...)
train = dsl.ContainerOp(...)

steps = [ingest, transformation, train]
for step in steps:
	step.apply(oprem.mount_pvc(pvc_name, 'local-storage', '/mnt'))

Kubeflow Pipelines – 파이프 라인 UI에서 결과 시각화

소개

Kubeflow Pipelines UI는 기본적으로 여러 유형의 시각화를 제공하고 있습니다. 그래서 성능 평가 및 비교 데이터를 보여주기 위한 방법으로 사용할 수 있습니다. 이 시각화 기능은 출력 뷰어를 사용하고 있습니다. 컴포넌트의 결과물을 출력 뷰어를 이용하여 시각화 할 수 있는 것입니다. 출력 뷰어를 사용하려면, 파이프 라인 컴포넌트의 애플리케이션에서 JSON 파일을 로컬 파일 시스템에 저장하면 됩니다.

시각화된 출력 결과는 Kubeflow 파이프 라인 UI를 통해서 확인 할 수 있습니다. Artifacts 페이지와 , Run output 페이지에서 시각화된 출력 결과를 조회 할 수 있습니다.

Artifacts

Artifacts 탭에는 선택한 파이프 라인 단계의 시각화가 표시됩니다.

Kubeflow 파이프 라인 UI에서 Artifacts 탭을 열려면, 다음 절차대로 진행하시면 됩니다.

  1. Experiments 클릭하여 현재 파이프 라인 실험 목록을 조회합니다.
  2. 보려는 실험의 ‘실험 이름’을 클릭하십시오.
  3. 보려는 실행의 “실행 이름”을 클릭하십시오.
  4. Graph 탭에서 보려는 파이프 라인 컴포넌트를 나타내는 단계를 클릭하십시오. 세부 사항이 Artifacts  탭을 표시하며 슬라이드 됩니다.

Run output

Run output 탭에는 선택한 실행의 모든 ​​단계에 대한 시각화가 표시됩니다.

Kubeflow 파이프 라인 UI에서 Run output 탭을 열려면, 다음 절차대로 진행하시면 됩니다.

  1. Experiments 클릭하여 현재 파이프 라인 실험 목록을 조회합니다.
  2. 보려는 실험의 ‘실험 이름’을 클릭하십시오.
  3. 보려는 실행의 “실행 이름”을 클릭하십시오.
  4. Run output 탭을 클릭하십시오.

출력 뷰어를 위한 메타데이터 저장하기

파이프 라인 컴포넌트의 결과물을 출력 뷰어를 통해서 시각화 할 수 있습니다. 출력 뷰어를 통해서 결과물을 시각화 하려면, JSON 형태의 메타 데이터 파일을 로컬 저장소에 작성해야 합니다. 파일 이름은 /mlpipeline-ui-metadata.json 을 사용합니다.

JSON은 outputs 배열을 가지고 있습니다. 각 outputs 항목은 출력 뷰어의 메타 데이터를 설명합니다. JSON 구조는 다음과 같습니다.

{
  "version": 1,
  "outputs": [
    {
      "type": "confusion_matrix",
      "format": "csv",
      "source": "my-dir/my-matrix.csv",
      "schema": [
        {"name": "target", "type": "CATEGORY"},
        {"name": "predicted", "type": "CATEGORY"},
        {"name": "count", "type": "NUMBER"},
      ],
      "labels": "vocab"
    },
    {
      ...
    }
  ]
}

출력 뷰어의 메타 데이터에는 생성할 뷰어와 사용할 데이터에 대한 정의가 포함되어 있습니다.

Kubeflow Pipilines 시스템은 컴포넌트가 컨테이너 파일 시스템에 저장한 메타 데이터 파일을 읽어와서 Kubeflow Pipelines UI에서 지정한 뷰어를 생성합니다. Kubeflow Pipelines UI는 데이터를 메모리에 로드하여 렌더링합니다.

컴포넌트가 해당 메타데이터 파일을 컨테이너 파일 시스템에 쓰는 경우 Kubeflow Pipelines 시스템은 파일을 자동으로 Artifact 저장소에 저장합니다. Kubeflow Pipelines UI는 이 파일을 사용하여 지정된 뷰어를 생성합니다. 메타 데이터에는 Kubeflow Pipelines UI 가 아티팩트 데이터를 가져오기 위한 위치가 정의되어 있습니다. Kubeflow Pipelines UI는 데이터를 메모리에 가져와서 렌더링합니다.

주의 : Kubeflow Pipilines 0.3.0 버전에서 지원하는 아티팩터 데이터의 사용 가능한 위치는 Google Cloud Storage, Amazon S3, http https 입니다. 이슈 데이터를 정상적으로 가져오기 위해서는 Kubeflow Pipelines UI 에서 해당 아티팩터 데이터 위치로 접근할 수 있어야합니다. 그래서 설치 환경에 따라서 별도의 접근 권한을 Kubeflow Pipelines UI 에 부여해줘야 할 수도 있습니다.

다음 표는 outputs 배열에서 지정할 수 있는 메타 데이터 필드입니다. 각 output 항목에는 type이 있어야합니다. type 값에 따라 사용하는 필드 값은 약간 다를 수 있습니다.나중에 페이지의 출력 뷰어 목록에 설명 된대로 다른 필드가 필요할 수도 있습니다.

outputs

필드 이름설명
format아티팩트 데이터의 형식입니다. 기본값은 csv입니다. (현재 사용 가능한 유일한 형식은 csv입니다.)
header아티팩터 데이터의 헤더로 사용될 문자열 목록입니다. 예를 들어, 테이블에서 이 문자열은 첫 번째 행에서 사용됩니다.
labels이슈 열 또는 행의 레이블로 사용되는 문자열 목록입니다.
predicted_col예측 열의 이름입니다.
schema아티팩트 데이터의 스키마를 지정하는 {type, name} 객체의 목록입니다.
source데이터의 전체 경로입니다. 사용 가능한 위치에는 http, https, Amazon S3 및 Google Cloud Storage가 포함됩니다. 경로에는 와일드 카드 ‘*’가 포함될 수 있으며,이 경우 Kubeflow Pipelines UI는 일치하는 소스 파일의 데이터를 연결합니다. source는 인라인 문자열 데이터도 포함 할 수 있습니다. storage 이 inline일 때 경로 대신 문자열 데이터를 포함합니다.
storagestorage 가 inline 인 경우 source값이 위치 대신 인라인 데이터로 사용 됩니다. 이것은 텐서 보드를 제외한 모든 유형의 출력에 적용됩니다.
target_col대상 열의 이름입니다.
type데이터를 시각화하는 데 사용되는 뷰어의 이름입니다. 아래 목록은 사용 가능한 유형을 보여줍니다.

사용 가능한 출력 뷰어

Kubeflow Pipelines 에서 제공하고 있는 출력 뷰어는 다음과 같습니다.

  • Confusion matrix
  • Markdown
  • ROC curve
  • Table
  • TensorBoard
  • Web app

사용 가능한 뷰어 유형과 각 유형에 필요한 메타 데이터 필드에 대해 알아보도록 하겠습니다.

출력 뷰어 사용하기

사용 가능한 뷰어 유형과 각 유형에 필요한 메타 데이터 필드에 대해 알아보도록 하겠습니다.

Confusion matrix

분류결과표(Confusion Matrix)는 대상의 원래 클래스와 모델이 예측한 클래스가 일치 하는 경우의 개수를 세어서, 그 결과를 표나 나타낸 것입니다. 정답 클래스는 행(row)으로 예측한 클래스는 열(column)로 나타냅니다.

타입: confusion_matrix

메타 데이터 필수 필드 :

  • format
  • labels
  • schema
  • source

메타 데이터 선택 필드 :

  • storage

뷰어는 다음 위치에서 분류 결과 데이터를 읽을 수 있습니다

  • source 필드에 포함 된 분류 결과 형식의 문자열입니다. storage 필드의 값은 inline 이어야 합니다.
  • source 필드에 지정한 경로의 원격 파일에 분류 결과를 읽어옵니다.. storage 필드는 비어 있거나 inline을 제외한 모든 값을 포함 할 수 있습니다.

confusion_matrix 뷰어는 데이터 구문을 분석하기 위해 schema에 정의된 값들을 사용합니다. labels은 x 및 y 축에 나타낼 클래스 이름입니다.

Example:

	metadata = {
    'outputs' : [
    # Confustion matrix that is hardcoded inline
    {
      'type': 'confusion_matrix',
      'format': 'csv',
      'schema': [
        {'name': 'target', 'type': 'CATEGORY'},
        {'name': 'predicted', 'type': 'CATEGORY'},
        {'name': 'count', 'type': 'NUMBER'},
      ],
      'source': <CONFUSION_MATRIX_CSV_INLINE>
			'storage': 'inline',
      'labels': list(map(str, vocab)),
    },
    # Confustion matrix that is read from a file
		{
      'type': 'confusion_matrix',
      'format': 'csv',
      'schema': [
        {'name': 'target', 'type': 'CATEGORY'},
        {'name': 'predicted', 'type': 'CATEGORY'},
        {'name': 'count', 'type': 'NUMBER'},
      ],
      'source': <CONFUSION_MATRIX_CSV_FILE>,
      'labels': list(map(str, vocab)),
    }]
  }
  with open('/mlpipeline-ui-metadata.json', 'w') as f:
    json.dump(metadata, f)

파이프 라인을 구성하고 실행하기

분류결과표(Confusion Matrix)는 출력하는 파이프 라인을 만들어 보겠습니다.

가. 프로그램 코드를 작성합니다.

컴포넌트를 만들기 위하여, sklearn을 사용한 간단한 분류결과표 프로그램 코드를 작성합니다. confusion_matrix() 메소드를 사용하여 분류결과표를 만들고, csv 형태의 파일로 저장하기 위해서 데이터를 가공합니다. 예제에서는 데이터를 외부 저장소에 저장하지 않고 inline 으로 사용하기 위해서, 저장한 csv 데이터를 다시 문자열로 읽어온 후, 메타데이터의 source 의 필드에 저장합니다. 만약 외부 저장소를 사용하고 싶으면, storage 필드를 삭제하고, source 필드에 저장할 곳의 위치를 지정하면 됩니다.

다음은 프로그램의 전체 코드입니다.

src/confusion_matrix.py

import json
import os

import pandas as pd
from sklearn.metrics import confusion_matrix

y_target = [2, 0, 2, 2, 0, 1]
y_pred = [0, 0, 2, 2, 0, 2]

vocab = [0, 1, 2]

cm = confusion_matrix(y_target, y_pred, labels=vocab)

data = []
for target_index, target_row in enumerate(cm):
    for predicted_index, count in enumerate(target_row):
        data.append((vocab[target_index], vocab[predicted_index], count))

df_cm = pd.DataFrame(data, columns=['target', 'predicted', 'count'])

output = '.'
cm_file = os.path.join(output, 'confusion_matrix.csv')
with open(cm_file, 'w') as f:
    df_cm.to_csv(f, columns=['target', 'predicted', 'count'], header=False, index=False)

lines = ''
with open(cm_file, 'r') as f:
    lines = f.read()

metadata = {
    'outputs': [{
            'type': 'confusion_matrix',
            'format': 'csv',
            'schema': [
                {'name': 'target', 'type': 'CATEGORY'},
                {'name': 'predicted', 'type': 'CATEGORY'},
                {'name': 'count', 'type': 'NUMBER'},
            ],
            'source': lines,
            'storage': 'inline',
            'labels': list(map(str, vocab)),
        }]
}

with open('/mlpipeline-ui-metadata.json', 'w') as f:
    json.dump(metadata, f)

나. 프로그램 코드가 포함된 컨테이너 이미지를 생성하고, 컨테이너 이미지 레지스트리에 업로드 합니다.

Dockerfile을 생성합니다.

Dockerfile

FROM python:3.6.10-slim

RUN pip install sklearn pandas

COPY ./src /app
WORKDIR /app

	CMD ["python", "/app/confusion_matrix.py"]

컨테이너 이미지를 빌드하겠습니다.

docker build -t kangwoo/kfp-confusion-matrix:0.0.1 .

빌드한 컨테이너 이미지를 컨테이너 이미지 레지스트리에 업로드 합니다.

docker push kangwoo/kfp-confusion-matrix:0.0.1

다. Kubeflow Pipelines DSL을 사용하여 컴포넌트를 작성합니다. 컴포넌트에서 사용하는 컨테이너 이미지를 정의합니다. 그리고 output_artifact_paths 파라미터를 사용하여, 메트릭 파일이 저장된 경로를 지정해 줍니다.

dsl.ContainerOp(
    name='confusion-matrix',
    image='kangwoo/kfp-confusion-matrix:0.0.1',
    output_artifact_paths={'mlpipeline-ui-metadata': '/mlpipeline-ui-metadata.json'}
  )

라. Kubeflow Pipelines DSL을 사용하여 파이프 라인 함수를 작성합니다. 파이프 라인을 정의하고 사용하는 컴포넌트들을 추가합니다. Kubeflow Pipelines SDK 를 사용하여 파이프라인을 빌드 한 후, 업로드하고 실행합니다.

import kfp
from kfp import dsl

def confusion_matrix_pipeline():
  dsl.ContainerOp(
    name='confusion-matrix',
    image='kangwoo/kfp-confusion-matrix:0.0.1',
    output_artifact_paths={'mlpipeline-ui-metadata': '/mlpipeline-ui-metadata.json'}
  )


if __name__ == '__main__':
    arguments = {}
    my_run = kfp.Client().create_run_from_pipeline_func(confusion_matrix_pipeline,
                                                        arguments=arguments,
                                                        experiment_name='Sample Experiment')

다음은 Kubeflow 파이프 라인 UI의 confusion_matrix 화면입니다.


Markdown

마크 다운 뷰어는 Kubeflow 파이프 라인 UI에서 마크 다운 문자열을 렌더링합니다.

타입 : markdown

메타 데이터 필수 필드 :

  • source

메타 데이터 선택 필드 :

  • storage :

뷰어는 다음 위치에서 마크 다운 데이터를 읽을 수 있습니다

  • source 필드에 포함 된 마크 다운 형식 문자열입니다. storage 필드의 값은 inline 이어야 합니다.
  • source 필드에 지정한 경로에서 원격 파일의 마크 다운 코드. storage 필드는 비어 있거나 inline을 제외한 모든 값을 포함 할 수 있습니다.

Example:

	metadata = {
    'outputs' : [
    # Markdown that is hardcoded inline
    {
      'storage': 'inline',
      'source': '# Inline Markdown\\n[A link](<https://www.kubeflow.org/>)',
      'type': 'markdown',
    },
    # Markdown that is read from a file
    {
      'source': 'gs:///your_bucket/your_markdown_file',
      'type': 'markdown',
    }]
  }
  with open('/mlpipeline-ui-metadata.json', 'w') as f:
    json.dump(metadata, f)

파이프 라인을 구성하고 실행하기

마크 다운을 출력하는 파이프 라인을 만들어 보겠습니다.

가. 프로그램 코드를 작성합니다.

컴포넌트를 만들기 위하여, 마크 다운을 저장하는 프로그램 코드를 작성합니다. 예제에서는 데이터를 외부 저장소에 저장하지 않고 inline 으로 사용하기 위해서, 저장한 csv 데이터를 다시 문자열로 읽어온 후, 메타데이터의 source 의 필드에 저장합니다. 만약 외부 저장소를 사용하고 싶으면, storage 필드를 삭제하고, source 필드에 저장할 곳의 위치를 지정하면 됩니다.

다음은 프로그램의 전체 코드입니다.

src/markdown.py

import json

metadata = {
    'outputs': [{
            'storage': 'inline',
            'source': '# Inline Markdown\\n[A link](<https://www.kubeflow.org/>)',
            'type': 'markdown',
        }]
}

with open('/mlpipeline-ui-metadata.json', 'w') as f:
    json.dump(metadata, f)

나. 프로그램 코드가 포함된 컨테이너 이미지를 생성하고, 컨테이너 이미지 레지스트리에 업로드 합니다.

Dockerfile을 생성합니다.

Dockerfile

FROM python:3.6.10-slim

COPY ./src /app
WORKDIR /app

CMD ["python", "/app/markdown.py"]

컨테이너 이미지를 빌드하겠습니다.

docker build -t kangwoo/kfp-markdown:0.0.1 .

빌드한 컨테이너 이미지를 컨테이너 이미지 레지스트리에 업로드 합니다.

docker push kangwoo/kfp-markdown:0.0.1

다. Kubeflow Pipelines DSL을 사용하여 컴포넌트를 작성합니다. 컴포넌트에서 사용하는 컨테이너 이미지를 정의합니다. 그리고 output_artifact_paths 파라미터를 사용하여, 메트릭 파일이 저장된 경로를 지정해 줍니다.

dsl.ContainerOp(
    name='markdown',
    image='kangwoo/kfp-markdown:0.0.1',
    output_artifact_paths={'mlpipeline-ui-metadata': '/mlpipeline-ui-metadata.json'}
  )

라. Kubeflow Pipelines DSL을 사용하여 파이프 라인 함수를 작성합니다. 파이프 라인을 정의하고 사용하는 컴포넌트들을 추가합니다. Kubeflow Pipelines SDK 를 사용하여 파이프라인을 빌드 한 후, 업로드하고 실행합니다.

import kfp
from kfp import dsl


def markdown_pipeline():
  dsl.ContainerOp(
    name='markdown',
    image='kangwoo/kfp-markdown:0.0.1',
    output_artifact_paths={'mlpipeline-ui-metadata': '/mlpipeline-ui-metadata.json'}
  )


if __name__ == '__main__':
    arguments = {}
    my_run = kfp.Client().create_run_from_pipeline_func(markdown_pipeline,
                                                        arguments=arguments,
                                                        experiment_name='Sample Experiment')

다음은 Kubeflow 파이프 라인 UI의 markdown 화면입니다.


ROC curve

ROC(Receiver Operator Characteristic) 곡선은 클래스 판별 기준값의 변화에 따른 위양성률(fall-out)과 재현율(recall)의 변화를 시각화한 것이다. 위양성율(fall-out)은 실제 양성 클래스에 속하지 않는 표본 중에 양성 클래스에 속한다고 출력한 표본의 비율을 뜻합니다. 위양성율은 FPR(false positive rate)라고도 합니다. 재현율(recall)은 실제 양성 클래스에 속한 표본 중에 양성 클래스에 속한다고 출력한 표본의 비율을 뜻합니다. 재현율은 TPR(true positive rate)라고도 합니다. 위양성률(fall-out)과 재현율(recall)은 일반적으로 양의 상관 관계가 있습니다.

타입 : roc

메타 데이터 필수 필드 :

  • format
  • schema
  • source

메타 데이터 선택 필드 :

  • storage

뷰어는 다음 위치에서 데이터를 읽을 수 있습니다

  • source 필드에 포함 된 ROC 곡선 형식의 문자열입니다. storage 필드의 값은 inline 이어야 합니다.
  • source 필드에 지정한 경로의 원격 파일에 분류 결과를 읽어옵니다. storage 필드는 비어 있거나 inline을 제외한 모든 값을 포함 할 수 있습니다.

Example:

	metadata = {
    'outputs': [
    # Roc that is hardcoded inline
    {
      'type': 'roc',
      'format': 'csv',
      'schema': [
        {'name': 'fpr', 'type': 'NUMBER'},
        {'name': 'tpr', 'type': 'NUMBER'},
        {'name': 'thresholds', 'type': 'NUMBER'},
      ],
      'source': <ROC_CSV_INLINE>
			'storage': 'inline',      
    },
    # Roc that is read from a file
		{
      'type': 'roc',
      'format': 'csv',
      'schema': [
        {'name': 'fpr', 'type': 'NUMBER'},
        {'name': 'tpr', 'type': 'NUMBER'},
        {'name': 'thresholds', 'type': 'NUMBER'},
      ],
      'source': <ROC_CSV_FILE>
    }]
  }
  with open('/mlpipeline-ui-metadata.json', 'w') as f:
    json.dump(metadata, f)

파이프 라인을 구성하고 실행하기

ROC(Receiver Operator Characteristic) 곡선을 출력하는 파이프 라인을 만들어 보겠습니다.

가. 프로그램 코드를 작성합니다.

컴포넌트를 만들기 위하여, sklearn을 이용하여 ROC 곡선을 생성하는 프로그램 코드를 작성합니다. roc_curve() 메소드를 사용하여 ROC 곡선 만들고, csv 형태의 파일로 저장하기 위해서 데이터를 가공합니다. 예제에서는 데이터를 외부 저장소에 저장하지 않고 inline 으로 사용하기 위해서, 저장한 csv 데이터를 다시 문자열로 읽어온 후, 메타데이터의 source 의 필드에 저장합니다. 만약 외부 저장소를 사용하고 싶으면, storage 필드를 삭제하고, source 필드에 저장할 곳의 위치를 지정하면 됩니다.

다음은 프로그램의 전체 코드입니다.

src/roc.py

import json
import os

import pandas as pd
from sklearn.datasets import make_classification
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import roc_curve

X, y = make_classification(n_samples=1000, weights=[0.95, 0.05], random_state=5)

model = LogisticRegression().fit(X, y)
y_hat = model.predict(X)

fpr, tpr, thresholds = roc_curve(y, model.decision_function(X))

output = '.'
df_roc = pd.DataFrame({'fpr': fpr, 'tpr': tpr, 'thresholds': thresholds})
roc_file = os.path.join(output, 'roc.csv')
with open(roc_file, 'w') as f:
    df_roc.to_csv(f, columns=['fpr', 'tpr', 'thresholds'], header=False, index=False)

lines = ''
with open(roc_file, 'r') as f:
    lines = f.read()

metadata = {
    'outputs': [{
        'type': 'roc',
        'format': 'csv',
        'schema': [
            {'name': 'fpr', 'type': 'NUMBER'},
            {'name': 'tpr', 'type': 'NUMBER'},
            {'name': 'thresholds', 'type': 'NUMBER'},
        ],
        'source': lines,
        'storage': 'inline',
    }]
}
with open('/mlpipeline-ui-metadata.json', 'w') as f:
    json.dump(metadata, f)

나. 프로그램 코드가 포함된 컨테이너 이미지를 생성하고, 컨테이너 이미지 레지스트리에 업로드 합니다.

Dockerfile을 생성합니다.

Dockerfile

FROM python:3.6.10-slim

RUN pip install sklearn pandas

COPY ./src /app
WORKDIR /app

CMD ["python", "/app/roc.py"]

컨테이너 이미지를 빌드하겠습니다.

docker build -t kangwoo/kfp-roc:0.0.1 .

빌드한 컨테이너 이미지를 컨테이너 이미지 레지스트리에 업로드 합니다.

docker push kangwoo/kfp-roc:0.0.1

다. Kubeflow Pipelines DSL을 사용하여 컴포넌트를 작성합니다. 컴포넌트에서 사용하는 컨테이너 이미지를 정의합니다. 그리고 output_artifact_paths 파라미터를 사용하여, 메트릭 파일이 저장된 경로를 지정해 줍니다.

dsl.ContainerOp(
    name='roc',
    image='kangwoo/kfp-roc:0.0.1',
    output_artifact_paths={'mlpipeline-ui-metadata': '/mlpipeline-ui-metadata.json'}
  )

라. Kubeflow Pipelines DSL을 사용하여 파이프 라인 함수를 작성합니다. 파이프 라인을 정의하고 사용하는 컴포넌트들을 추가합니다. Kubeflow Pipelines SDK 를 사용하여 파이프라인을 빌드 한 후, 업로드하고 실행합니다.

import kfp
from kfp import dsl


def roc_pipeline():
    dsl.ContainerOp(
        name='roc',
        image='kangwoo/kfp-roc:0.0.1',
        output_artifact_paths={'mlpipeline-ui-metadata': '/mlpipeline-ui-metadata.json'}
    )


if __name__ == '__main__':
    arguments = {}
    my_run = kfp.Client().create_run_from_pipeline_func(roc_pipeline,
                                                        arguments=arguments,
                                                        experiment_name='Sample Experiment')

다음은 Kubeflow 파이프 라인 UI의 roc 화면입니다.

ROC 곡선을 볼 때, 커서를 ROC 곡선 위로 가져 가면 커서의 가장 가까운 fpr 및 tpr 값에 사용 된 thresholds 값을 볼 수 있습니다.


Table

table 뷰어는 지정된 소스 경로의 데이터에서 HTML 테이블을 작성합니다. 여기서 헤더 필드는 테이블의 첫 번째 행에 표시 될 값을 지정합니다. 테이블은 페이징 기능을 지원합니다.

타입 : table

메타 데이터 필수 필드 :

  • format
  • header
  • source

메타 데이터 선택 필드 :

  • storage

뷰어는 다음 위치에서 데이터를 읽을 수 있습니다

  • source 필드에 포함 된 데이터 문자열입니다. storage 필드의 값은 inline 이어야 합니다.
  • source 필드에 지정한 경로의 원격 파일에 분류 결과를 읽어옵니다. storage 필드는 비어 있거나 inline을 제외한 모든 값을 포함 할 수 있습니다.

Example:

	metadata = {
    'outputs' : [
    # Table that is hardcoded inline
    {
      'type': 'table',
      'format': 'csv',
      'header': [x['name'] for x in schema],
      'source': <TABLE_CSV_INLINE>
			'storage': 'inline',
    },
    # Table that is read from a file
		{
      'type': 'table',
      'format': 'csv',
      'header': [x['name'] for x in schema],
      'source': <TABLE_CSV_FILE>
    }]
  }
  with open('/mlpipeline-ui-metadata.json', 'w') as f:
    json.dump(metadata, f)

파이프 라인을 구성하고 실행하기

테이블을 출력하는 파이프 라인을 만들어 보겠습니다.

가. 프로그램 코드를 작성합니다.

컴포넌트를 만들기 위하여, 테이블을 생성하는 프로그램 코드를 작성합니다. 테이블롤 출력할 데이터를 만든 후, csv 형태의 파일로 저장하기 위해서 데이터를 가공합니다. 예제에서는 데이터를 외부 저장소에 저장하지 않고 inline 으로 사용하기 위해서, 저장한 csv 데이터를 다시 문자열로 읽어온 후, 메타데이터의 source 의 필드에 저장합니다. 만약 외부 저장소를 사용하고 싶으면, storage 필드를 삭제하고, source 필드에 저장할 곳의 위치를 지정하면 됩니다.

다음은 프로그램의 전체 코드입니다.

src/table.py

import json
import os

import pandas as pd
from sklearn.metrics import classification_report

y_true = [1, 0, 1, 1, 0, 1]
y_pred = [0, 1, 1, 0, 0, 1]

target_names = ['class 0', 'class 1']
report = classification_report(y_true, y_pred, target_names=target_names, output_dict=True)
print(report)

df_report = pd.DataFrame(report).transpose()

output = '.'
table_file = os.path.join(output, 'table.csv')
with open(table_file, 'w') as f:
    df_report.to_csv(f, header=False)

lines = ''
with open(table_file, 'r') as f:
    lines = f.read()

metadata = {
    'outputs': [{
        'type': 'table',
        'format': 'csv',
        'header': [''] + [x for x in df_report],
        'source': lines,
        'storage': 'inline',
    }]
}

with open('/mlpipeline-ui-metadata.json', 'w') as f:
    json.dump(metadata, f)

나. 프로그램 코드가 포함된 컨테이너 이미지를 생성하고, 컨테이너 이미지 레지스트리에 업로드 합니다.

Dockerfile을 생성합니다.

Dockerfile

FROM python:3.6.10-slim

RUN pip install sklearn pandas

COPY ./src /app
WORKDIR /app

CMD ["python", "/app/table.py"]

컨테이너 이미지를 빌드하겠습니다.

docker build -t kangwoo/kfp-table:0.0.1 .

빌드한 컨테이너 이미지를 컨테이너 이미지 레지스트리에 업로드 합니다.

docker push kangwoo/kfp-table:0.0.1

다. Kubeflow Pipelines DSL을 사용하여 컴포넌트를 작성합니다. 컴포넌트에서 사용하는 컨테이너 이미지를 정의합니다. 그리고 output_artifact_paths 파라미터를 사용하여, 메트릭 파일이 저장된 경로를 지정해 줍니다.

dsl.ContainerOp(
    name='table',
    image='kangwoo/kfp-table:0.0.1',
    output_artifact_paths={'mlpipeline-ui-metadata': '/mlpipeline-ui-metadata.json'}
  )

라. Kubeflow Pipelines DSL을 사용하여 파이프 라인 함수를 작성합니다. 파이프 라인을 정의하고 사용하는 컴포넌트들을 추가합니다. Kubeflow Pipelines SDK 를 사용하여 파이프라인을 빌드 한 후, 업로드하고 실행합니다.

import kfp
from kfp import dsl


def table_pipeline():
  dsl.ContainerOp(
    name='table',
    image='kangwoo/kfp-table:0.0.1',
    output_artifact_paths={'mlpipeline-ui-metadata': '/mlpipeline-ui-metadata.json'}
  )


if __name__ == '__main__':
    arguments = {}
    my_run = kfp.Client().create_run_from_pipeline_func(table_pipeline,
                                                        arguments=arguments,
                                                        experiment_name='Sample Experiment')

다음은 Kubeflow 파이프 라인 UI의 table 화면입니다.

TensorBoard

tensorboard 뷰어는 Start Tensorboard 버튼을 출력 페이지에 추가합니다.

타입 : tensorboard

메타 데이터 필수 필드 :

  • source

출력 페이지에서 다음을 수행 할 수 있습니다.

  • Start Tensorboard 을 클릭하면, Kubeflow 클러스터에 Tensorboard 인스턴스가 시작됩니다. Tensorboard 포드가 실행되면, 버튼 텍스트가Open Tensorboard 로 전환됩니다.
  • Open Tensorboard 를 클릭하면 source 필드에 지정한 logdir 데이터를 읽어오는 TensorBoard 화면에 접속 할 수 있습니다..
  • Delete Tensorboard 을 클릭하면, Tensorboard 인스턴스가 종료됩니다.

Kubeflow Pipelines UI는 TensorBoard 인스턴스를 완전히 관리하지 않습니다. Start Tensorboard  버튼은 편리한 기능이므로 파이프 라인 실행을 볼 때 워크 플로우를 중단 할 필요가 없습니다. Kubernetes 관리 도구를 사용하여 TensorBoard 포드를 재활용하거나 삭제해야합니다.

Example:

	metadata = {
    'outputs' : [{
      'type': 'tensorboard',
      'source': <TENSORBOARD_PATH>,
    }]
  }
  with open('/mlpipeline-ui-metadata.json', 'w') as f:
    json.dump(metadata, f)

외부 저장소 설정 하기

tensorboard 뷰어는 텐서보드를 실행할 수 있는 버튼을 제공하고 있습니다. 버튼을 클릭하면 텐서보드가 실행이 됩니다. 이때 실행되는 텐서보드는 텐서플로우의 로그가 저장된 위치에 접근할 수 있어야합니다. 그래서 텐서보드가 지원하는 형태의 외부 저장소가 필요합니다.

예제에서는 kubeflow와 함께 설치된 minio를 S3 호환 저장소로 사용하겠습니다. 만약 별도의 저장소를 사용하고 있다면, 외부 저장소 설정 하기는 넘어가도 됩니다.

가. minio에 사용할 버킷을 생성하겠습니다.

버킷을 생성하기 위해서 minio에 접속해야합니다.

다음 명령어를 실행하면, minio의 서비스 정보를 조회할 수 있습니다.

kubectl -n kubeflow get service minio-service

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

NAME            TYPE        CLUSTER-IP     EXTERNAL-IP   PORT(S)    AGE
minio-service   ClusterIP   10.103.56.47   <none>        9000/TCP   15d

로컬 환경에서 minio에 접속하기 위해서 port-forward 를 실행합니다.

kubectl -n kubeflow port-forward svc/minio-service 9000:9000

포트 포워딩이 실행되면, 다음과 같은 결과를 확인 할 수 있습니다.

Forwarding from 127.0.0.1:9000 -> 9000
Forwarding from [::1]:9000 -> 9000

웹브라이저를 실행시켜서 http://localhost:9000/ 으로 접속합니다.

정상적으로 접속되면, 로그인 화면을 확인 할 수 있습니다.

AccessKey 와 Secret Key 를 입력합니다. 기본 값을 minio / minio123 입니다.

로그인 되면 다음과 같은 화면을 볼 수 있습니다.

화면 오른쪽 하단의 + 버튼을 누른 후, “Create bucket”을 눌러서 버킷을 생성합니다.

예제서 사용할 버컷의 이름은 tensorboard 입니다.

나. 텐서보드가 s3에 접근할 수 있도록 인증 정보를 설정하겠습니다.

s3에 접근 하기위한 AccessKey와 SecretKey 정보를 쿠버네티스의 Secret 리소스로 저장하겠습니다.

다음 명령어를 실행하면 시크릿 리소스가 생성됩니다.

export AWS_ACCESS_KEY_ID=minio
export AWS_SECRET_ACCESS_KEY=minio123

kubectl -n kubeflow create secret generic ml-pipeline-aws-secret \\
    --from-literal=accesskey=$AWS_ACCESS_KEY_ID \\
    --from-literal=secretkey=$AWS_SECRET_ACCESS_KEY

텐서보드가 실행될때 s3에 접근할 수 있도록, s3 설정 정보를 쿠버네티스 ConfigMap 리소스로 저장하겠습니다. 중요한 정보인 AWS_ACCESS_KEY_ID 와 AWS_SECRET_ACCESS_KEY 는 앞서 생성한 Secret 리소스에서 가져오게 되어있습니다.

viewer-tensorboard-template-configmap.yaml

apiVersion: v1
kind: ConfigMap
metadata:
  name: ml-pipeline-ui-viewer-template
data:
  viewer-tensorboard-template.json: |
    {
        "spec": {
            "containers": [
                {
                    "env": [
                        {
                            "name": "AWS_ACCESS_KEY_ID",
                            "valueFrom": {
                                "secretKeyRef": {
                                    "name": "ml-pipeline-aws-secret",
                                    "key": "accesskey"
                                }
                            }
                        },
                        {
                            "name": "AWS_SECRET_ACCESS_KEY",
                            "valueFrom": {
                                "secretKeyRef": {
                                    "name": "ml-pipeline-aws-secret",
                                    "key": "secretkey"
                                }
                            }
                        },
                        {
                            "name": "S3_ENDPOINT",
                            "value": "minio-service.kubeflow.svc.cluster.local:9000"
                        },
                        {
                            "name": "AWS_ENDPOINT_URL",
                            "value": "<http://minio-service.kubeflow.svc.cluster.local:9000>"
                        },
                        {
                            "name": "AWS_REGION",
                            "value": "us-east-1"
                        },
                        {
                            "name": "S3_USE_HTTPS",
                            "value": "0"
                        },
                        {
                            "name": "S3_VERIFY_SSL",
                            "value": "0"
                        }
                    ]
                }
            ]
        }
    }

다음 명령어를 실행하여, ml-pipeline-ui-viewer-template ConfigMap을 생성합니다.

kubectl -n kubeflow create -f viewer-tensorboard-template-configmap.yaml

다. 이제 생성한 ConfigMap을 사용할 수 있도록, ml-pipeline-ui 의 설정 정보를 변경하도록 하겠습니다.

다음 명령어를 실행하여 ml-pipeline-ui 디플로이먼트를 수정합니다.

kubectl -n kubeflow edit deployment ml-pipeline-ui

생성한 ConfigMap 을 볼륨으로 마운트 해 줍니다.

...
        volumeMounts:
        - mountPath: /etc/config
          name: config-volume
...
     volumes:
      - configMap:
          defaultMode: 420
          name: ml-pipeline-ui-viewer-template
        name: config-volume

마운트한 볼륨에 있는 viewer-tensorboard-template.json 파일을 VIEWER_TENSORBOARD_POD_TEMPLATE_SPEC_PATH 라는 환경 변수를 지정해 줍니다.

      containers:
      - env:
        - name: VIEWER_TENSORBOARD_POD_TEMPLATE_SPEC_PATH
          value: /etc/config/viewer-tensorboard-template.json

다음은 변경한 ml-pipeline-ui 디플로이먼트 매니페스트의 일부분 입니다.

apiVersion: extensions/v1beta1
kind: Deployment
metadata:
...
  name: ml-pipeline-ui
  namespace: kubeflow
...
spec:
...
  template:
...
    spec:
      containers:
      - env:
        - name: VIEWER_TENSORBOARD_POD_TEMPLATE_SPEC_PATH
          value: /etc/config/viewer-tensorboard-template.json
...
        image: gcr.io/ml-pipeline/frontend:0.3.0
        imagePullPolicy: IfNotPresent
        name: ml-pipeline-ui
...
        volumeMounts:
        - mountPath: /etc/config
          name: config-volume
...
      dnsPolicy: ClusterFirst
...
      volumes:
      - configMap:
          defaultMode: 420
          name: ml-pipeline-ui-viewer-template
        name: config-volume
...

변경한 내용을 저장하면, 새로운 설정이 적용된 포드가 실행됩니다.

파이프 라인을 구성하고 실행하기

텐서보드 뷰어를 사용하는 파이프 라인을 만들어 보겠습니다.

가. 프로그램 코드를 작성합니다.

컴포넌트를 만들기 위하여, 텐서플로우 로그 데이터를 저장하는 프로그램 코드를 작성합니다. 텐서플로우 케라스를 이용하여 mnist 이미지를 식별하는 모델을 생성합니다. 그리고 tf.keras.callbacks.TensorBoard 를 사용하여 학습할때 로그를 남깁니다. 그리고 Kubeflow Pipelines 에서 텐서보드 뷰어를 사용할 수 있도록 /mlpipeline-ui-metadata.json 파일에 메타데이터를 저장합니다.

다음은 프로그램의 전체 코드입니다.

src/tensorflow_mnist.py

from __future__ import absolute_import, division, print_function, unicode_literals

import argparse
import json

import tensorflow as tf


def train():
    parser = argparse.ArgumentParser()
    parser.add_argument('--tb_log_dir', default='./data/logs', type=str)
    args = parser.parse_args()

    tb_log_dir = args.tb_log_dir

    print("TensorFlow version: ", tf.__version__)

    mnist = tf.keras.datasets.mnist

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

    model = tf.keras.models.Sequential([
        tf.keras.layers.Flatten(input_shape=(28, 28)),
        tf.keras.layers.Dense(128, activation='relu'),
        tf.keras.layers.Dropout(0.2),
        tf.keras.layers.Dense(10, activation='softmax')
    ])

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

    model.summary()

    callbacks = [tf.keras.callbacks.TensorBoard(log_dir=tb_log_dir)]

    print("Training...")
    model.fit(x_train, y_train, epochs=5, validation_split=0.2, callbacks=callbacks)

    score = model.evaluate(x_test, y_test, batch_size=128)
    print('Test accuracy: ', score[1])

    metadata = {
        'outputs': [{
            'type': 'tensorboard',
            'source': tb_log_dir,
        }]
    }
    with open('/mlpipeline-ui-metadata.json', 'w') as f:
        json.dump(metadata, f)


if __name__ == '__main__':
    train()

나. 프로그램 코드가 포함된 컨테이너 이미지를 생성하고, 컨테이너 이미지 레지스트리에 업로드 합니다.

Dockerfile을 생성합니다.

Dockerfile

FROM tensorflow/tensorflow:2.1.0-py3

COPY ./src /app
WORKDIR /app


ENTRYPOINT ["python", "/app/tensorflow_mnist.py"]

컨테이너 이미지를 빌드하겠습니다.

docker build -t kangwoo/kfp-tensorboard:0.0.1 .

빌드한 컨테이너 이미지를 컨테이너 이미지 레지스트리에 업로드 합니다.

docker push kangwoo/kfp-tensorboard:0.0.1

다. Kubeflow Pipelines DSL을 사용하여 컴포넌트를 작성합니다. 컴포넌트에서 사용하는 컨테이너 이미지를 정의합니다. 그리고 output_artifact_paths 파라미터를 사용하여, 메트릭 파일이 저장된 경로를 지정해 줍니다.

텐서플로우가 실행되는 컴포트넌트에서 s3에 접근할 수 있도록, s3 설정 정보를 환경 변수로 넘겨 주었습니다.

s3_endpoint = 'minio-service.kubeflow.svc.cluster.local:9000'
    minio_endpoint = "http://" + s3_endpoint
    minio_username = "minio"
    minio_key = "minio123"
    minio_region = "us-east-1"

    dsl.ContainerOp(
        name='tensorboard',
        image='kangwoo/kfp-tensorboard:0.0.1',
        arguments=['--tb_log_dir', tb_log_dir],
        output_artifact_paths={'mlpipeline-ui-metadata': '/mlpipeline-ui-metadata.json'}
    ).add_env_variable(V1EnvVar(name='S3_ENDPOINT', value=s3_endpoint)) \\
        .add_env_variable(V1EnvVar(name='AWS_ENDPOINT_URL', value=minio_endpoint)) \\
        .add_env_variable(V1EnvVar(name='AWS_ACCESS_KEY_ID', value=minio_username)) \\
        .add_env_variable(V1EnvVar(name='AWS_SECRET_ACCESS_KEY', value=minio_key)) \\
        .add_env_variable(V1EnvVar(name='AWS_REGION', value=minio_region)) \\
        .add_env_variable(V1EnvVar(name='S3_USE_HTTPS', value='0')) \\
        .add_env_variable(V1EnvVar(name='S3_VERIFY_SSL', value='0'))

라. Kubeflow Pipelines DSL을 사용하여 파이프 라인 함수를 작성합니다. 파이프 라인을 정의하고 사용하는 컴포넌트들을 추가합니다. Kubeflow Pipelines SDK 를 사용하여 파이프라인을 빌드 한 후, 업로드하고 실행합니다.

import kfp
from kfp import dsl

from kubernetes.client.models import V1EnvVar


def tensorboard_pipeline(tb_log_dir):
    s3_endpoint = 'minio-service.kubeflow.svc.cluster.local:9000'
    minio_endpoint = "http://" + s3_endpoint
    minio_username = "minio"
    minio_key = "minio123"
    minio_region = "us-east-1"

    dsl.ContainerOp(
        name='tensorboard',
        image='kangwoo/kfp-tensorboard:0.0.1',
        arguments=['--tb_log_dir', tb_log_dir],
        output_artifact_paths={'mlpipeline-ui-metadata': '/mlpipeline-ui-metadata.json'}
    ).add_env_variable(V1EnvVar(name='S3_ENDPOINT', value=s3_endpoint)) \\
        .add_env_variable(V1EnvVar(name='AWS_ENDPOINT_URL', value=minio_endpoint)) \\
        .add_env_variable(V1EnvVar(name='AWS_ACCESS_KEY_ID', value=minio_username)) \\
        .add_env_variable(V1EnvVar(name='AWS_SECRET_ACCESS_KEY', value=minio_key)) \\
        .add_env_variable(V1EnvVar(name='AWS_REGION', value=minio_region)) \\
        .add_env_variable(V1EnvVar(name='S3_USE_HTTPS', value='0')) \\
        .add_env_variable(V1EnvVar(name='S3_VERIFY_SSL', value='0'))


if __name__ == '__main__':
    arguments = {'tb_log_dir': 's3://tensorboard/mnist'}
    my_run = kfp.Client().create_run_from_pipeline_func(tensorboard_pipeline,
                                                        arguments=arguments,
                                                        experiment_name='Sample Experiment')

다음은 Kubeflow 파이프 라인 UI의 tensorboard 화면입니다.

구동하면 싶은 텐서보드의 버전을 선택할 수 있습니다. “Start Tensorboard”를 클릭하면 텐서보드가 실행됩니다.

텐서보드가 실행되면 버튼이 “Open Tensorboard”로 바뀝니다. “Open Tensorboard” 버튼을 클릭하면 텐서보드 화면으로 접속할 수 있습니다.

다음은 텐서보드 접속 화면입니다.


Web app

web-app 뷰어는 사용자 정의 출력을 렌더링 할 수있는 유연성을 제공합니다.

타입 : web-app

메타 데이터 필수 필드 :

  • source

메타 데이터 선택 필드 :

  • storage

뷰어는 다음 위치에서 데이터를 읽을 수 있습니다

  • source 필드에 포함 된 데이터 문자열입니다. storage 필드의 값은 inline 이어야 합니다.
  • source 필드에 지정한 경로의 원격 파일에 분류 결과를 읽어옵니다. storage 필드는 비어 있거나 inline을 제외한 모든 값을 포함 할 수 있습니다.

컴포넌트가 생성하는 HTML 파일을 지정할 수 있습니다. Kubeflow Pipelines UI는 해당 HTML을 출력 페이지에서 렌더링합니다. HTML 파일은 파일 시스템의 다른 파일에 대한 참조가 없어야 합니다. HTML 파일에는 웹 파일에 대한 절대 참조가 포함될 수 있습니다. web-app 내에서 실행되는 콘텐츠는 iframe에서 샌드박스 처리되며 Kubeflow Pipelines UI와 통신 할 수 없습니다.

Example:

static_html_path = os.path.join(output_dir, _OUTPUT_HTML_FILE)
  file_io.write_string_to_file(static_html_path, rendered_template)

  metadata = {
    'outputs' : [
    # Web app that is hardcoded inline
    {
      'type': 'web-app',
      'storage': 'inline',
      'source': <STATIC_HTML_INLINE>,
    },
    # Web app  that is read from a file
		{
      '
      'type': 'web-app',
      'source': <STATIC_HTML_PATH>,
    }]
  }
  with file_io.FileIO('/mlpipeline-ui-metadata.json', 'w') as f:
    json.dump(metadata, f)

파이프 라인을 구성하고 실행하기

web-app을 출력하는 파이프 라인을 만들어 보겠습니다.

가. 프로그램 코드를 작성합니다.

컴포넌트를 만들기 위하여, HTML 생성하는 프로그램 코드를 작성합니다. 예제에서는 데이터를 외부 저장소에 저장하지 않고 inline 으로 사용하기 위해서, 저장한 csv 데이터를 다시 문자열로 읽어온 후, 메타데이터의 source 의 필드에 저장합니다. 만약 외부 저장소를 사용하고 싶으면, storage 필드를 삭제하고, source 필드에 저장할 곳의 위치를 지정하면 됩니다.

다음은 프로그램의 전체 코드입니다.

src/webapp.py

import json

metadata = {
    'outputs': [{
        'type': 'web-app',
        'storage': 'inline',
        'source': '<p><strong>Kubeflow pipelines</strong> are reusable end-to-end ML workflows built using the Kubeflow Pipelines SDK.</p>',
    }]
}

with open('/mlpipeline-ui-metadata.json', 'w') as f:
    json.dump(metadata, f)

나. 프로그램 코드가 포함된 컨테이너 이미지를 생성하고, 컨테이너 이미지 레지스트리에 업로드 합니다.

Dockerfile을 생성합니다.

Dockerfile

FROM python:3.6.10-slim

COPY ./src /app
WORKDIR /app

CMD ["python", "/app/webapp.py"]

컨테이너 이미지를 빌드하겠습니다.

docker build -t kangwoo/kfp-webapp:0.0.1 .

빌드한 컨테이너 이미지를 컨테이너 이미지 레지스트리에 업로드 합니다.

docker push kangwoo/kfp-webapp:0.0.1

다. Kubeflow Pipelines DSL을 사용하여 컴포넌트를 작성합니다. 컴포넌트에서 사용하는 컨테이너 이미지를 정의합니다. 그리고 output_artifact_paths 파라미터를 사용하여, 메트릭 파일이 저장된 경로를 지정해 줍니다.

  dsl.ContainerOp(
    name='webapp',
    image='kangwoo/kfp-webapp:0.0.1',
    output_artifact_paths={'mlpipeline-ui-metadata': '/mlpipeline-ui-metadata.json'}
  )

라. Kubeflow Pipelines DSL을 사용하여 파이프 라인 함수를 작성합니다. 파이프 라인을 정의하고 사용하는 컴포넌트들을 추가합니다. Kubeflow Pipelines SDK 를 사용하여 파이프라인을 빌드 한 후, 업로드하고 실행합니다.

import kfp
from kfp import dsl


def webapp_pipeline():
  dsl.ContainerOp(
    name='webapp',
    image='kangwoo/kfp-webapp:0.0.1',
    output_artifact_paths={'mlpipeline-ui-metadata': '/mlpipeline-ui-metadata.json'}
  )


if __name__ == '__main__':
    arguments = {}
    my_run = kfp.Client().create_run_from_pipeline_func(webapp_pipeline,
                                                        arguments=arguments,
                                                        experiment_name='Sample Experiment')

다음은 Kubeflow 파이프 라인 UI의 web-app 화면입니다.

Kubeflow Pipelines – 파이프라인 메트릭

Kubeflow Pipelines는 스칼라 메트릭을 저장하고 시각화해서 보여주는 기능을 제공하고 있습니다. 단순히 메트릭을 정해진 형식대로 로컬 파일에 저장하기만 하면 사용할 수 있습니다. 메트릭을 지정된 형식에 맞게 로컬 파일에 저장하기만 하면, 파이프 라인 에이전트가 로컬 파일을 런타임 메트릭으로 자동 업로드합니다. 업로드 된 메트릭은 Kubeflow Pipelines UI의 실행 페이지에서 조회해 볼 수 있습니다.

메트릭 파일 출력하기

메트릭을 사용하려면 프로그램이 /mlpipeline-metrics.json이라는 파일을 작성해야합니다.

예를 들면 다음과 같습니다.

 accuracy = 0.9712345
 metrics = {
	 'metrics': [{
	   'name': 'accuracy', 
     'numberValue':  float(accuracy), 
     'format': "PERCENTAGE",
   }]
 }
 with file_io.FileIO('/mlpipeline-metrics.json', 'w') as f:
	 json.dump(metrics, f)

메트릭 파일은 다음과 같은 규칙을 지켜야 합니다.

  • name 은 ^[a-z]([-a-z0-9]{0,62}[a-z0-9])?$ 패턴을 따라야합니다
  • numberValue 숫자 값이어야합니다.
  • format 은 PERCENTAGERAW 을 사용할 수 있고, 설정하지 않을 수 있습니다.

그리고 dsl.ContainerOp()의 output_artifact_paths 에 mlpipeline-metrics 를 추가해줘야 합니다.

dsl.ContainerOp(
    ...
    output_artifact_paths={'mlpipeline-metrics': '/mlpipeline-metrics.json'}
)

파이썬 함수를 이용하는 경량 컴포넌트의 경우 구문이 약간 다릅니다. 경량 컴포넌트에서 메트릭을 출력하려면, 다음과 같이 NamedTuple 을 사용해야 합니다.

def my_function() -> \\
        NamedTuple('output', [('mlpipeline_metrics', 'metrics')]):
    accuracy = 0.9712345
    metrics = {
        'metrics': [{
            'name': 'accuracy',
            'numberValue': float(accuracy),
            'format': "PERCENTAGE",
        }]
    }
    from collections import namedtuple

    output = namedtuple('output', ['mlpipeline_metrics'])
    return output(json.dumps(metrics))

컴포넌트에서 메트릭 파일 출력하기

파이프 라인을 구성하고 실행하기

파이프 라인 메트릭을 출력하는 파이프 라인을 만들어 보겠습니다.

텐서플로우 케라스를 사용한 mnist의 숫자 이미지 판별 모델의 accuracy 값과 loss 값을 메트릭으로 출력해 보겠습니다.

가. 애플리케이션 코드를 작성합니다.

모델 코드는 기존과 거의 동일합니다. 메트릭을 출력하기위해서, json 형태의 메트릭을 파일로 저장하는 코드를 추가합니다. name 은 소문자로 시작해야하며, 소문자와 숫자 그리고 ‘-‘를 사용할 수 있습니다. numberValue 는 숫자 값이어야 합니다. 이 규칙을 지키지 않으면, UI 화면에서 결과를 확인할 수 없습니다.

    metrics = {
        'metrics': [{
            'name': 'accuracy',
            'numberValue': float(accuracy),
            'format': "PERCENTAGE",
        }, {
            'name': 'loss',
            'numberValue': float(loss),
            'format': "RAW",
        }]
    }

    with file_io.FileIO('/mlpipeline-metrics.json', 'w') as f:
        json.dump(metrics, f)

다음은 애플리케이션의 전체 코드입니다.

mnist-simple.py

from __future__ import absolute_import, division, print_function, unicode_literals

import json

import tensorflow as tf
from tensorflow.python.lib.io import file_io


def train():
    print("TensorFlow version: ", tf.__version__)

    mnist = tf.keras.datasets.mnist

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

    model = tf.keras.models.Sequential([
        tf.keras.layers.Flatten(input_shape=(28, 28)),
        tf.keras.layers.Dense(128, activation='relu'),
        tf.keras.layers.Dropout(0.2),
        tf.keras.layers.Dense(10, activation='softmax')
    ])

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

    print("Training...")
    training_history = model.fit(x_train, y_train, epochs=5, validation_split=0.2)

    results = model.evaluate(x_test, y_test, batch_size=128)
    print('test loss, test acc:', results)

    loss = results[0]
    accuracy = results[1]
    metrics = {
        'metrics': [{
            'name': 'accuracy',
            'numberValue': float(accuracy),
            'format': "PERCENTAGE",
        }, {
            'name': 'loss',
            'numberValue': float(loss),
            'format': "RAW",
        }]
    }

    with file_io.FileIO('/mlpipeline-metrics.json', 'w') as f:
        json.dump(metrics, f)


if __name__ == '__main__':
    train()

나. 애플리케이션 코드가 포함된 컨테이너 이미지를 생성하고, 컨테이너 이미지 레지스트리에 업로드 합니다.

Dockerfile을 생성합니다.

Dockerfile

FROM tensorflow/tensorflow:2.1.0-py3

RUN mkdir -p /app
ADD mnist-simple.py /app/
WORKDIR /app

CMD ["python", "/app/mnist-simple.py"]

컨테이너 이미지를 빌드하겠습니다.

docker build -t kangwoo/mnist-kfp-metrics:kfp.

빌드한 컨테이너 이미지를 컨테이너 이미지 레지스트리에 업로드 합니다.

docker push kangwoo/mnist-kfp-metrics:kfp

다. Kubeflow Pipelines DSL을 사용하여 컴포넌트를 작성합니다. 컴포넌트에서 사용하는 컨테이너 이미지를 정의합니다. 그리고 output_artifact_paths 파라미터를 사용하여, 메트릭 파일이 저장된 경로를 지정해 줍니다.

dsl.ContainerOp(
	name='mnist-kfp-metrics',
  image='kangwoo/mnist-kfp-metrics:kfp',
  output_artifact_paths={'mlpipeline-metrics': '/mlpipeline-metrics.json'}
)

라. Kubeflow Pipelines DSL을 사용하여 파이프 라인 함수를 작성합니다. 파이프 라인을 정의하고 사용하는 컴포넌트들을 추가합니다. Kubeflow Pipelines SDK 를 사용하여 파이프라인을 빌드 한 후, 업로드하고 실행합니다.

@dsl.pipeline(
    name='Pipeline Metrics',
    description='Export and visualize pipeline metrics'
)
def pipeline_metrics_pipeline():
    dsl.ContainerOp(
        name='mnist-kfp-metrics',
        image='kangwoo/mnist-kfp-metrics:kfp',
        output_artifact_paths={'mlpipeline-metrics': '/mlpipeline-metrics.json'}
    )


pipeline_package_path = 'pipeline_metrics_pipeline.zip'
kfp.compiler.Compiler().compile(pipeline_metrics_pipeline, pipeline_package_path)

client = kfp.Client()
my_experiment = client.create_experiment(name='Sample Experiment')
my_run = client.run_pipeline(my_experiment.id, 'pipeline_metrics_pipeline', pipeline_package_path)

경랑 컴포넌트에서 메트릭 파일 출력하기

파이프 라인을 구성하고 실행하기

파이썬 함수를 사용하는 경량 컴포넌트에서 메트릭을 출력하는 방법에 대해서 알아보겠습니다.

메트릭이 저장된 파일 경로를 파이프라인 시스템에 알려주기 위해서 파이썬 함수에서 NamedTuple 을 사용하였습니다. 이 부분을 제외하면 기존과 동일하기 때문에 설명은 생략하겠습니다.

코드의 단순화를 위해서 accuracy 값과 loss 값을 하드 코딩 하였습니다.

pipeline_metrics_fn_pipeline.py

from typing import NamedTuple

import kfp
from kfp.components import func_to_container_op


@func_to_container_op
def train() -> \\
        NamedTuple('output', [('mlpipeline_metrics', 'metrics')]):
    import json
    loss = 0.812345
    accuracy = 0.9712345
    metrics = {
        'metrics': [{
            'name': 'accuracy',
            'numberValue': float(accuracy),
            'format': "PERCENTAGE",
        }, {
            'name': 'loss',
            'numberValue': float(loss),
            'format': "RAW",
        }]
    }
    from collections import namedtuple

    output = namedtuple('output', ['mlpipeline_metrics'])
    return output(json.dumps(metrics))


def pipeline_metrics_fn_pipeline():
    train()


if __name__ == '__main__':
    arguments = {}
    my_run = kfp.Client().create_run_from_pipeline_func(pipeline_metrics_fn_pipeline, arguments=arguments,
                                                        experiment_name='Sample Experiment')

메트릭 보기

시각화된 메트릭을 보려면, Kubeflow Pipelines UI에서 Experiments 페이지를 엽니다. 실험 중 하나를 클릭하면, 실행 페이지가 열리고 메트릭이 실행 목록 테이블에 표시됩니다. 메트릭은 각 실행에 대한 열로 나타납니다.

다음은 실행에 대한 정확도를 보여주는 화면입니다.

해당 실행 단계의 “Run output” 탭에서도 메트릭을 확인 할 수 있습니다.

Kubeflow Pipelines – 경량 파이썬 컴포넌트

파이썬 코드를 이용한 경량 컴포넌트를 사용하기

경량 파이썬 컴포넌트는 생성한 코드를 컨테이너 이미지로 빌드 하지 않아도 됩니다. 그래서 빠르게 반복하여 코드를 변경하는 경우 유용하게 사용할 수 있습니다.

경량 파이썬 컴포넌트

경량 컴포넌트를 빌드하려면, 먼저 독립형 파이썬 함수를 정의해야 합니다. 그런 다음 kfp.components.func_to_container_op()를 호출하여 파이썬 함수를 파이프 라인에서 사용할 수 있는 컴포넌트로 변환해야 합니다

경량 컴포넌트를 만들기 위해서는 몇 가지 제약 사항이 있습니다.

  • 파이썬 함수의 기능은 독립적이어야 합니다.
    • 정의한 함수 외부에서 선언한 코드를 사용해서는 안됩니다.
    • import 는 함수 내부에 선언해야합니다.
  • import 는 기본 이미지에서 사용 가능한 패키지만 가져올 수 있습니다.
    • 사용할 패키지가 기본 이미지에 없는 경우에는, 해당 패키지가 포함된 이미지를 사용해야 합니다.
  • 파이썬 함수의 파리미터로 숫자를 사용하려면, 파라미터에 타입 힌트가 있어야 합니다. 지원되는 타입은 int, float, bool입니다. 모든 파라미터는 문자열로 전달됩니다.
  • 출력값을 여러개 사용하려면, 파이썬의 typing.NamedTuple 타입 힌트를 사용해야합니다.

kfp.components.func_to_container_op(func)

이 함수는 입력 받은 파이썬 함수를 컴포넌트로 변환합니다. 변환한 컴포넌트는 파이프라인에서 사용할 수 있습니다.

def func_to_container_op(func, output_component_file=None, base_image: str = None, extra_code='', packages_to_install: List[str] = None, modules_to_capture: List[str] = None, use_code_pickling=False):

전달 인자

다음은 func_to_container_op()에서 사용하는 주요 전달 인자입니다.

  • base_image : 컴포넌트에서 사용할 기본 컨테이너 이미지입니다. 경량 컴포넌트의 경우 이미지에는 파이썬 3.5 이상이 설치 되어 있어야합니다. 기본값은 tensorflow/tensorflow:1.13.2-py3 입니다. (선택)
  • output_component_file: 컴포넌트 정의를 로컬 파일에 작성합니다. 이 파일은 공유할 때 사용할 수 있습니다. (선택)
  • packages_to_install: 사용자의 함수를 실행하기 전에 설치할 파이썬 패키지 목록입니다. (선택)

경량 파이썬 컴포넌트 만들기

하나의 값을 출력하는 경량 파이썬 컴포넌트 만들기

먼저 컴포넌트로에서 사용할 파이썬 함수를 작성합니다. 입력된 두 개의 값을 더한 값을 반환하는 간단한 함수입니다.

def add(a: float, b: float) -> float:
    return a + b

func_to_container_op() 호출하여 파이썬 함수를 컴포넌트로 변환합니다. 별도의 컨테이너 이미지를 지정하지 않으면, 기본 이미지인 tensorflow/tensorflow:1.13.2-py3 를 사용합니다.

add_op = comp.func_to_container_op(add)

DSL을 사용하여 파이프 라인을 구성합니다.

KFP SDK 사용하여 코드에서 파이프 라인을 바로 실행하였습니다.

def lightweight_component_pipeline(a='10', b='20'):
    add_task = add_op(a, b)
    print_text(add_task.output)


if __name__ == '__main__':
    arguments = {'a': '1000', 'b': '4'}
    my_run = kfp.Client().create_run_from_pipeline_func(lightweight_component_pipeline, arguments=arguments, experiment_name='Basic Experiment')

다음은 파이프라인을 구성하고 실행하는 전체 코드입니다.

lightweight_component.py

import kfp
from kfp import dsl
from kfp import components
from kfp.components import func_to_container_op, InputPath, OutputPath


@func_to_container_op
def print_text(text_path: InputPath()):
    with open(text_path, 'r') as reader:
        for line in reader:
            print(line, end = '')


def add(a: float, b: float) -> float:
    return a + b


add_op = components.func_to_container_op(add)


def lightweight_component_pipeline(a='10', b='20'):
    add_task = add_op(a, b)
    print_text(add_task.output)


if __name__ == '__main__':
    arguments = {'a': '1000', 'b': '4'}
    my_run = kfp.Client().create_run_from_pipeline_func(lightweight_component_pipeline, arguments=arguments, experiment_name='Basic Experiment')

파이프 라인 실행 결과 확인하기

KFP UI 화면에서 결과를 조회해 볼 수 있습니다. 왼쪽 메뉴에서 Experiments 클릭하면 Experiment 목록을 조회할 수 있습니다. 예제에서 사용한 Experiment 이름이 “Sample Experiment” 이므로, Sample Experiment를 클릭하면 실행 목록을 확인 할 수 있습니다.

“Run name”이 “Multiply component pipeline” 인 것을 클릭하면 실행 결과를 확인할 수 있습니다.


여러개의 값을 출력하는 경량 파이썬 컴포넌트 만들기

먼저 컴포넌트로에서 사용할 파이썬 함수를 작성합니다. 입력된 두 개의 값을 더한 값과 곱한 값 두 개를 반환하는 함수입니다. 출력값이 여러개 이기 때문에 파이썬의 typing.NamedTuple 타입 힌트를 사용해야합니다.

def add_multiply_two_numbers(a: float, b: float) \\
        -> NamedTuple('Outputs', [('sum', float), ('product', float)]):
    return (a + b, a * b)

func_to_container_op() 함수를 호출하여 컴포넌트를 로드합니다. 별도의 컨테이너 이미지를 지정하지 않으면, 기본 이미지인 tensorflow/tensorflow:1.13.2-py3 를 사용합니다.

add_multiply_two_numbers_op = comp.func_to_container_op(add_multiply_two_numbers)

DSL을 사용하여 파이프 라인을 구성합니다.

KFP SDK 사용하여 코드에서 파이프 라인을 바로 실행하였습니다.

@dsl.pipeline(
    name='Multiple outputs pipeline',
    description='A pipeline to showcase multiple outputs.'
)
def multiple_outputs_pipeline(a='10', b='20'):
    add_multiply_task = add_multiply_two_numbers_op(a, b)
    print_op('sum={}, product={}'.format(add_multiply_task.outputs['sum'],
                                         add_multiply_task.outputs['product']))


if __name__ == '__main__':
    arguments = {'a': '3', 'b': '4'}
    my_run = kfp.Client().create_run_from_pipeline_func(multiple_outputs_pipeline,
                                                        arguments=arguments, experiment_name='Basic Experiment')

다음은 파이프라인을 구성하고 실행하는 전체 코드입니다.

multiple_outputs.py

import kfp
from kfp import dsl
import kfp.components as comp
from typing import NamedTuple


def print_op(msg):
    return dsl.ContainerOp(
        name='Print',
        image='alpine:3.6',
        command=['echo', msg],
    )


def add_multiply_two_numbers(a: float, b: float) \\
        -> NamedTuple('Outputs', [('sum', float), ('product', float)]):
    return (a + b, a * b)


add_multiply_two_numbers_op = comp.func_to_container_op(add_multiply_two_numbers)


@dsl.pipeline(
    name='Multiple outputs pipeline',
    description='A pipeline to showcase multiple outputs.'
)
def multiple_outputs_pipeline(a='10', b='20'):
    add_multiply_task = add_multiply_two_numbers_op(a, b)
    print_op('sum={}, product={}'.format(add_multiply_task.outputs['sum'],
                                         add_multiply_task.outputs['product']))


if __name__ == '__main__':
    arguments = {'a': '3', 'b': '4'}
    my_run = kfp.Client().create_run_from_pipeline_func(multiple_outputs_pipeline,
                                                        arguments=arguments, experiment_name='Basic Experiment')

파이프 라인 실행 결과 확인하기

KFP UI 화면에서 결과를 조회해 볼 수 있습니다. 왼쪽 메뉴에서 Experiments 클릭하면 Experiment 목록을 조회할 수 있습니다. 예제에서 사용한 Experiment 이름이 “Sample Experiment” 이므로, Sample Experiment를 클릭하면 실행 목록을 확인 할 수 있습니다.

“Run name”이 “Multiply component pipeline” 인 것을 클릭하면 실행 결과를 확인할 수 있습니다.


경량 파이썬 컴포넌트에서 데이터 입력과 출력

컴포넌트에는 입력 및 출력이 있습니다. 한 작업의 출력값을 다른 작업의 입력값에 전달하여 데이터 입력과 출력을 연결하여 컴포넌트의 작업간에 데이터를 생성하고 소비하는 것입니다. 파이썬 함수를 컴포넌트로 변환하여 사용할 때, 입력 및 출력 데이터를 생성하고 소비하는 방법에 대해서 알아 보겠습니다.

작은 데이터

작은 데이터는 프로그램의 명령 줄 인수로 쉽게 전달할 수있는 데이터입니다. 작은 데이터 크기는 몇 킬로바이트를 초과하지 않아야 합니다. 숫자나, 작은 문자열, 그리고 URL 같은 것을 예로 들 수 있습니다. 작은 리스트, 사전 및 JSON 구조는 사용해도 괜찮지만, 데이터의 크기가 크다면 파일 기반의 데이터 전달 방법을 사용하는 것이 좋습니다.

작은 출력 데이터들은 문자열로 직렬화 됩니다. 그리고 입력 데이터로 전달할때 역직렬화 됩니다. 예를 들어 일반적인 유형인 str, int, float, bool, list, dict 유형들은 내장된 직렬화/역직렬화 변환가 있습니다. 하지만 다른 유형의 데이터를 사용하다면, 수동으로 직렬화 해야합니다. 유형 어노테이션을 올바르게 지정하지 않으면, 자동으로 직렬화 되지 않고 문자열로 전달합니다.

큰 데이터

큰 데이터는 파일을 통해서 전달합니다. 출력 데이터를 파일로 쓰고, 다른 컴포넌트를 파일을 읽어서 입력 데이터로 사용합니다. 입력 및 출력 파일의 경로는 시스템에 의해 결정되며, 함수에는 문자열 형태로 전달됩니다.

파일을 통해서 데이터를 전달하려면, InputPathOutputPath 파라미터 어노테이션을 사용하며 됩니다.

InputPath 파라미터 어노테이션은 함수가 해당 입력 데이터를 파일로 소비하고 싶다고 시스템에 알려줍니다. 시스템은 데이터를 다운로드하여 로컬 파일에 쓴 다음, 해당 파일의 경로(path)를 함수에 전달합니다.

OutputPath 파라미터 어노테이션은 함수가 해당 출력 데이터를 파일로 생성하고 싶다고 시스템에 알려줍니다. 시스템은 함수가 데이터를 출력해야하는 파일의 경로(paht) 준비해서 전달해줍니다. 함수가 종료되면 시스템은 스토리지에 데이터를 업로드하여 다운 스트림 컴포넌트로 전달합니다.

InputPathOutputPath 에 타입을 지정하여 소비/생산 데이터의 유형을 지정할 수 있습니다. 유형은 파이썬 유형이거나, 임의의 문자열일 수 있습니다. OutputPath('TFModel')은 함수가 파일에 쓴 데이터의 유형이 TFModel 임을 나타냅니다. InputPath('TFModel') 은 함수가 파일에서 읽는 데이터의 유형이 TFModel 을 나타냅니다. 파이프라인 작성자가 입력과 출력을 연결하면, 시스템이 해당 유형이 일치하는 확인합니다.

일반적으로 함수가 컴포넌트로 변환될 때 입력 및 출력 이름은 파라미터 이름을 따르지만, _path_file 같은 접미사가 사용된 경우에는 이름에서 제거됩니다. 예를들어 파라미터 가 number_file_path: InputPath(int) 인 경우에는 입력은 number: int 로 바뀌게 됩니다. 이것은 인자 전달을 보다 자연스럽게 보이게 하기 위함입니다.

InputPathOutputPath 사용하기

write_numbers() 함수는 OutputPath 어노테이션을 사용하여, 시스템에게 출력 데이터를 파일로 쓴다는 것을 알려줍니다. 파일을 저장할 경로인 numbers_path 를 시스템에게 전달받아서 생성한 숫자를 파일로 씁니다.

@func_to_container_op
def write_numbers(numbers_path: OutputPath(str), start: int = 1, count: int = 10):
    with open(numbers_path, 'w') as writer:
        for i in range(start, start + count):
            writer.write(str(i) + '\\n')

sum_multiply_numbers() 함수는 InputPath 어노테이션을 사용하여, 시스템으로 부터 입력 데이터를 파일로 전달 받습니다. 입력 데이터가 저장된 파일의 경로인 numbers_path 를 전달받아서, 데이터를 읽어옵니다. 입력 데이터는 숫자가 저장되어 있습니다. 읽어온 숫자를 모두 더하고, 곱한 후에 OutputPath 어노테이션을 사용하여 파일로 저장합니다.

@func_to_container_op
def sum_multiply_numbers(
        numbers_path: InputPath(str),
        sum_path: OutputPath(str),
        product_path: OutputPath(str)):

    sum = 0
    product = 1
    with open(numbers_path, 'r') as reader:
        for line in reader:
            sum = sum + int(line)
            product = product * int(line)
    with open(sum_path, 'w') as writer:
        writer.write(str(sum))
    with open(product_path, 'w') as writer:
        writer.write(str(product))

print_text() 함수는 InputPath 어노테이션을 사용하여, 시스템으로 부터 입력 데이터를 파일로 전달 받아서 화면에 출력합니다.

@func_to_container_op
def print_text(text_path: InputPath()):
    with open(text_path, 'r') as reader:
        for line in reader:
            print(line, end = '')

컴포넌트들을 사용하여, 파이프라인을 구성합니다. 앞서 얘기 했듯이, 함수가 컴포넌트로 변환될 때 입력 및 출력 이름은 파라미터 이름을 따르지만, _path_file 같은 접미사가 사용된 경우에는 이름에서 제거됩니다. sum_multiply_numbers() 함수에서 출력 파라미터 이름은 sum_pathproduct_path 였지만 출력에서 값을 가지고 오기 위해 사용한 이름은 sumproduct 입니다. 즉 _path 접미사가 이름에서 제거되었습니다.

def python_input_output_pipeline(count='10'):
    numbers_task = write_numbers(count=count)
    sum_multiply_task = sum_multiply_numbers(numbers_task.output)

    print_text(sum_multiply_task.outputs['sum'])
    print_text(sum_multiply_task.outputs['product'])

다음은 파이프라인을 구성하고 실행하는 전체 코드입니다.

python_input_output_pipeline.py

import kfp
from kfp.components import func_to_container_op, InputPath, OutputPath


@func_to_container_op
def write_numbers(numbers_path: OutputPath(str), start: int = 1, count: int = 10):
    with open(numbers_path, 'w') as writer:
        for i in range(start, start + count):
            writer.write(str(i) + '\\n')


@func_to_container_op
def print_text(text_path: InputPath()):
    with open(text_path, 'r') as reader:
        for line in reader:
            print(line, end = '')


@func_to_container_op
def sum_multiply_numbers(
        numbers_path: InputPath(str),
        sum_path: OutputPath(str),
        product_path: OutputPath(str)):

    sum = 0
    product = 1
    with open(numbers_path, 'r') as reader:
        for line in reader:
            sum = sum + int(line)
            product = product * int(line)
    with open(sum_path, 'w') as writer:
        writer.write(str(sum))
    with open(product_path, 'w') as writer:
        writer.write(str(product))


def python_input_output_pipeline(count='10'):
    numbers_task = write_numbers(count=count)
    sum_multiply_task = sum_multiply_numbers(numbers_task.output)

    print_text(sum_multiply_task.outputs['sum'])
    print_text(sum_multiply_task.outputs['product'])


~~i~~f __name__ == '__main__':
    arguments = {'count': '10'}
    my_run = kfp.Client().create_run_from_pipeline_func(python_input_output_pipeline,
                                                        arguments=arguments, experiment_name='Basic Experiment')

파이프 라인 실행 결과 확인하기

KFP UI 화면에서 결과를 조회해 볼 수 있습니다. 왼쪽 메뉴에서 Experiments 클릭하면 Experiment 목록을 조회할 수 있습니다. 예제에서 사용한 Experiment 이름이 “Sample Experiment” 이므로, Sample Experiment를 클릭하면 실행 목록을 확인 할 수 있습니다.

“Run name”이 “python_input_output_pipeline” 인 것을 클릭하면 실행 결과를 확인할 수 있습니다.

Kubeflow Pipelines – 재사용 가능한 컴포넌트

다양한 파이프 라인에서 재사용할 수 있는 컴포넌트에 대해서 알아 보겠습니다. 그리고 재사용 가능한 컴포넌트를 만들기 위한 컴포넌트 프로그램을 작성하는 방법과 컴포넌트를 정의하는 파일에 대해서 알아 보도록 하겠습니다.

재사용 가능한 컴포넌트 이해하기

재사용 가능한 컴포넌트를 만들면, 다양한 파이프라인에서 쉽게 가져다 사용할 수 있습니다. 재사용 가능한 컴포넌트를 만들기 위해서는 먼저 컴포넌트에서 사용할 프로그램을 작성하고 컨테이너화 합니다. 그리고 Kubeflow Pipelines 시스템에서 사용하는 컴포넌트 스펙을 정의하는 파일을 YAML 형식으로 작성하면 됩니다.

재사용 컴포넌트 만드는 단계

다음은 재사용 컴포넌트를 만드는 단계를 요약 한 것입니다.

  • 컴포넌트에서 사용할 프로그램을 작성하십시오. 프로그램은 파일이나 명령 행 인수를 사용하여 다른 컴퍼넌트와 데이터를 주고 받을 수 있습니다.
  • 프로그램을 컨테이너화 하십시오.
  • Kubeflow Pipelines 시스템의 컴포넌트를 설명하는 컴포넌트 스펙을 YAML 형식으로 작성하십시오.
  • Kubeflow Pipelines SDK 를 사용하여 파이프라인에서 사용할 컴포넌트를 로드하고, 파이프라인을 실행하십시오.

컴포넌트 프로그램

컴포넌트에서 사용할 프로그램을 작성해야 합니다. 합니다. 프로그램은 다른 컴포넌트로부터 데이터를 받기 위해서, 파일이나 명령행 인수를 사용해야 합니다.

컴포넌트 컨테이너화

컴포넌트를 만들기 위해서는 작성한 프로그램을 컨테이너 이미지로 패키지해야 합니다. 컨테이너화 된 컴포넌트는 파이프라인에서 독립적으로 실행됩니다. 컴포넌트는 동일한 프로세스에서 실행되지 않기 때문에, 메모리를 사용하여 컴포넌트간에 데이터를 직접 공유 할 수 없습니다. 그래서, 데이터를 네트워크를 통해 이동할 수 있도록 전달하는 데이터를 직렬화 해야합니다. 그러면, 다운스트림 컴포넌트에서 데이터를 역직렬화 해서 사용할 수 있습니다.

컴포넌트 스펙

Kubeflow Pipelines 시스템은 컴포넌트의 데이터 모델을 정의하기 위해서 YAML 형식의 파일을 사용합니다.

다음은 컴포넌트 정의의 주요 부분입니다.

  • Metadata : 이름, 설명등의 메타데이터를 정의합니다.
  • Interface : 입력과 출력을 위한 값들의 이름, 유형, 기본값등을 정의합니다.
  • Implementation : 제공된 컴포넌트 입력 값들을 가지고 컴포넌트를 실행하는 방법을 정의합니다. 컴포넌트의 실행이 완료되면 출력 값을 얻는 방법도 정의해야합니다.

다음은 컨퍼넌트 스펙을 정의한 YAML 파일입니다.

name: Keras - Train classifier
description: Trains classifier using Keras sequential model
inputs:
  - {name: training_set_features_path, type: {GcsPath: {data_type: TSV}}, description: 'Local or GCS path to the training set features table.'}
  - {name: training_set_labels_path, type: {GcsPath: {data_type: TSV}}, description: 'Local or GCS path to the training set labels (each label is a class index from 0 to num-classes - 1).'}
  - {name: output_model_uri, type: {GcsPath: {data_type: Keras model}}, description: 'Local or GCS path specifying where to save the trained model. The model (topology + weights + optimizer state) is saved in HDF5 format and can be loaded back by calling keras.models.load_model'} #Remove GcsUri and move to outputs once artifact passing support is checked in.
  - {name: model_config, type: {GcsPath: {data_type: Keras model config json}}, description: 'JSON string containing the serialized model structure. Can be obtained by calling model.to_json() on a Keras model.'}
  - {name: number_of_classes, type: Integer, description: 'Number of classifier classes.'}
  - {name: number_of_epochs, type: Integer, default: '100', description: 'Number of epochs to train the model. An epoch is an iteration over the entire `x` and `y` data provided.'}
  - {name: batch_size, type: Integer, default: '32', description: 'Number of samples per gradient update.'}
outputs:
  - {name: output_model_uri, type: {GcsPath: {data_type: Keras model}}, description: 'GCS path where the trained model has been saved. The model (topology + weights + optimizer state) is saved in HDF5 format and can be loaded back by calling keras.models.load_model'} #Remove GcsUri and make it a proper output once artifact passing support is checked in.
implementation:
  container:
    image: gcr.io/ml-pipeline/sample/keras/train_classifier
    command: [python3, /pipelines/component/src/train.py]
    args: [
      --training-set-features-path, {inputValue: training_set_features_path},
      --training-set-labels-path, {inputValue: training_set_labels_path},
      --output-model-path, {inputValue: output_model_uri},
      --model-config-json, {inputValue: model_config},
      --num-classes, {inputValue: number_of_classes},
      --num-epochs, {inputValue: number_of_epochs},
      --batch-size, {inputValue: batch_size},

      --output-model-path-file, {outputPath: output_model_uri},
    ]

컴포넌트 사용하기

Kubeflow Pipelines SDK를 이용하여 컴포넌트를 로드하면, 파이프 라인에서 컴포넌트를 사용할 수 있습니다.

#Load the component
train_op = comp.load_component(url='<https://raw.githubusercontent.com/Ark-kun/pipelines/Added-sample-component/components/sample/keras/train_classifier/component.yaml>')

#Use the component as part of the pipeline
def pipeline():
    train_task = train_op(
        training_set_features_path=os.path.join(testdata_root, 'training_set_features.tsv'),
        training_set_labels_path=os.path.join(testdata_root, 'training_set_labels.tsv'),
        output_model_uri=os.path.join(temp_dir_name, 'outputs/output_model/data'),
        model_config=Path(testdata_root).joinpath('model_config.json').read_text(),
        number_of_classes=2,
        number_of_epochs=10,
        batch_size=32,
    )

컴포넌트간에 데이터 전달하기

컴포넌트의 개념은 함수의 개념과 매우 유사합니다. 모든 컴포넌트는 입력 및 출력을 가질 수 있습니다. 컴포넌트 코드는 입력으로 전달 된 데이터를 가져 와서 출력용 데이터를 생성합니다. 파이프 라인은 실행된 컴포넌트의 출력을, 다른 컴포넌트의 입력으로 데이터를 전달하여 컴포넌트의 데이터를 서로 공유합니다. 이는 함수가 다른 함수를 호출하고 그 결과를 전달하는 방법과 매우 유사합니다. 파이프 라인 시스템은 실제 데이터 전달을 처리하는 반면 컴포넌트는 입력 데이터를 소비하고 출력 데이터를 생성합니다.

컨테이너화 된 프로그램들간의 데이터 전달

컴포넌트를 작성할 때 컴포넌트가 업스트림 및 다운 스트림 컴포넌트와 통신하는 방법에 대해 생각해야합니다. 즉, 입력 데이터를 소비하고 출력 데이터를 생성하는 방법입니다.

데이터 생성

데이터를 출력하려면 컴포넌트의 프로그램이 출력 데이터를 특정 위치에 기록하여, 시스템이 다른 컴포넌트에게 데이터를 전달할 수 있도록 해야합니다. 즉 프로그램은 데이터를 저장하는 경로를 하드 코딩해서는 안됩니다. 프로그램은 출력 데이터 경로를 명령행 인수로 받아서 처리해야합니다.

외부 시스템에서 데이터 생성

일부 시나리오에서는 컴포넌트의 출력 데이터를 외부 서비스에 저장할 수도 있습니다. 이 경우에 컴포넌트의 프로그램은 데이터를 외부 서비스에 저장한 후, 해당 데이터의 위치 식별자 정보를 출력해야합니다. 그래야 다른 컴포넌트에서 데이터를 가져올 수 있기 때문입니다. 이 경우에는 데이터가 Kubeflow Pipilines 시스템에 보관되지 않기 때문에, 파이프 라인 시스템은 데이터에 대한 일관 성 및 재현성을 보장 할 수 없습니다.

데이터 소비

명령행 프로그램이 데이터를 소비하는 방법은 일반적으로 두 가지가 있습니다.

  • 데이터가 작을 경우에는 보통 명령행 인수로 바로 전달합니다. program.py --param 100
  • 데이터 크거나, 이진 데이터일 경우에는 파일로 저장한 후, 파일 경로를 명령행 인수로 전달합니다. 파일로 저장한 데이터를 다른 컴포넌트에게 전달하려면, Kubeflow Pipelines 시스템이 이 정보를 알고 있어야 합니다.

재사용 가능한 컴포넌트 만들기

컴포넌트 파일 구성

다음은 Kubeflow Pipielins에서 권장하는, 컴포넌트 파일의 구성입니다. 반드시 이러한 방식으로 파일을 구성할 필요는 없습니다. 하지만 표준을 정해 놓으면, 이미지 작성, 테스트에 동일한 스크립트를 재사용할 수 있는 장정이 있습니다.

components/<component group>/<component name>/

    src/*            #Component source code files
    tests/*          #Unit tests
    run_tests.sh     #Small script that runs the tests
    README.md        #Documentation. Move to docs/ if multiple files needed

    Dockerfile       #Dockerfile to build the component container image
    build_image.sh   #Small script that runs docker build and docker push

    component.yaml   #Component definition in YAML format

프로그램 코드 작성

두 개의 입력 데이터와 하나의 출력 데이터가 있는 프로그램을 작성해 보겠습니다. 두 개의 입력 데이터는 명령행 인수로 바로 전달하는 작은 데이터와, 파일 경로를 명령행 인수로 전달하는 큰 데이테를 사용합니다. 이 예제는 파이썬 3 으로 작성되었습니다.

program.py

#!/usr/bin/env python3
import argparse
from pathlib import Path

def do_work(input1_file, output1_file, param1):
  for x in range(param1):
    line = next(input1_file)
    if not line:
      break
    _ = output1_file.write(line)

# Defining and parsing the command-line arguments
parser = argparse.ArgumentParser(description='Program description')
parser.add_argument('--input-path', type=str, help='Path of the local file containing the Input data.')
parser.add_argument('--param', type=int, default=100, help='Parameter.')
parser.add_argument('--output-path', type=str, help='Path of the local file where the Output data should be written.')
args = parser.parse_args()

Path(args.output1_path).parent.mkdir(parents=True, exist_ok=True)

with open(args.input1_path, 'r') as input1_file:
    with open(args.output1_path, 'w') as output1_file:
        do_work(input1_file, output1_file, args.param1)

이 프로그램의 명령행 호출은 다음과 같습니다

python3 program.py --input1-path <local file path to Input 1 data> \\
                   --param1 <value of Param1 input> \\
                   --output1-path <local file path for the Output 1 data>

Dockerfile을 작성하고, 프로그램을 컨테이너화하기

컨테이너 이미지를 만들기 위해서 Dockerfile 을 생성합니다. 파이썬 코드를 사용하고 있으므로, 베이스 이미지를 파이썬으로 하였습니다.

Dockerfile

FROM python:3.6

COPY ./src /pipelines/component/src

Dockerfile을 기반으로 컨테이너 이미지를 빌드하고, 해당 이미지를 컨테이너 이미지 레지스트리로 푸시하기 위해서 build_image.sh  스크립트를 작성하였습니다.

build_image.sh

#!/bin/bash -e
image_name=kangwoo/kfp-component # Specify the image name here
image_tag=latest
full_image_name=${image_name}:${image_tag}

cd "$(dirname "$0")"
docker build -t "${full_image_name}" .
docker push "$full_image_name"

# Output the strict image name (which contains the sha256 image digest)
docker inspect --format="{{index .RepoDigests 0}}" "${full_image_name}"

스크립트를 실행 가능하게 만듭니다.

chmod +x build_image.sh

build_image.sh  스크립트를 실행하면, 컨테이너 이미지가 빌드되고, 지정한 컨테이너 이미지 레시트리로 이미지가 푸시됩니다.

Sending build context to Docker daemon  7.168kB
Step 1/2 : FROM python:3.6
 ---> 1daf62e8cab5
Step 2/2 : COPY ./src /pipelines/component/src
 ---> Using cache
 ---> 2bc266c5c9d8
Successfully built 2bc266c5c9d8
Successfully tagged kangwoo/kfp-component:latest
The push refers to repository [docker.io/kangwoo/kfp-component]
cce013c10a7c: Preparing 
cce013c10a7c: Layer already exists 
... 
latest: digest: sha256:30ea205b7cb1253a36f82f1ec99f0eec87cadd95938317ee3c802f2b78cec368 size: 2424
kangwoo/kfp-component@sha256:30ea205b7cb1253a36f82f1ec99f0eec87cadd95938317ee3c802f2b78cec368

컴포넌트 정의 파일 작성

To create a component from your containerized program you need to write component specification in YAML format that describes the component for the Kubeflow Pipelines system.

컨테이너화 된 프로그램을 이용해서 컴포넌트로 만들려면, Kubeflow Pipelines 시스템에서 사용하는 YAML 형식의 컴포넌트 스펙을 작성해야합니다.

component.yaml파일을 생성하고, 컴포넌트의 구현(implementation) 섹션에서 사용할 컨테이너 이미지를 지정합니다. 그리고 명령(command) 섹션에서 컨테이에 포함된 프로그램을 실행하기 위해 명령을 지정합니다.

implementation:
  container:
    image: kangwoo/kfp-component:latest
    command: [
      python3, /pipelines/component/src/program.py,
      --input-path,  {inputPath:  input_1},
      --param,       {inputValue: parameter_1},
      --output-path, {outputPath: output_1},
    ]

command 섹션에는 앵글 괄호로 표현되는 플레이스홀더가 포함되어 있습니다. 플레이스홀더는 프로그램이 실행 되기 전에 특정 값 또는 경로로 대체됩니다. component.yaml 에서는 매핑 구문을 사용하여 플레이스홀더를 지정할 수 있습니다.

사용 가능한 플레이스홀더는 세 가지가 있습니다.

  • {inputValue: Some input name} : 이 플레이스홀더는 지정한 입력을 인수의 값으로 대체됩니다. 작은 데이터에 유용합니다.
  • {inputPath: Some input name} : 이 플레이스홀더는 입력 데이터를 컴포넌트로 전달하기 위해서 자동 생성된 로컬 파일의 경로로 대체됩니다. 즉, 파이프라인 시스템이 입력 인수 데이터를 파일로 쓰고, 해당 데이터 파일의 경로를 컴포넌트 프로그램에 전달하게 되는 것입니다.
  • {outputPath: Some output name}: 이 플레이스홀더는 프로그램이 출력 데이터를 저장해야 하는 자동 생성된 로컬 파일 경로로 대체됩니다.

명령행에 플레이스 홀더를 배치하는 것 외에도, 입력(inputs) 및 출력(outputs) 섹션에 해당 입력 및 출력 스펙을 추가해야합니다. 입력/출력 스펙에는 이름, 유형, 설명 및 기본값이 포함되어 있습니다. 이중에서 이름(name)은 반드시 포함되어야합니다. 입력/출력에 사용하는 이름은 자유로운 형식의 문자열이지만 YAML 문법을 따라야 합니다.

inputs:
- {name: input_1, type: String, description: 'Data for Input 1'}
- {name: parameter_1, type: Integer, default: '1', description: 'Parameter 1 description'}
outputs:
- {name: output_1, description: 'Output 1 data'}

컴포넌트의 이름과 설명 같은 메타 데이터를 추가합니다.

name: Multiply component
description: Multiplication.

component.yaml

name: Multiply component
description: Multiplication.
inputs:
- {name: input_1, type: String, description: 'Data for Input 1'}
- {name: parameter_1, type: Integer, default: '1', description: 'Parameter 1 description'}
outputs:
- {name: output_1, description: 'Output 1 data'}
implementation:
  container:
    image: kangwoo/kfp-component:latest
    command: [
      python3, /pipelines/component/src/program.py,
      --input-path,  {inputPath:  input_1},
      --param,       {inputValue: parameter_1},
      --output-path, {outputPath: output_1},
    ]

Kubeflow Pipelines SDK를 사용하여 파이프 라인에서 컴포넌트 사용하기

컴포넌트를 로드하고, 이를 사용하여 파이프 라인을 구성하는 방법을 알아보겠습니다. 예제 파이프라인은 3단계로 이루어져 있습니다.

1단계 number_op()는 1-10까지의 숫자를 출력합니다. 2단계인 multiply_op()는 출력된 숫자를 입력으로 받아서 곱하기 연산을 하고 그 결과를 출력합니다. 그리고 3단계인 print_op()는 출력된 곱하기 결과를 화면에 출력합니다.

2단계인 multiply_op()에서 직접 만든 재사용 컹포넌트를 사용하겠습니다.

1단계에 사용하는 number_op()는 1-10까지의 숫자를 출력합니다.

def number_op():
    return dsl.ContainerOp(
        name='Generate numbers',
        image='python:alpine3.6',
        command=['sh', '-c'],
        arguments=['python -c "print(\\'1\\\\n2\\\\n3\\\\n4\\\\n5\\\\n6\\\\n7\\\\n8\\\\n9\\\\n10\\')" | tee /tmp/output'],
        file_outputs={'output': '/tmp/output'}
    )

3단계에서 사용하는 print_op()는 입력된 결과를 화면에 출력합니다

def print_op(msg):
    return dsl.ContainerOp(
        name='Print',
        image='alpine:3.6',
        command=['echo', msg],
    )

load_component_from_file 호출하여 컴포넌트를 로드합니다. 만약 파일이 아니라 URL을 사용한다면 load_component_from_url을 대신 사용할 수 있습니다. 컴포넌트를 로드하기 위해서는 component.yaml 파일에 접근할 수 있으면 됩니다. 그리고 파이프 라인이 실행되는 쿠버네티스 클러스터는 컴포넌트에 정의된 컨테이너 이미지에 접근할 수 있어야합니다.

component_root = './multiply'
multiply_op = kfp.components.load_component_from_file(os.path.join(component_root, 'component.yaml'))
# multiply_op = kfp.components.load_component_from_url('https://....../component.yaml')

DSL을 사용하여 파이프 라인을 구성합니다. load_component_from_file을 호출하여 로드한 multiply_op 컴포넌트를 사용합니다.

KFP SDK 사용하여 코드에서 파이프 라인을 컴파일하고, 바로 실행하였습니다.

@dsl.pipeline(
    name='My multiply component pipeline',
    description='A pipeline with my component.'
)
def multiply_pipeline():
    numbers = number_op()
    multiply_task = multiply_op(
        input_1=numbers.output,
        parameter_1='6',
    )
    print_op(multiply_task.outputs['output_1'])

multiply_op()의 출력은 multiply_task.outputs [ ‘output_1’] 형식으로 사용할 수 있습니다.

다음은 파이프라인을 구성하고 실행하는 전체 코드입니다.

multiply_pipeline.py

import os
import kfp
from kfp import dsl



def number_op():
    return dsl.ContainerOp(
        name='Generate numbers',
        image='python:alpine3.6',
        command=['sh', '-c'],
        arguments=['python -c "print(\\'1\\\\n2\\\\n3\\\\n4\\\\n5\\\\n6\\\\n7\\\\n8\\\\n9\\\\n10\\')" | tee /tmp/output'],
        file_outputs={'output': '/tmp/output'}
    )


def print_op(msg):
    return dsl.ContainerOp(
        name='Print',
        image='alpine:3.6',
        command=['echo', msg],
    )


component_root = './multiply'
multiply_op = kfp.components.load_component_from_file(os.path.join(component_root, 'component.yaml'))
# multiply_op = kfp.components.load_component_from_url('https://....../component.yaml')

@dsl.pipeline(
    name='My multiply component pipeline',
    description='A pipeline with my component.'
)
def multiply_pipeline():
    numbers = number_op()
    multiply_task = multiply_op(
        input_1=numbers.output,
        parameter_1='6',
    )
    print_op(multiply_task.outputs['output_1'])


if __name__ == '__main__':
    kfp.compiler.Compiler().compile(multiply_pipeline, __file__ + '.zip')

    client = kfp.Client()
    my_experiment = client.create_experiment(name='Basic Experiment')
    my_run = client.run_pipeline(my_experiment.id, 'Multiply component pipeline', __file__ + '.zip')

파이프 라인 실행 결과 확인하기

KFP UI 화면에서 결과를 조회해 볼 수 있습니다. 왼쪽 메뉴에서 Experiments 클릭하면 Experiment 목록을 조회할 수 있습니다. 예제에서 사용한 Experiment 이름이 “Sample Experiment” 이므로, Sample Experiment를 클릭하면 실행 목록을 확인 할 수 있습니다.

“Run name”이 “Multiply component pipeline” 인 것을 클릭하면 실행 결과를 확인할 수 있습니다.

Kubeflow Pipelines – DSL 이해하기 #2

Retry를 사용하는 파이프 라인 만들기

파이프 라인을 구성하고 실행하기

Retry를 사용하는 파이프 라인을 만들어 보겠습니다. Retry를 사용하면 작업이 실패로 끝났을 때, 재시도 할 수 있도록 할 수 있습니다. 재시도 횟수는 사용자가 설정할 수 있습니다.

입력받은 시스템 종료 코드들 중에서 하나를 랜덤으로 선택해서 반환합니다.

def flip_coin_op():
    return dsl.ContainerOp(
        name='Flip coin',
        image='python:alpine3.6',
        command=['sh', '-c'],
        arguments=['python -c "import random; result = \\'heads\\' if random.randint(0,1) == 0 '
                  'else \\'tails\\'; print(result)" | tee /tmp/output'],
        file_outputs={'output': '/tmp/output'}
    )

DSL을 사용하여 파이프 라인을 구성합니다. set_retry() 사용해서, 해당 작업이 실패했을 경우 재시작 하도록 설정하였습니다.

KFP SDK 사용하여 코드에서 파이프 라인을 컴파일하고, 바로 실행하였습니다.

@dsl.pipeline(
    name='Retry random failures',
    description='A pipeline with retry.'
)
def retry_pipeline():
    random_exit_op('0,1,2,3').set_retry(10)

다음은 파이프라인을 구성하고 실행하는 전체 코드입니다.

retry.py

import kfp
from kfp import dsl


def random_exit_op(exit_codes):
    return dsl.ContainerOp(
        name='random_failure',
        image='python:alpine3.6',
        command=['python', '-c'],
        arguments=['import random; import sys; exit_code = int(random.choice(sys.argv[1].split(","))); print(exit_code); sys.exit(exit_code)', exit_codes]
    )


@dsl.pipeline(
    name='Retry random failures',
    description='A pipeline with retry.'
)
def retry_pipeline():
    random_exit_op('0,1,2,3').set_retry(10)


if __name__ == '__main__':
    kfp.compiler.Compiler().compile(retry_pipeline, __file__ + '.zip')

    client = kfp.Client()
    my_experiment = client.create_experiment(name='Basic Experiment')
    my_run = client.run_pipeline(my_experiment.id, 'Retry pipeline', __file__ + '.zip')

파이프 라인 실행 결과 확인하기

KFP UI 화면에서 결과를 조회해 볼 수 있습니다. 왼쪽 메뉴에서 Experiments 클릭하면 Experiment 목록을 조회할 수 있습니다. 예제에서 사용한 Experiment 이름이 “Basic Experiment” 이므로, Basic Experiment를 클릭하면 실행 목록을 확인 할 수 있습니다.

“Run name”이 “Retry pipeline” 인 것을 클릭하면 실행 결과를 확인할 수 있습니다.


쿠버네티스 리소스를 사용하는 파이프 라인 만들기

쿠버네티스 리소스를 사용하는 파이프 라인을 만들어 보겠습니다. 사용자는 쿠버네티스의 모든 리소스를 사용할 수 있습니다.

kfp.dsl.ResourceOp

이 클래스는 쿠버네티스 리소스를 사용하는 단계를 나타냅니다. Argo의 리소스 템플릿을 구현하고 있습니다.

classkfp.dsl.ResourceOp(k8s_resource=None, action: str = 'create', merge_strategy: str = None, success_condition: str = None, failure_condition: str = None, attribute_outputs: Dict[str, str] = None, **kwargs)

사용자는 이 기능을 이용하여 쿠버네티스 리소스의 일부 작업(get, create, apply, delete, replace, place)을 수행할 수 있습니다. 그리고 해당 작업을 수행하는 단계의 성공 또는 실패 했는지를 조건을 통해서 설정할 수 있습니다.

전달 인자

다음은 VolumeOp에서 사용하는 주요 전달 인자입니다.

  • k8s_resource: 쿠버네티시 리소스를 정의한 것입니다. (필수)
  • action: 수행할 작업의 종류입니다. 기본값은 create 입니다.
  • merge_strategy: 수행할 작업의 종류가 patch 일 때 사용할 병합 전략입니다. (선택)
  • success_condition: 단계의 성공을 판별하는 조건입니다.(선택)
  • failure_condition: 단계의 실패를 판별하는 조건입니다. (선택)
  • attribute_outputs: [kfp.dsl.ContainerOp](<https://kubeflow-pipelines.readthedocs.io/en/latest/source/kfp.dsl.html#kfp.dsl.ContainerOp>)file_outputs 과 비슷합니다 . 출력 파리미터 이름을 쿠버네티스 객체의 JSON 경로에 매핑합니다.

출력

ResourceOps는 출력 파라미터를 생성 할 수 있습니다. 사용하는 쿠버네티스 리소스의 필드 값을 출력 할 수 있습니다. 예를 들면 다음과 같습니다.

job = kubernetes_client.V1Job(...)

rop = kfp.dsl.ResourceOp(
    name="create-job",
    k8s_resource=job,
    action="create",
    attribute_outputs={"name": "{.metadata.name}"}
)

기본적으로 ResourceOps는 리소스 이름과 리소스 사양을 출력합니다.

파이프 라인을 구성하고 실행하기

생성할 리소스의 매니페스트를 정의 하였습니다.

_job_manifest = """
{
    "apiVersion": "batch/v1",
    "kind": "Job",
    "metadata": {
        "generateName": "kfp"
    },
    "spec": {
        "template": {
            "metadata": {
                "name": "resource-pipeline"
            },
            "spec": {
                "containers": [{
                    "name": "mnist",
                    "image": "kangwoo/mnist-simple:job",
                    "command": ["python", "/app/mnist-simple.py"]
                }],
                "restartPolicy": "Never"
            }
        }   
    }
}
"""

DSL을 사용하여 파이프 라인을 구성합니다. dsl.ResourceOp() 사용해서, 쿠버네티스 리소소스를 생성하였습니다.

KFP SDK 사용하여 코드에서 파이프 라인을 컴파일하고, 바로 실행하였습니다.

@dsl.pipeline(
    name='Kubernetes Resource',
    description='A pipeline with resource.'
)
def resource_pipeline():
    op = dsl.ResourceOp(
        name='resource-job',
        k8s_resource=json.loads(_job_manifest),
        action='create'
    )

다음은 파이프라인을 구성하고 실행하는 전체 코드입니다.

resource.py

import kfp
from kfp import dsl
import json


_job_manifest = """
{
    "apiVersion": "batch/v1",
    "kind": "Job",
    "metadata": {
        "generateName": "kfp"
    },
    "spec": {
        "template": {
            "metadata": {
                "name": "resource-pipeline"
            },
            "spec": {
                "containers": [{
                    "name": "mnist",
                    "image": "kangwoo/mnist-simple:job",
                    "command": ["python", "/app/mnist-simple.py"]
                }],
                "restartPolicy": "Never"
            }
        }   
    }
}
"""

@dsl.pipeline(
    name='Kubernetes Resource',
    description='A pipeline with resource.'
)
def resource_pipeline():
    op = dsl.ResourceOp(
        name='resource-job',
        k8s_resource=json.loads(_job_manifest),
        action='create'
    )


if __name__ == '__main__':
    kfp.compiler.Compiler().compile(resource_pipeline, __file__ + '.zip')

    client = kfp.Client()
    my_experiment = client.create_experiment(name='Basic Experiment')
    my_run = client.run_pipeline(my_experiment.id, 'Resource pipeline', __file__ + '.zip')

파이프 라인 실행 결과 확인하기

KFP UI 화면에서 결과를 조회해 볼 수 있습니다. 왼쪽 메뉴에서 Experiments 클릭하면 Experiment 목록을 조회할 수 있습니다. 예제에서 사용한 Experiment 이름이 “Basic Experiment” 이므로, Basic Experiment를 클릭하면 실행 목록을 확인 할 수 있습니다.

“Run name”이 “Resource pipeline” 인 것을 클릭하면 실행 결과를 확인할 수 있습니다.


퍼시스턴스 볼륨를 사용하는 파이프 라인 만들기

쿠버네티스 퍼시스턴스 볼륨을 사용하는 파이프 라인을 만들어 보겠습니다. dsl.VolumeOp() 이용하여 퍼시스턴스 볼륨을 쉽게 생성할 수 있습니다.

dsl.VolumeOp

VolumeOp 클래스는 ResourceOp 에서 확장되었습니다. 이 클래스를 사용하면 퍼시스턴스 볼륨을 쉽게 생성할 수 있습니다.

전달 인자

다음은 VolumeOp에서 사용하는 주요 전달 인자입니다.

  • resource_name: 생성할 리소스에 사용할 이름입니다. 이 문자열 앞에 워크플로우 이름이 붙습니다. (필수)
  • size: 요청할 볼륨의 크기입니다. (필수)
  • storage_class: 사용할 스토리지 클래스입니다. (선택)
  • modes: 퍼시스턴스 볼륨의 접근 모드( accessModes) 입니다.기본 값을  VOLUME_MODE_RWM 입니다.
    • VOLUME_MODE_RWO["ReadWriteOnce"]
    • VOLUME_MODE_RWM["ReadWriteMany"]
    • VOLUME_MODE_ROM["ReadOnlyMany"]

출력

쿠버네티스 리소스의 이름과 사양 이외에도 바인딩된 퍼시스턴스 볼륨의 스토리지 크기를 step.outputs [“size”] 로 출력합니다. 하지만 스토리지 제공자가 WaitForFirstConsumer 바인딩 모드를 지원하는 경우, 비어 있을 수 있습니다. 이 값은 비어 있지 않으면, 항상 요청된 크기보다 크거나 같습니다.

파이프 라인을 구성하고 실행하기

dsl.VolumeOp() 이용하여 퍼시스턴스 볼륨을 생성할 수 있습니다.

vop = dsl.VolumeOp(
        name="pipeline-volume",
        resource_name="pipeline-pvc",
        modes=dsl.VOLUME_MODE_RWO,
        size="100Mi"
    )

DSL을 사용하여 파이프 라인을 구성합니다. dsl.dsl.ContainerOp() 의 pvolumes 파라미터를 이용하여 볼륨을 마운트 할 수 있습니다.

step1 에서는 dsl.VolumeOp()으로 생성한 볼륨을 마운트하였고, step2에서는 step1에 마운트된 볼륨을 그대로 다시 마운트 하였습니다.

KFP SDK 사용하여 코드에서 파이프 라인을 컴파일하고, 바로 실행하였습니다.

@dsl.pipeline(
    name='Volume pipeline',
    description='A pipeline with volume.'
)
def volume_pipeline():
    vop = dsl.VolumeOp(
        name="pipeline-volume",
        resource_name="pipeline-pvc",
        modes=dsl.VOLUME_MODE_RWO,
        size="100Mi"
    )

    step1 = dsl.ContainerOp(
        name='Flip coin',
        image='python:alpine3.6',
        command=['sh', '-c'],
        arguments=['python -c "import random; result = \\'heads\\' if random.randint(0,1) == 0 '
                   'else \\'tails\\'; print(result)" | tee /data/output'],
        pvolumes={"/data": vop.volume}
    )

    step2 = dsl.ContainerOp(
        name='Print',
        image='alpine:3.6',
        command=['cat', '/data/output'],
        pvolumes={"/data": step1.pvolume}
    )

다음은 파이프라인을 구성하고 실행하는 전체 코드입니다.

volume.py

import kfp
from kfp import dsl


@dsl.pipeline(
    name='Volume pipeline',
    description='A pipeline with volume.'
)
def volume_pipeline():
    vop = dsl.VolumeOp(
        name="pipeline-volume",
        resource_name="pipeline-pvc",
        modes=dsl.VOLUME_MODE_RWO,
        size="100Mi"
    )

    step1 = dsl.ContainerOp(
        name='Flip coin',
        image='python:alpine3.6',
        command=['sh', '-c'],
        arguments=['python -c "import random; result = \\'heads\\' if random.randint(0,1) == 0 '
                   'else \\'tails\\'; print(result)" | tee /data/output'],
        pvolumes={"/data": vop.volume}
    )

    step2 = dsl.ContainerOp(
        name='Print',
        image='alpine:3.6',
        command=['cat', '/data/output'],
        pvolumes={"/data": step1.pvolume}
    )


if __name__ == '__main__':
    kfp.compiler.Compiler().compile(volume_pipeline, __file__ + '.zip')

    client = kfp.Client()
    my_experiment = client.create_experiment(name='Basic Experiment')
    my_run = client.run_pipeline(my_experiment.id, 'Volume pipeline', __file__ + '.zip')

파이프 라인 실행 결과 확인하기

KFP UI 화면에서 결과를 조회해 볼 수 있습니다. 왼쪽 메뉴에서 Experiments 클릭하면 Experiment 목록을 조회할 수 있습니다. 예제에서 사용한 Experiment 이름이 “Basic Experiment” 이므로, Basic Experiment를 클릭하면 실행 목록을 확인 할 수 있습니다.

“Run name”이 “Volume pipeline” 인 것을 클릭하면 실행 결과를 확인할 수 있습니다.


파이프라인에 환경 변수 사용하기

Kubeflow piepelins 에서 환경 변수를 설정하고 사용하는 방법에 대해서 알아 보겠습니다. 컴포넌트에서 환경 변수를 설정하려면, dsl.ContainerOp()의 add_env_variable() 메소드를 사용하면 됩니다. kubernetes.client.models 패키지에 있는 V1EnvVar 인스턴스를 생성한 후, add_env_variable() 메소드를 이용하여 환경 변수를 컴포넌트에 추가할 수 있습니다.

파이프 라인을 구성하고 실행하기

생성할 리소스의 매니페스트를 정의 하였습니다.

import kfp.dsl as dsl
from kubernetes.client.models import V1EnvVar

@dsl.pipeline(
  name='Env example',
  description='A pipline showing how to use environment variables'
)
def environment_pipeline():
  env_var = V1EnvVar(name='example_env', value='env_variable')
 
  container_op = logg_env_function_op().add_env_variable(env_var)

더 많은 환경 변수를 컴포넌트에 전달하려면 add_env_variable () 더 추가하면 됩니다.

컴포넌트에 추가한 환경 변수를 출력하기 위하여 echo를 사용하였습니다.

def print_env_op():
    return dsl.ContainerOp(
        name='Print',
        image='alpine:3.6',
        command=['sh', '-c', 'echo $example_env'],
    )

다음은 파이프라인을 구성하고 실행하는 전체 코드입니다.

enviroment_variables.py

import kfp
from kfp import dsl

import kfp.dsl as dsl
from kubernetes.client.models import V1EnvVar


def print_env_op():
    return dsl.ContainerOp(
        name='Print',
        image='alpine:3.6',
        command=['sh', '-c', 'echo $example_env'],
    )


@dsl.pipeline(
  name='Env example',
  description='A pipline showing how to use environment variables'
)
def environment_pipeline():
    env_var = V1EnvVar(name='example_env', value='env_variable')

    print_env_op().add_env_variable(env_var)


if __name__ == '__main__':
    kfp.compiler.Compiler().compile(environment_pipeline, __file__ + '.zip')

    client = kfp.Client()
    my_experiment = client.create_experiment(name='Sample Experiment')
    my_run = client.run_pipeline(my_experiment.id, 'Environment pipeline', __file__ + '.zip')

파이프 라인 실행 결과 확인하기

KFP UI 화면에서 결과를 조회해 볼 수 있습니다. 왼쪽 메뉴에서 Experiments 클릭하면 Experiment 목록을 조회할 수 있습니다. 예제에서 사용한 Experiment 이름이 “Sample Experiment” 이므로, Sample Experiment를 클릭하면 실행 목록을 확인 할 수 있습니다.

“Run name”이 “Environment pipeline” 인 것을 클릭하면 실행 결과를 확인할 수 있습니다.


ContainerOp에 사이드카 추가하기

파이프 라인을 구성하고 실행하기

dsl.Sidecar()를 이용하여 사이드카를 생성합니다.

	echo = dsl.Sidecar(
        name="echo",
        image="hashicorp/http-echo:latest",
        args=['-text="hello world"'],
    )

dsl.ContainerOp()의 sidcars 파라미터를 이용하여 생성한 사이드카를 추가합니다.

	op1 = dsl.ContainerOp(
        name="download",
        image="busybox:latest",
        command=["sh", "-c"],
        arguments=[
            "sleep %s; wget localhost:5678 -O /tmp/results.txt" % sleep_sec
        ],  # sleep for X sec and call the sidecar and save results to output
        sidecars=[echo],
        file_outputs={"downloaded": "/tmp/results.txt"},
    )

다음은 파이프라인을 구성하고 실행하는 전체 코드입니다.

sidecar.py

import kfp
import kfp.dsl as dsl

@dsl.pipeline(
    name="pipeline_with_sidecar", 
    description="A pipeline that demonstrates how to add a sidecar to an operation."
)
def pipeline_with_sidecar(sleep_sec: int = 30):

    # sidecar with sevice that reply "hello world" to any GET request
    echo = dsl.Sidecar(
        name="echo",
        image="hashicorp/http-echo:latest",
        args=['-text="hello world"'],
    )

    # container op with sidecar
    op1 = dsl.ContainerOp(
        name="download",
        image="busybox:latest",
        command=["sh", "-c"],
        arguments=[
            "sleep %s; wget localhost:5678 -O /tmp/results.txt" % sleep_sec
        ],  # sleep for X sec and call the sidecar and save results to output
        sidecars=[echo],
        file_outputs={"downloaded": "/tmp/results.txt"},
    )

    op2 = dsl.ContainerOp(
        name="echo",
        image="library/bash",
        command=["sh", "-c"],
        arguments=["echo %s" % op1.output],  # print out content of op1 output
    )

if __name__ == '__main__':
    kfp.compiler.Compiler().compile(pipeline_with_sidecar, __file__ + '.yaml')

파이프 라인 실행 결과 확인하기

KFP UI 화면에서 결과를 조회해 볼 수 있습니다. 왼쪽 메뉴에서 Experiments 클릭하면 Experiment 목록을 조회할 수 있습니다. 예제에서 사용한 Experiment 이름이 “Sample Experiment” 이므로, Sample Experiment를 클릭하면 실행 목록을 확인 할 수 있습니다.

“Run name”이 “Sidecar pipeline” 인 것을 클릭하면 실행 결과를 확인할 수 있습니다.



파이프 라인의 기본 Artifact 저장소를 변경하기

KFP 는 파이프 라인에서 사용하는 아티팩트들은 내부에 설치된 minio에 저장하고 있습니다. 아티팩트의 기본 저장소를 바꾸려면 argo workflow의 workflow-controller configmap을 수정하면 됩니다.

다음 명령어를 실행하면 아티팩트 저장소의 설정 정보를 수정할 수 있습니다.

kubectl -n kubeflow edit configmap workflow-controller-configmap

다음은 아티팩트 저장소의 설정 정보입니다.

...
data:
  config: |
    {
    artifactRepository:
    {
        s3: {
            bucket: mlpipeline,
            keyPrefix: artifacts,
            endpoint: minio-service.kubeflow:9000,
            insecure: true,
            accessKeySecret: {
                name: mlpipeline-minio-artifact,
                key: accesskey
            },
            secretKeySecret: {
                name: mlpipeline-minio-artifact,
                key: secretkey
            }
        }
    }
    }
...

자세한 사항은 https://github.com/argoproj/argo/blob/master/docs/configure-artifact-repository.md 를 참고 하실 수 있습니다.