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 화면입니다.