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” 탭에서도 메트릭을 확인 할 수 있습니다.

댓글 남기기

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