작성일자

Kubeflow Pipelines – DSL 이해하기 #1

순차적 처리를 하는 파이프 라인 만들기

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

순차적으로 작업을 하는 파이프 라인을 만들어 보겠습니다. 동전을 던지는 단계와 그 결과를 출력하는 단계로 구성 되어 있습니다.

flip_coin_op() 는 동전을 던지는 단계입니다. 파이썬 코드로 작성되었습니다. 랜덤 함수를 사용해서 0이면 앞면(heads), 1이면 뒷면(tails) 이라고 출력합니다. 그리고 출력 결과를 /tmp/output 파일에 저장합니다. 결과가 저장된 /tmp/output 파일을 다음 단계에서 사용하기 위해 file_outputs 으로 정의합니다.

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'}
    )

print_op(msg) 라는 입력받은 메시지를 화면으로 출력하는 단계입니다. 단순히 입력 받은 메시지를 echo 명령어를 사용하여 화면에 출력합니다.

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

DSL을 사용하여 파이프 라인을 구성합니다. flip_coin_op()가 먼저 실행되고, 그 다음 print_op()가 실행됩니다. 그리고 flip_coin_op()의 출력 결과를 print_op()의 입력 파라미터로 사용하고 있습니다.

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

@dsl.pipeline(
    name='Sequential pipeline',
    description='A pipeline with two sequential steps.'
)
def sequential_pipeline():
    """A pipeline with two sequential steps."""

    flip = flip_coin_op()
    print_op(flip.output)


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

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

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

sequential.py

import kfp
from kfp import dsl


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'}
    )


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


@dsl.pipeline(
    name='Sequential pipeline',
    description='A pipeline with two sequential steps.'
)
def sequential_pipeline():
    """A pipeline with two sequential steps."""

    flip = flip_coin_op()
    print_op(flip.output)


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

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

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

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

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

Graph 탭에서 flip-coin 단계를 선택하고, Input/Output 탭을 클릭하면 출력 결과를 볼 수 있습니다.

조건에 의해서 분기를 하는 파이프 라인 만들기

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

조건에 의해서 분기를 파이프 라인을 만들어 보겠습니다. 동전을 던지는 단계와 그 결과에 따러 “승리” 또는 “패배” 를 출력하는 단계로 구성 되어 있습니다.

flip_coin_op() 는 동전을 던지는 단계입니다. 이전 예제 코드와 동일합니다.

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'}
    )

print_op(msg) 라는 입력받은 메시지를 화면으로 출력하는 단계입니다. 이전 예제 코드와 동일합니다.

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

DSL을 사용하여 파이프 라인을 구성합니다. flip_coin_op()가 먼저 실행되고, 그 결과에 따라서 print_op()를 사용해서 “승리” 또는 “패배”를 출력하게 됩니다. 동전이 앞면인지 뒷면이지 판단하기 위해서 dsl.Condition() 을 사용하였습니다. 앞면이면 ‘YOUT WIN’을 출력하고, 뒷면이면 ‘YOU LOSE’를 출력합니다.

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

@dsl.pipeline(
    name='Conditional execution pipeline',
    description='Shows how to use dsl.Condition().'
)
def condition_pipeline():
    flip = flip_coin_op()
    with dsl.Condition(flip.output == 'heads'):
        print_op('YOUT WIN')

    with dsl.Condition(flip.output == 'tails'):
        print_op('YOU LOSE')


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

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

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

condition.py

import kfp
from kfp import dsl


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'}
    )


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


@dsl.pipeline(
    name='Conditional execution pipeline',
    description='Shows how to use dsl.Condition().'
)
def condition_pipeline():
    flip = flip_coin_op()
    with dsl.Condition(flip.output == 'heads'):
        print_op('YOUT WIN')

    with dsl.Condition(flip.output == 'tails'):
        print_op('YOU LOSE')


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

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

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

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

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

병렬 루프를 사용하는 파이프 라인 만들기

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

정적 항목 세트에 대한 병렬 루프를 사용하는 파이프 라인을 만들어 보겠습니다. 동전을 여러번 던지고, 그 결과를 json 형식으로 저장 단계와 그 결과들을 병렬 루프를 통해서 출력하는 단계로 구성 되어 있습니다.

flip_coins_op() 는 동전을 여러번 던지는 단계입니다. 동전을 던진 결과 값을 json 형식으로 저정합니다.

def flip_coins_op():
    return dsl.ContainerOp(
        name='Flip coin',
        image='python:alpine3.6',
        command=['sh', '-c'],
        arguments=['python -c "import json; import sys; import random; '
                   'json.dump([(\\'heads\\' if random.randint(0,1) == 1 else \\'tails\\') for i in range(10)], '
                   'open(\\'/tmp/output.json\\', \\'w\\'))"'],
        file_outputs={'output': '/tmp/output.json'}
    )

print_op(msg) 라는 입력받은 메시지를 화면으로 출력하는 단계입니다. 이전 예제 코드와 동일합니다.

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

DSL을 사용하여 파이프 라인을 구성합니다. flip_coins_op()가 먼저 실행되고, 그 결과가 json 형식으로 반환됩니다. json에 포함된 아이템의 개수 만큼 print_op() 를 실행합니다.

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

@dsl.pipeline(
    name='Loop pipeline',
    description='A pipeline with parallel loop'
)
def loop_pipeline():
    flips = flip_coins_op()
    with dsl.ParallelFor(flips.output) as item:
        print_op(item)

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

loop.py

import kfp
from kfp import dsl


def flip_coins_op():
    return dsl.ContainerOp(
        name='Flip coin',
        image='python:alpine3.6',
        command=['sh', '-c'],
        arguments=['python -c "import json; import sys; import random; '
                   'json.dump([(\\'heads\\' if random.randint(0,1) == 1 else \\'tails\\') for i in range(10)], '
                   'open(\\'/tmp/output.json\\', \\'w\\'))"'],
        file_outputs={'output': '/tmp/output.json'}
    )



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


@dsl.pipeline(
    name='Loop pipeline',
    description='A pipeline with parallel loop'
)
def loop_pipeline():
    flips = flip_coins_op()
    with dsl.ParallelFor(flips.output) as item:
        print_op(item)


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

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

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

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

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

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

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

Exit Handler 사용하는 파이프 라인을 만들어 보겠습니다. Exit Handler를 사용하면 작업 그룹이 종료 될 때 지정한 작업을 실행시킬 수 있습니다. 작업 그룹의 성공 여부와는 상관 없이 무조건 지정한 작업이 실행됩니다.

동전을 여러번 던지고, 그 결과를 json 형식으로 저장 단계와 그 결과들을 병렬 루프를 통해서 출력하는 단계로 구성 되어 있습니다.

flip_coin_op() 는 동전을 던지는 단계입니다. 이전 예제 코드와 동일합니다.

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'}
    )

print_op(msg) 라는 입력받은 메시지를 화면으로 출력하는 단계입니다. 이전 예제 코드와 동일합니다.

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

DSL을 사용하여 파이프 라인을 구성합니다. dsl.ExitHandler()를 사용해서, 작업 그룹이 종료될때 실행할 작업을 지정할 수 있습니다. 작업 그룹은 flip_coin_op() 과 print_op(flip.output) 으로 구성되어 있습니다. 작업 그룹에 속한 작업들이 종료되면 dsl.ExitHandler(exit_op)에 지정한 exit_op 작업이 실행됩니다.

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

@dsl.pipeline(
    name='Exit handler',
    description=' The exit handler will run after the pipeline finishes (successfully or not)'
)
def sequential_pipeline():
    exit_op = print_op('Exit')

    with dsl.ExitHandler(exit_op):
        flip = flip_coin_op()
        print_op(flip.output)

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

exit_handler.py

import kfp
from kfp import dsl


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'}
    )


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


@dsl.pipeline(
    name='Exit handler',
    description=' The exit handler will run after the pipeline finishes (successfully or not)'
)
def sequential_pipeline():
    exit_op = print_op('Exit')

    with dsl.ExitHandler(exit_op):
        flip = flip_coin_op()
        print_op(flip.output)


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

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

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

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

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

파이프 라인 파라미터 사용하기

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

파라미터를 입력 받아서, 파이프 라인을 실행할 수 있는 파이프 라인을 만들어 보겠습니다.

파이프 라인을 실행할 때 앞면(heads) 또는 뒷면(tails)을 파라미터로 입력받아서, 동전을 던진 결과와 같으면 승링를 다르면 패배를 출력하게 하였습니다.

flip_coin_op() 는 동전을 던지는 단계입니다. 이전 예제 코드와 동일합니다.

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'}
    )

print_op(msg) 라는 입력받은 메시지를 화면으로 출력하는 단계입니다. 이전 예제 코드와 동일합니다.

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

DSL을 사용하여 파이프 라인을 구성합니다. 파이프 라인 함수에 predict 파라미터를 입력 받을 수 있도록 추가하였습니다. 파이프 라인 함수가 호출되면 각 파라미터는 PipelineParam 객체가 됩니다.

@dsl.pipeline(
    name='Pipeline parameters',
    description='Pipeline parameters'
)
def condition_pipeline(
        predict : str = 'heads'):
    flip = flip_coin_op()
    with dsl.Condition(flip.output == predict):
        print_op('YOU WIN')

    with dsl.Condition(flip.output != predict):
        print_op('YOU LOSE')

KFP SDK 사용하여 코드에서 파이프 라인을 컴파일하고, 바로 실행하였습니다. 파이프 라인 실행 시 파라미터를 넘기기 위해서, params 파라미터를 사용하였습니다.

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

    client = kfp.Client()
    my_experiment = client.create_experiment(name='Basic Experiment')
    my_run = client.run_pipeline(my_experiment.id, 'Pipeline parameters', __file__ + '.zip', params={'predict' : 'tails'})

파라미터는 Kubeflow Pipelines UI를 통해서 실행할 때도 입력할 수 있습니다.

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

parameters.py

import kfp
from kfp import dsl
from kfp.dsl import PipelineParam


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'}
    )


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


@dsl.pipeline(
    name='Pipeline parameters',
    description='Pipeline parameters'
)
def condition_pipeline(
        predict : str = 'heads'):
    flip = flip_coin_op()
    with dsl.Condition(flip.output == predict):
        print_op('YOU WIN')

    with dsl.Condition(flip.output != predict):
        print_op('YOU LOSE')


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

    client = kfp.Client()
    my_experiment = client.create_experiment(name='Basic Experiment')
    my_run = client.run_pipeline(my_experiment.id, 'Pipeline parameters', __file__ + '.zip', params={'predict' : 'tails'})

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

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

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

작성일자

Kubeflow Pipelines – SDK를 사용해서 파이프라인 만들기

Kubeflow Pipelines SDK를 사용하여 파이프라인과 컴포넌트를 구성하고, 빌드하는 방법에 대해서 알아보겠습니다. 그리고 SDK를 사용하여 파이프라인을 실행하는 방법과, Kubeflow Pipelines UI를 사용하여 파이프라인을 실행하는 방법에 대해서 알아보겠습니다.

Pipelines SDK 소개

Kubeflow Pipelines SDK는 머신 러닝 워크 플로우를 정의하고, 실행시킬 수 있는 파이썬 패키지 세트를 제공합니다. 파이프 라인은 워크 플로우의 단계를 구성하는 컴포넌트들과, 각 컴포넌트들이 서로 상호 작용하는 방식을 정의해 놓은 것입니다.

Kubeflow Pipelines SDK는 파이프 라인을 컴파일하고 실행하는 등의 여러 상호 작용 기능을 제공하고 있습니다. 그리고 파이프 라인의 구성 요소인 컴포넌트를 만들고 로드 하는 등의 기능도 제공하고 있습니다. 컴포넌트에서 사용할 컨테이너 이미지를 빌드 하는 기능도 제공하고 있습니다. 다만 Kubeflow Pipelines SDK에 포함되어 있는 컨테이너 빌더 기능은 Google Cloud Platform (GCP) 환경에서만 원활하게 사용할 수 있습니다.

SDK 패키지

Kubeflow Pipelines SDK에는 다음과 같은 패키지가 포함되어 있습니다.

  • kfp.compiler : 파이프 라인을 컴파일 할 수 있는 기능을 제공하고 있습니다. 이 패키지의 주요 사용 메소드는 다음과 같습니다
    • kfp.compiler.Compiler.compile : Python DSL 코드를 Kubeflow Pipelines 서비스가 처리 할 수 있는 단일 정적 구성 (YAML 형식)으로 컴파일합니다. Kubeflow Pipelines 서비스는 정적 구성을 실행을 위해 Kubernetes 리소스 세트로 변환합니다. (현재는 컴파일하면 Argo Workflows 형태로 변환합니다.)
  • kfp.component : 파이프 라인 컴포넌트와 상호 작용하기 위한 기능을 제공하고 있습니다. 이 패키지의 주요 사용 메소드는 다음과 같습니다.
    • kfp.components.func_to_container_op : Python 함수를 파이프 라인 컴포넌트로 변환하고 팩토리 함수를 리턴합니다. 그런 다음 팩토리 함수를 호출하여 컨테이너에서 원래 함수를 실행하는 파이프 라인 태스크 (ContainerOp)의 인스턴스를 구성 할 수 있습니다.
    • kfp.components.load_component_from_file : 파일에서 파이프 라인 컴포넌트를 로드하고 팩토리 함수를 리턴합니다. 그런 다음 팩토리 함수를 호출하여 컴포넌트 컨테이너 이미지를 실행하는 파이프 라인 태스크 (ContainerOp)의 인스턴스를 구성 할 수 있습니다.
    • kfp.components.load_component_from_url : URL에서 파이프 라인 컴포넌트를 로드하고 팩토리 함수를 리턴합니다. 그런 다음 팩토리 함수를 호출하여 컴포넌트 컨테이너 이미지를 실행하는 파이프 라인 태스크 (ContainerOp)의 인스턴스를 구성 할 수 있습니다.
  • kfp.containers : 컴포넌트 컨테이너 이미지를 빌드하는 기능을 제공하고 있습니다. 이 패키지의 주요 사용 메소드는 다음과 같습니다
    • build_image_from_working_dir : 파이썬 작업 디렉토리를 사용하여 새 컨테이너 이미지를 빌드하고 푸시합니다. Python 컨테이너 이미지를 기본 이미지로 사용하는 Dockerfile을 생성하고, requirements.txt 파일 있는 경우 패키지를 설치하고 대상 Python 파일을 컨테이너 이미지에 복사합니다. 작업 디렉토리의 루트에 사용자 정의 Dockerfile을 만들어서 대체 할 수 있습니다. (현재는 Google Cloud Platform (GCP) 환경에서만 사용할 수 있습니다.)
  • kfp.dsl : 파이프 라인 및 컴포넌트를 정의하고 상호 작용하는 데 사용할 수있는 DSL (Domain-Specific Language)이 포함되어 있습니다. 이 패키지의 주요 사용 메소드는 다음과 같습니다.
    • kfp.dsl.ContainerOp : 컨테이너 이미지로 구현 된 파이프 라인 작업을 나타냅니다.
    • kfp.dsl.PipelineParam 한 파이프 라인 컴포넌트에서 다른 파이프 라인 컴포넌트로 전달할 수있는 파이프 라인 파라미터를 나타냅니다.
    • kfp.dsl.component : 파이프 라인 컴포넌트를 반환하는 DSL 함수의 데코레이터입니다. (ContainerOp).
    • kfp.dsl.pipeline : 파이프 라인을 반환하는 Python 함수의 데코레이터입니다.
    • kfp.dsl.python_component: 파이프 라인 컴포넌트 메타 데이터를 함수 객체에 추가하는 Python 함수의 데코레이터입니다.
    • kfp.dsl.types:  Kubeflow Pipelines SDK에서 사용하는 타입들이 정의되어 있습니다. 타입에는 String, Integer, Float 및 Bool과 같은 기본 타입과 GCPProjectID 및 GCRPath와 같은 도메인 별 타입이 있습니다. DSL 정적 유형 검사에 대해서는 안내서를 참조하실 수 있습니다.
    • kfp.dsl.ResourceOp : 쿠버네티스 리소스를 직접 조작할 수 작업을 나타냅니다.(create, get, apply 등 ).
    • kfp.dsl.VolumeOp : 쿠버네티스 PersistentVolumeClaim 을 생성하는 파이프 라인 작업 을 나타냅니다.
    • kfp.dsl.VolumeSnapshotOp : 새로운 볼륨 스냅 샷을 생성하는 파이프 라인 작업을 나타냅니다.
    • kfp.dsl.PipelineVolume : 파이프 라인의 단계간에 데이터를 전달하기 위해 사용하는 볼륨을 나타냅니다.
  • kfp.Client : Kubeflow Pipelines API 용 Python 클라이언트 라이브러리가 포함되어 있습니다. 이 패키지의 주요 사용 메소드는 다음과 같습니다.
    • kfp.Client.create_experiment : 파이프 라인 experiment 을 만들고, experiment  개체를 반환합니다.
    • kfp.Client.run_pipeline 파이프 라인을 실행(run)하고 실행(run) 개체를 반환합니다.
  • KFP extension modules : Kubeflow Pipelines에서 사용할 수 있는 특정 플랫폼에 대한 기능을 가지고 있습니다. 온 프레미스, Google Cloud Platform (GCP), Amazon Web Services (AWS) 및 Microsoft Azure에 대한 유틸리티 기능을 제공하고 있습니다.

KFP CLI tool

KFP CLI 도구를 사용하면 커맨드 라인에서 직접 Kubeflow Pipelines SDK의 일부분을 사용할 수 있습니다. KFP CLI 도구는 다음과 같은 명령을 제공합니다.

  • kfp diagnose_me : 지정된 파라미터로 환경 진단을 실행합니다
    • --json : 명령 실행 결과를 JSON으로 반환하도록 합니다. 별도로 설정하지 않으면, 결과는 사람이 읽을 수 있는 형식으로 반환됩니다.
    • --namespace TEXT : 대상 쿠버네티스 네임스페이스를 지정합니다. 별도로 설정하지 않으면, 모든 네임스페이스를 대상으로 합니다.
  • kfp pipeline <COMMAND> : 파이프 라인을 관리하는 데 도움이 되는 명령을 제공합니다.
    • get : Kubeflow Pipelines 클러스터의 Kubeflow 파이프 라인에 대한 상세한 정보를 조회합니다.
    • list : Kubeflow Pipelines 클러스터에 업로드 된 파이프 라인 목록을 조회 합니다.
    • upload : Kubeflow Pipelines 클러스터에 파이프 라인을 업로드합니다.
  • kfp run <COMMAND> 파이프 라인 실행을 관리하는 데 도움이 되는 명령을 제공합니다.
    • get : 파이프 라인 실행의 상세한 정보를 조회합니다.
    • list : 최근 실행한 파이프 라인 실행 목록을 조회 합니다.
    • submit – 파이프 라인을 실행 시킵니다.

파이프 라인과 컴포넌트 만들기

SDK를 사용하여 파이프 라인과 컴포넌트를 만드는 방법에 대해서 알아 보도록 하겠습니다.

컴포넌트 만드는 방법

파이프라인은 컴포넌트로 구성되어 있습니다. 그래서 파이프라인을 만들기 위해서는 사용할 컴포넌트를 먼저 만들어야합니다. 이미 만들어 놓은 컴포넌트가 있으면 재사용할 수도 있습니다.

컴포넌트를 만드는 단계는 다음과 같습니다.

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

나. 컴포넌트 컨테이너화 : 작성한 프로그램을 컨테이너 이미지로 만들어야 합니다.

다. 컴포넌트 스펙 작성 : 컴포넌트의 데이터 모델을 정의하기 위해서 YAML 형식의 파일을 작성해야 합니다. 재사용 가능한 컴포넌트를 만들때는 스펙을 작성하는 것이 좋지만, 생략 가능합니다. 컴포넌트 스펙 파일이 있는 경우에는 스펙 파일을 로드해서 컴포넌트를 생성할 수 있습니다. 자세한 내용은 “재사용 가능한 컴포넌트”를 참고 하시기 바랍니다.

파이프라인 만드는 방법

컴포넌트를 이용해 파이프라인을 만들 수 있습니다. 파이프라인을 만들기 위해서 파이프라인 파이썬 코드를 작성해야 합니다.

파이프라인을 만드는 단계는 다음과 같습니다.

가. Kubeflow Pipelines DSL을 사용하여 파이프라인 함수와 컴포넌트 함수를 작성합니다.

나. 파이프 라인을 컴파일 하여 압축 된 YAML 파일을 생성합니다.

파이프 라인을 컴파일 하기 위한 방법은 두 가지가 있습니다.

  • kfp.compiler.Compiler.compile 메소드를 사용하는 방법 kfp.compiler.Compiler().compile(my_pipeline, 'my_pipeline.zip')
  • 커맨드 라인에서 dsl-compile 커맨드를 사용하는 방법 dsl-compile --py [path/to/python/file] --output my_pipeline.zip

다. 파이프 라인을 업로드하고, 실행합니다.

파이프 라인을 실행하는 방법은 두 가지가 있습니다.

  • Kubeflow Pipelines SDK 를 사용하는 방법 
client = kfp.Client()
my_experiment = client.create_experiment(name='demo')
my_run = client.run_pipeline(my_experiment.id, 'my-pipeline', 
  'my_pipeline.zip')
  • Kubeflow Pipelines UI를 사용하는 방법

컴포넌트 만들기

프로그램이 포함된 컨테이너 이미지를 사용하여 컴포넌트를 생성하는 방법에 대해서 알아 보도록 하겠습니다. 생성한 컴포넌트는 파이프라인을 작성하는데 사용됩니다.

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

mnist-simple.py

from __future__ import absolute_import, division, print_function, unicode_literals

import tensorflow as tf
import numpy as np


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

    # Reserve 10,000 samples for validation
    x_val = x_train[-10000:]
    y_val = y_train[-10000:]
    x_train = x_train[:-10000]
    y_train = y_train[:-10000]

    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_data=(x_val, y_val))

    print("Average test loss: ", np.average(training_history.history['loss']))


if __name__ == '__main__':
    train()

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

Dockerfile을 생성합니다.

Dockerfile

FROM tensorflow/tensorflow:2.1.0-py3

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

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

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

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

docker push kangwoo/mnist-simple:kfp

SDK를 사용하여 파이프라인을 만들고 실행하기

생성한 컴포넌트를 이용해 파이프라인을 만들어 보겠습니다. 파이프라인을 만들기 위해서 파이프라인 파이썬 코드를 작성해야 합니다.

가. Kubeflow Pipelines DSL을 사용하여 컴포넌트 함수를 작성합니다. image 파라미터에 컴포넌트에서 사용하는 컨테이너 이미지를 정의합니다. 컴포넌트 함수는 kfp.dsl.ContainerOp를 리턴해야합니다. 선택적으로 kfp.dsl.component 라는 데코레이터를 사용하여 DSL 컴파일러에서 정적 타입 검사를 활성화 할 수 있습니다. 데코레이터를 사용하려면 @kfp.dsl.component 어노테이션을 컴포넌트 함수에 추가 하면 됩니다.

@kfp.dsl.component
def train_component_op():
    return kfp.dsl.ContainerOp(
        name='mnist-train',
        image='kangwoo/kfp-mnist:kfp'
    )

나. Kubeflow Pipelines DSL을 사용하여 파이프 라인 함수를 작성합니다. 파이프 라인을 정의하는 함수에 사용할 컴포넌트들을 추가합니다. 파이프 라인 함수에서 파이프 라인을 빌드하기 위해서 kfp.dsl.pipeline 데코레이터를 사용합니다. 데코레이터를 사용하려면 @kfp.dsl.pipeline 어노테이션을 파이프 라인 함수에 추가 하면 됩니다.

@dsl.pipeline(
    name='My pipeline',
    description='My machine learning pipeline'
)
def my_pipeline():
    train_task = train_component_op()

다. 파이프 라인을 컴파일하여 압축 된 YAML 파일을 생성하겠습니다. YAML 파일에는 파이프 라인 실행을 위한 쿠버네티스 리소스들이 정의되어 있습니다. kfp.compiler.Compiler.compile 메소드를 사용하는 컴파일 하겠습니다.

kfp.compiler.Compiler().compile(my_pipeline, 'my_pipeline.zip')

라. 파이프 라인을 업로드하고, 실행합니다. Kubeflow Pipelines SDK 를 사용하여 파이프라인을 업로드하고 실행하겠습니다.

client = kfp.Client()
my_experiment = client.create_experiment(name='Basic Experiment')
my_run = client.run_pipeline(my_experiment.id, 'my_pipeline', 
  'my_pipeline.zip')

다음은 파이프라인 전체 코드 입니다.

import kfp
from kfp import dsl


@kfp.dsl.component
def train_component_op():
    return kfp.dsl.ContainerOp(
        name='mnist-train',
        image='kangwoo/kfp-mnist:kfp'
    )


@dsl.pipeline(
    name='My pipeline',
    description='My machine learning pipeline'
)
def my_pipeline():
    train_task = train_component_op()


if __name__ == '__main__':
    # Compile
    pipeline_package_path = 'my_pipeline.zip'
    kfp.compiler.Compiler().compile(my_pipeline, pipeline_package_path)

    # Run
    client = kfp.Client()
    my_experiment = client.create_experiment(name='Basic Experiment')
    my_run = client.run_pipeline(my_experiment.id, 'my_pipeline', pipeline_package_path)

파이프라인 코드를 실행합니다.

python my_pipeline.py

파이프라인 코드가 실행되면, 파이프라인 패키지가 컴파일 됩니다. 그리고 컴파일된 패키지를 Kubeflow Pipelines 에 전달하여 실행 시킵니다. 실행 결과는 Kubeflow Pipelines UI를 통해서 확인할 수 있습니다.

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

Kubeflow 파이프라인 UI에서 실행 결과를 확인하려면, 다음 절차대로 진행하시면 됩니다.

  1. Kubeflow 대시보드 화면의 왼쪽 메뉴에서 Pipelines를 클릭합니다.
  2. Kubeflow Pipelines UI 화면의 왼쪽 메뉴에서 Experiments 클릭하여, 현재 파이프 라인 실험 목록을 조회합니다.
  3. 보려는 실험(Experiment)의 ‘실험 이름’을 클릭합니다.
  4. 보려는 실행(Run)의 “실행 이름”을 클릭합니다.
  5. Graph 탭에서 보려는 파이프 라인 컴포넌트를 나타내는 단계를 클릭합니다. 단계 세부 사항이 Artifacts  탭을 표시하며 슬라이드 됩니다.
작성일자

Kubeflow – Kubeflow Pipelines 이해하기

Kubeflow Pipelines 살펴보기

실 세계에서 머신 러닝 워크 플로우를 만들고 배포하는 것은 무척이나 어렵고 힘든 일입니다.

Kubeflow Pipelines는 머신 러닝 워크 플로우를 만들고 배포하기 위한 컨테이너 기반의 플랫폼으로서, 사용자가 편리하게 사용할 수 있고, 쉽게 확장이 가능합니다.

Kubeflow Pipelines는 머신 러닝 워크 플로우를 생성하기 위해서 파이프 라인을 정의합니다. 파이프 라인에는 사용하는 컴포넌트들과 작업 처리 규칙이 정의되어 있습니다. Kubeflow Pipelines는 파이프라인 뿐만 아니라 컴포넌트도 쉽게 재사용할 수 있도록 만들어져 있습니다. 그래서 만들어진 컴포넌트나 파이프라인 있다면, 레고를 조립하는 것처럼 쉽게 워크 플로우를 구성할 수 있습니다.

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

아쉽게도 Kubeflow Pipelines 는 아직 멀티 테넌시를 지원하지 않습니다.

Kubeflow Pipelines 목표

Kubeflow Pipelines 가 추구하는 목표는 다음과 같습니다.

  • 엔드 투 엔드 오케스트레이션 : 머신 러닝 파이프 라인의 오케스트레이션을 지원하고 단순화 시킵니다
  • 손쉬운 실험 : 수많은 아이디어와 기술을 쉽게 시도 할 수 있고, 다양한 시험/실험을 관리 할 수 ​​있도록 합니다.
  • 손쉬운 재사용 : 구성 요소 및 파이프 라인을 재사용하여, 매번 재 구축 할 필요 없이 엔드 투 엔드 솔루션을 신속하게 생성 할 수 있도록 합니다.

Kubeflow Pipelines 개념

Kubeflow Pipelines 에서 사용하는 개념에 대해서 알아보겠습니다.

Pipeline

Pipeline은 머신 러닝 워크 플로우에서 사용하는 컴포넌트들과, 해당 컴토넌트들 간의 작업 처리 규칙을 그래프 형태로 정의한 것입니다. Pipeline에는 파이프 라인을 실행하는 데 필요한 입력 매개 변수와 각 컴포넌트의 입력 및 출력에 대한 정의가 포함되어 있습니다.

Pipeline을 실행하면 시스템이 워크 플로우의 단계에 해당하는 하나 이상의 쿠버네티스 포드를 시작합니다. 포드는 컨테이너를 시작하고 컨테이너는 정의된 프로그램을 실행합니다.

파이프 라인을 개발 한 후 Kubeflow Pipelines UI에서 파이프 라인을 업로드하고 실행 할 수 있습니다.

Component

Pipeline의 컴포넌트는 하나의 단계를 수행하는 코드가 모여있는 컨테이너 이미지입니다. 이 컴포넌트들은 파이프 라인의 실행 단계에서 각자의 담당 역할을 수행하게 됩니다. 예를 들어 데이터 전처리, 데이터 변환, 모델 학습 등이 있습니다. 컴포넌트는 입력 및 출력에 대한 정의를 포함하고 있습니다.

Graph

그래프는 Kubeflow Pipelines UI에서 파이프 라인의 런타임 실행을 나타내는 그림입니다. 그래프는 파이프 라인의 실행된 단계나 실행중인 단계를 나타냅니다. 화살표는 각 단계로 표시되는 파이프 라인 컴포넌트 간의 상/하위 관계를 나타냅니다.

파이프 라인이 실행되면 그래프를 볼 수 있습니다. 그래프 안의 각 노드는 파이프 라인의 단계를 나타냅니다.

각 노드의 오른쪽 상단에는 상태, 실행 중, 성공, 실패 또는 건너뜀 상태를 나타내는 아이콘이 있습니다. 조건절이 있을 경우에는 노드를 건너 뛸 수 있습니다.

Experiment

Experiment는 파이프 라인을 실행 할 수 있는 작업 공간입니다. experiment 사용하여 파이프 라인의 실행(run)을 논리적 그룹으로 묶을 수 있습니다. Experiments에는 임의의 실행(run)뿐만 아니라 반복 실행(recurring run)도 포함될 수 있습니다.

Run and Recurring Run

Run은 파이프 라인을 한번 실행 하는 것을 의미합니다. Run은 사용자가 시도하는 실행에 대한 정보를 저장하고 있기 때문에, 재현이 가능합니다. Kubeflow Pipelines UI의 세부 정보 페이지를 보면, 실행 진행률을 볼 수 있습니다. 여기에서 실행의 각 단계에 대한 런타임 그래프, 출력 결과 및 로그를 확인 할 수 있습니다.

Recurring run 은 파이프 라인의 반복 실행을 의미합니다. 반복 실행을 위한 설정에는 파이프 라인에서 사용하는 파라미터와 실행 트리거를 위한 파라미터가 포함되어 있습니다.

모든 Experiment 내에서 반복 실행을 시작할 수 있습니다. 반복 실행이 설정되어 있으면, 주기적으로 파이프 라인을 실행하게 됩니다. Kubeflow Pipelines UI에서 반복 실행을 활성화/비활성화 할 수 있습니다. 동시에 실행되는 최대 실행 개수를 제한하기 위해서, 최대 동시 실행 개수를 지정할 수도 있습니다. 최대 동시 실행 개수는, 파이프 라인의 실행 시간이 오래 걸리면서, 자주 실행되게 트리거 되는 경우에 도움이 될 수 있습니다.

Run Trigger

실행 트리거는 주어진 실행 조건에 의해서 새로운 실행을 시작하기 위하여 시스템에 알리는 역할을 합니다. 다음과 같은 유형의 실행 트리거를 사용할 수 있습니다.

  • Periodic: 주기적인 간격 기반의 실행을 위한 트리거입니다. (예 : 3 시간마다 또는 45 분마다).
  • Cron : 실행 예약을 위한 cron 표현식을 사용하는 트리거입니다.

Step

Step은 파이프 라인의 컴포넌트 중 하나를 실행한다는 것을 의미합니다. 복잡한 파이프 라인에서 컴포넌트는 반복적으로 여러 번 실행될 수 있습니다. 그리고 if 같은 조건문을 사용하여, 조건부로 실행될 수 있습니다.

Output Artifact

Output Artifact 는 파이프 라인 컴포넌트에서 생성한 출력물입니다. 이 출력물은 Kubeflow Pipelines UI를 사용하여 이해하기 쉽게 시각화로 렌더링 할 수 있습니다.

파이프 라인 컴포넌트에 아티팩트를 포함시켜서 성능 평가 같은 일을 할 수 있으며, 의사 결정을 위한 자료로 사용할 수도 있습니다. 그리고 파이프 라인의 다양한 컴포넌트들이 어떻게 작동하는 이해하는데 많은 도움을 줄 수도 있습니다. 아티팩트는 일반적인 텍스트 뿐만 아니라, 시각화를 위한 데이터까지 다양하게 존재합니다.

Kubeflow Pipelines 구성 요소

Kubeflow Pipelines은 다음과 같은 구성 요소로 이루어져 있습니다.

  • 실험, 작업 및 실행을 관리하고 추적하기 위한 사용자 인터페이스 (UI)
  • 파이프라인을 관리하는 파이프라인 서비스
  • 파이프라인과 컴포넌트를 정의하고 제어하기 위한 SDK.
  • 머신 러닝 워크 플로우 실행을 위한 컨트롤러.

Kubeflow Pipelines UI(User interface)

Kubeflow Pipelines UI 는 현재 실행 중인 파이프 라인 목록, 파이프 라인 실행 기록, 데이터 아티팩트 목록, 개별 파이프 라인 실행에 대한 디버깅 정보, 개별 파이프 라인 실행에 대한 실행 상태를 표시합니다.

https://www.kubeflow.org/docs/images/pipelines-ui.png

Kubeflow Pipelines UI 에서 다음과 같은 작업을 수행 할 수 있습니다.

  • 압축 파일로 만들어진 파이프 라인을 업로드 할 수 있습니다. 업로드 된 파이프 라인은 다른 사람들과 공유 할 수 있습니다.
  • 파이프 라인의 실행을 그룹화하는 “Experiment“을 생성할 수 있습니다.
  • Experiment 내에서 파이프라인을 실행할 수 있습니다.
  • 파이프 라인 실행의 구성, 그래프 및 출력을 확인할 수 있습니다
  • 반복 실행을 작성하여 실행을 예약할 수 있습니다.

Python SDK

Kubeflow Pipelines SDK는 머신 러닝 워크 플로우를 정의하고, 실행시킬 수 있는 파이썬 패키지 세트입니다.

다음은 SDK의 주요 패키지 입니다.

  • kfp.compiler : 파이프 라인을 컴파일 할 수 있는 기능을 제공하고 있습니다.
  • kfp.component : 파이프 라인 컴포넌트와 상호 작용하기 위한 기능을 제공하고 있습니다.
  • kfp.containers : 컴포넌트 컨테이너 이미지를 빌드하는 기능을 제공하고 있습니다.
  • kfp.Client : Kubeflow Pipelines API 용 Python 클라이언트 라이브러리가 포함되어 있습니다.

Pipeline Service

파이프라인 서비스는 파이프 라인을 생성하고 실행하는 등의 관리 역할을 하고 있습니다. 그리고 Experiment, Run 같은 파이프라인 메타데이터를 메타데이터 저장소에 저장하는 역할도 하고 있습니다.

또한 REST API도 제공하고 있습니다. REST API는 셸 스크립트 또는 다른 시스템에 통합하려는 경우 유용하게 사용할 수 있습니다.

Pipelines 데이터 저장소

Kubeflow Pipelines에는 머신 러닝 파이프 라인에 관련된 데이터 관리하기 위해서 다음과 같은 두 개의 저장소를 가지고 있습니다.

  • Metadata : Experiment, Run 등 Kubeflow Pipelines는 파이프 라인 메타 데이터를 MySQL 데이터베이스에 저장합니다.
  • Artifacts : 파이프라인 패키지, 메트릭, 뷰 등 아티팩트를 Minio 서버에 저장합니다.

Kubeflow Pipelines는 쿠버네티스의 퍼시스턴스 볼륨(PV)을 사용하여 MySQL 데이터베이스와 Minio 서버의 데이터를 저장합니다.

Orchestration Controllers

오케스트레이션 컨트롤러는 머신 러닝 워크 플로우 실행 다시 말해서, 파이프 라인을 완료하는데 필요한 컨테이너들을 실행시키는 역할을 하고 있습니다. 컨테이너들은 쿠버네티스의 포드 형태로 실행됩니다. 현재 Kubeflow Pipelines 에서는 Argo Workflow 를 워크플로우 컨트롤러로 사용하고 있습니다.

작성일자

Kubeflow – Katib : Metrics Collector

Metrics Collector 알아보기

앞서 하이퍼 파라미터 튜닝에서 사용했던 메트릭 수집기는 기본 수집기인 StdOut 메트릭 수집기였습니다. 이번에는 StdOud 메트릭 수집기에 필터를 적용하는 방법과 TensorFlowEvent, File 그리고 Custom 메트릭 수집기에 대해서 알아보겠습니다.

StdOud 메트릭 수집기에 필터 적용하기

StdOut 메트릭 수집기에 필터를 적용하는 방법에 대해서 알아보겠습니다. 기존 예제에서는 StdOut 으로 출력되는 메트릭을 수집하기 위해서 {{MetricsName}}={{MetricsValue}} 형태로 출력을 하였습니다. 필터를 사용하면 메트릭을 나타내는 형식을 지정할 수 있기 때문에, 모델 학습시 출력되는 기본적인 로그를 그대로 사용할 수 있습니다.

예를 든다면, mnist-simple.py 를 실행하면 다음과 같은 로그가 출력됩니다.

Epoch 1/5
50000/50000 [==============================] - 2s 46us/sample - loss: 0.3268 - accuracy: 0.9055 - val_loss: 0.1509 - val_accuracy: 0.9574
Epoch 2/5
50000/50000 [==============================] - 2s 42us/sample - loss: 0.1581 - accuracy: 0.9534 - val_loss: 0.1115 - val_accuracy: 0.9684
Epoch 3/5
50000/50000 [==============================] - 2s 40us/sample - loss: 0.1166 - accuracy: 0.9642 - val_loss: 0.1017 - val_accuracy: 0.9708
Epoch 4/5
50000/50000 [==============================] - 2s 40us/sample - loss: 0.0959 - accuracy: 0.9707 - val_loss: 0.0836 - val_accuracy: 0.9756
Epoch 5/5
50000/50000 [==============================] - 2s 42us/sample - loss: 0.0808 - accuracy: 0.9747 - val_loss: 0.0774 - val_accuracy: 0.9773

로그를 보면, 메트릭이 “accuracy: 0.9055 “, “val_accuracy: 0.9574” 이런 형식으로 출력되는 것을 확인 할 수 있습니다. 필터에 {{MetricsName}}:{{MetricsValue}} 형식을 추가해서 기본 로그에서 메트릭을 추출하도록 하겠습니다. 형식은 go 언어의 정규표현식을 사용할 수 있습니다.

다음은 {{MetricsName}}:{{MetricsValue}} 형식을 필터로 사용하는 metricsCollectorSpec 입니다.

metricsCollectorSpec:
    collector:
      kind: StdOut
    source:
      filter:
        metricsFormat:
          - "([\\\\w|-]+)\\\\s*:\\\\s*((-?\\\\d+)(\\\\.\\\\d+)?)"

모델 코드 만들기

텐서플로우 케라스로 작성한 mnist 숫자를 판별하는 모델입니다. Katib를 위한 별도의 로그는 출력하지 않습니다.

mnist-simple.py

from __future__ import absolute_import, division, print_function, unicode_literals

import argparse
import tensorflow as tf
import numpy as np

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

    parser = argparse.ArgumentParser()
    parser.add_argument('--learning_rate', default=0.01, type=float)
    parser.add_argument('--dropout', default=0.2, type=float)
    args = parser.parse_args()

    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

    # Reserve 10,000 samples for validation
    x_val = x_train[-10000:]
    y_val = y_train[-10000:]
    x_train = x_train[:-10000]
    y_train = y_train[:-10000]

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

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

    print("Training...")
    training_history = model.fit(x_train, y_train, epochs=5, validation_data=(x_val, y_val))

    print("Average test loss: ", np.average(training_history.history['loss']))


if __name__ == '__main__':
    train()

모델 컨테이너 이미지 만들기

모델 학습용 컨테이너 이미지를 만들기 위해서 Dockerfile을 생성하겠습니다.

다음은 텐서플로우 2.1을 기반 이미지로 해서, 모델 파일을 추가하는 Dockerfile 입니다.

Dockerfile

FROM tensorflow/tensorflow:2.1.0-py3

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

다음 명령어로 “kangwoo/mnist-simple:katib” 라는 이름으로 컨테이너 이미지를 빌드할 수 있습니다.

docker build -t kangwoo/mnist-simple:katib.

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

docker push kangwoo/mnist-simple:katib

Experiment 생성하기

Experiment라는 사용자 리소스를 정의합니다. metricsCollectorSpec 필드에 filter가 추가되어 있습니다.

random-stdout-filter-example.yaml

apiVersion: "kubeflow.org/v1alpha3"
kind: Experiment
metadata:
  namespace: admin
  name: random-stdout-filter-example
spec:
  metricsCollectorSpec:
    collector:
      kind: StdOut
    source:
      filter:
        metricsFormat:
          - "([\\\\w|-]+)\\\\s*:\\\\s*((-?\\\\d+)(\\\\.\\\\d+)?)"
  parallelTrialCount: 1
  maxTrialCount: 12
  maxFailedTrialCount: 3
  objective:
    type: maximize
    goal: 0.99
    objectiveMetricName: val_accuracy
    additionalMetricNames:
      - accuracy
  algorithm:
    algorithmName: random
  parameters:
    - name: --learning_rate
      parameterType: double
      feasibleSpace:
        min: "0.01"
        max: "0.2"
    - name: --dropout
      parameterType: double
      feasibleSpace:
        min: "0.1"
        max: "0.5"
  trialTemplate:
    goTemplate:
        rawTemplate: |-
          apiVersion: batch/v1
          kind: Job
          metadata:
            name: {{.Trial}}
            namespace: {{.NameSpace}}
          spec:
            template:
              spec:
                containers:
                - name: {{.Trial}}
                  image: kangwoo/mnist-simple:katib
                  imagePullPolicy: Always
                  command:
                  - "python3"
                  - "/app/mnist-simple.py"
                  {{- with .HyperParameters}}
                  {{- range .}}
                  - "{{.Name}}={{.Value}}"
                  {{- end}}
                  {{- end}}
                restartPolicy: Never

정의한 Experiment 사용자 리소스를 쿠버네티스 클러스터에 생성합니다.

kubectl apply -f random-stdout-filter-example.yaml

Experiment 결과 보기

Katib UI를 통해서 다음과 같은 결과를 확인할 수 있습니다.

TensorFlowEvent 메트릭 수집기 사용하기

TensorFlowEvent 메트릭 수집기를 사용해 보겠습니다. TensorFlowEvent 메트릭 수집기는 텐서플로우에서 생성하는 이벤트를 추출해서 메트릭을 수집합니다. 그래서 기존의 텐서플로우 코드를 사용할 때 유용합니다. 다만 혀재는 텐서플로우 1 버전만을 지원하기 때문에, 텐서플로우 2 버전에 사용하기에는 약간의 문제가 있습니다.

다음은 TensorFlowEvent 메트릭 수집기를 사용하는 metricsCollectorSpec 입니다. fileSystemPath 필드를 사용해서 이벤트가 저장되어 있는 경로를 지정해 주어야합니다.

metricsCollectorSpec:
    collector:
      kind: TensorFlowEvent
    source:
      fileSystemPath:
        path: /train
        kind: Directory

모델 코드 만들기

텐서플로우 1 버전으로 작성한 mnist 숫자를 판별하는 모델입니다. tf.summary를 사용하여 이벤트를 출력하고 있습니다.

mnist-with-summaries.py

from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

import argparse
import os
import sys

import tensorflow as tf

from tensorflow.examples.tutorials.mnist import input_data

FLAGS = None


def train():
  # Import data
  mnist = input_data.read_data_sets(FLAGS.data_dir,
                                    fake_data=FLAGS.fake_data)

  sess = tf.InteractiveSession()
  # Create a multilayer model.

  # Input placeholders
  with tf.name_scope('input'):
    x = tf.placeholder(tf.float32, [None, 784], name='x-input')
    y_ = tf.placeholder(tf.int64, [None], name='y-input')

  with tf.name_scope('input_reshape'):
    image_shaped_input = tf.reshape(x, [-1, 28, 28, 1])
    tf.summary.image('input', image_shaped_input, 10)

  # We can't initialize these variables to 0 - the network will get stuck.
  def weight_variable(shape):
    """Create a weight variable with appropriate initialization."""
    initial = tf.truncated_normal(shape, stddev=0.1)
    return tf.Variable(initial)

  def bias_variable(shape):
    """Create a bias variable with appropriate initialization."""
    initial = tf.constant(0.1, shape=shape)
    return tf.Variable(initial)

  def variable_summaries(var):
    """Attach a lot of summaries to a Tensor (for TensorBoard visualization)."""
    with tf.name_scope('summaries'):
      mean = tf.reduce_mean(var)
      tf.summary.scalar('mean', mean)
      with tf.name_scope('stddev'):
        stddev = tf.sqrt(tf.reduce_mean(tf.square(var - mean)))
      tf.summary.scalar('stddev', stddev)
      tf.summary.scalar('max', tf.reduce_max(var))
      tf.summary.scalar('min', tf.reduce_min(var))
      tf.summary.histogram('histogram', var)

  def nn_layer(input_tensor, input_dim, output_dim, layer_name, act=tf.nn.relu):
    """Reusable code for making a simple neural net layer.
    It does a matrix multiply, bias add, and then uses ReLU to nonlinearize.
    It also sets up name scoping so that the resultant graph is easy to read,
    and adds a number of summary ops.
    """
    # Adding a name scope ensures logical grouping of the layers in the graph.
    with tf.name_scope(layer_name):
      # This Variable will hold the state of the weights for the layer
      with tf.name_scope('weights'):
        weights = weight_variable([input_dim, output_dim])
        variable_summaries(weights)
      with tf.name_scope('biases'):
        biases = bias_variable([output_dim])
        variable_summaries(biases)
      with tf.name_scope('Wx_plus_b'):
        preactivate = tf.matmul(input_tensor, weights) + biases
        tf.summary.histogram('pre_activations', preactivate)
      activations = act(preactivate, name='activation')
      tf.summary.histogram('activations', activations)
      return activations

  hidden1 = nn_layer(x, 784, 500, 'layer1')

  with tf.name_scope('dropout'):
    keep_prob = tf.placeholder(tf.float32)
    tf.summary.scalar('dropout_keep_probability', keep_prob)
    dropped = tf.nn.dropout(hidden1, keep_prob)

  # Do not apply softmax activation yet, see below.
  y = nn_layer(dropped, 500, 10, 'layer2', act=tf.identity)

  with tf.name_scope('cross_entropy'):
    # The raw formulation of cross-entropy,
    #
    # tf.reduce_mean(-tf.reduce_sum(y_ * tf.log(tf.softmax(y)),
    #                               reduction_indices=[1]))
    #
    # can be numerically unstable.
    #
    # So here we use tf.losses.sparse_softmax_cross_entropy on the
    # raw logit outputs of the nn_layer above, and then average across
    # the batch.
    with tf.name_scope('total'):
      cross_entropy = tf.losses.sparse_softmax_cross_entropy(
          labels=y_, logits=y)
  tf.summary.scalar('cross_entropy', cross_entropy)

  with tf.name_scope('train'):
    train_step = tf.train.AdamOptimizer(FLAGS.learning_rate).minimize(
        cross_entropy)

  with tf.name_scope('accuracy'):
    with tf.name_scope('correct_prediction'):
      correct_prediction = tf.equal(tf.argmax(y, 1), y_)
    with tf.name_scope('accuracy'):
      accuracy = tf.reduce_mean(tf.cast(correct_prediction, tf.float32))
  tf.summary.scalar('accuracy', accuracy)

  # Merge all the summaries and write them out to
  # /tmp/tensorflow/mnist/logs/mnist_with_summaries (by default)
  merged = tf.summary.merge_all()
  train_writer = tf.summary.FileWriter(FLAGS.log_dir + '/train', sess.graph)
  test_writer = tf.summary.FileWriter(FLAGS.log_dir + '/test')
  tf.global_variables_initializer().run()

  # Train the model, and also write summaries.
  # Every 10th step, measure test-set accuracy, and write test summaries
  # All other steps, run train_step on training data, & add training summaries

  def feed_dict(train):     # pylint: disable=redefined-outer-name
    """Make a TensorFlow feed_dict: maps data onto Tensor placeholders."""
    if train or FLAGS.fake_data:
      xs, ys = mnist.train.next_batch(FLAGS.batch_size, fake_data=FLAGS.fake_data)
      k = FLAGS.dropout
    else:
      xs, ys = mnist.test.images, mnist.test.labels
      k = 1.0
    return {x: xs, y_: ys, keep_prob: k}

  for i in range(FLAGS.max_steps):
    if i % 10 == 0:  # Record summaries and test-set accuracy
      summary, acc = sess.run([merged, accuracy], feed_dict=feed_dict(False))
      test_writer.add_summary(summary, i)
      print('Accuracy at step %s: %s' % (i, acc))
    else:  # Record train set summaries, and train
      if i % 100 == 99:  # Record execution stats
        run_options = tf.RunOptions(trace_level=tf.RunOptions.FULL_TRACE)
        run_metadata = tf.RunMetadata()
        summary, _ = sess.run([merged, train_step],
                              feed_dict=feed_dict(True),
                              options=run_options,
                              run_metadata=run_metadata)
        train_writer.add_run_metadata(run_metadata, 'step%03d' % i)
        train_writer.add_summary(summary, i)
        print('Adding run metadata for', i)
      else:  # Record a summary
        summary, _ = sess.run([merged, train_step], feed_dict=feed_dict(True))
        train_writer.add_summary(summary, i)
  train_writer.close()
  test_writer.close()


def main(_):
  if tf.gfile.Exists(FLAGS.log_dir):
    tf.gfile.DeleteRecursively(FLAGS.log_dir)
  tf.gfile.MakeDirs(FLAGS.log_dir)
  train()


if __name__ == '__main__':
  parser = argparse.ArgumentParser()
  parser.add_argument('--fake_data', nargs='?', const=True, type=bool,
                      default=False,
                      help='If true, uses fake data for unit testing.')
  parser.add_argument('--max_steps', type=int, default=1000,
                      help='Number of steps to run trainer.')
  parser.add_argument('--learning_rate', type=float, default=0.001,
                      help='Initial learning rate')
  parser.add_argument('--batch_size', type=int, default=100,
                      help='Training batch size')
  parser.add_argument('--dropout', type=float, default=0.9,
                      help='Keep probability for training dropout.')
  parser.add_argument(
      '--data_dir',
      type=str,
      default=os.path.join(os.getenv('TEST_TMPDIR', '/tmp'),
                           'tensorflow/mnist/input_data'),
      help='Directory for storing input data')
  parser.add_argument(
      '--log_dir',
      type=str,
      default=os.path.join(os.getenv('TEST_TMPDIR', '/tmp'),
                           'tensorflow/mnist/logs/mnist_with_summaries'),
      help='Summaries log directory')
  FLAGS, unparsed = parser.parse_known_args()
  tf.app.run(main=main, argv=[sys.argv[0]] + unparsed)

모델 컨테이너 이미지 만들기

모델 학습용 컨테이너 이미지를 만들기 위해서 Dockerfile을 생성하겠습니다.

다음은 텐서플로우 1.11을 기반 이미지로 해서, 모델 파일을 추가하는 Dockerfile 입니다.

Dockerfile

FROM tensorflow/tensorflow:1.11.0

RUN mkdir -p /app
ADD mnist-with-summaries.py /app/

다음 명령어로 “kangwoo/mnist-simple:katib” 라는 이름으로 컨테이너 이미지를 빌드할 수 있습니다.

docker build -t kangwoo/mnist-with-summaries:katib .

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

docker push kangwoo/mnist-with-summaries:katib

Experiment 생성하기

Experiment라는 사용자 리소스를 정의합니다. metricsCollectorSpec 필드에 filter가 추가되어 있습니다.

random-tf-event-example.yaml

apiVersion: "kubeflow.org/v1alpha3"
kind: Experiment
metadata:
  namespace: admin
  name: random-tf-event-example
spec:
  metricsCollectorSpec:
    source:
      fileSystemPath:
        path: /train
        kind: Directory
    collector:
      kind: TensorFlowEvent
  parallelTrialCount: 1
  maxTrialCount: 12
  maxFailedTrialCount: 3
  objective:
    type: maximize
    goal: 0.99
    objectiveMetricName: accuracy_1
  algorithm:
    algorithmName: random
  parameters:
    - name: --learning_rate
      parameterType: double
      feasibleSpace:
        min: "0.01"
        max: "0.05"
    - name: --batch_size
      parameterType: int
      feasibleSpace:
        min: "100"
        max: "200"
  trialTemplate:
    goTemplate:
        rawTemplate: |-
          apiVersion: "kubeflow.org/v1"
          kind: TFJob
          metadata:
            name: {{.Trial}}
            namespace: {{.NameSpace}}
          spec:
           tfReplicaSpecs:
            Worker:
              replicas: 1
              restartPolicy: OnFailure
              template:
                spec:
                  containers:
                    - name: tensorflow
                      image: kangwoo/mnist-with-summaries:katib
                      imagePullPolicy: Always
                      command:
                        - "python"
                        - "/app/mnist-with-summaries.py"
                        - "--log_dir=/train/metrics"
                        {{- with .HyperParameters}}
                        {{- range .}}
                        - "{{.Name}}={{.Value}}"
                        {{- end}}
                        {{- end}}

정의한 Experiment 사용자 리소스를 쿠버네티스 클러스터에 생성합니다.

kubectl apply -f random-tf-event-example.yaml

Experiment 결과 보기

Katib UI를 통해서 다음과 같은 결과를 확인할 수 있습니다.

File 메트릭 수집기 사용하기

File 메트릭 수집기를 사용해 보겠습니다. File 메트릭 수집기는 파일로 출력되는 로그를 추출해서 메트릭을 수집합니다. File 메트릭 수집기도 필터를 사용하여 메트릭 형식을 지정할 수 있습니다. 메트릭 형식을 지정하지 않으면, 기본 형식인 “([\w|-]+)\s*=\s*((-?\d+)(\.\d+)?)” 즉 {{MetricsName}}={{MetricsValue}} 을 사용합니다.

다음은 File 메트릭 수집기를 사용하는 metricsCollectorSpec 입니다. fileSystemPath 필드를 사용해서 로그가 저장되어 있는 파일 경로를 지정해 주어야 합니다. 파일 경로를 지정하지 않으면 기본 경로인 “/var/log/katib/metrics.log”을 사용합니다.

metricsCollectorSpec:
    source:
      filter:
        metricsFormat:
        - "([\\\\w|-]+)\\\\s*=\\\\s*((-?\\\\d+)(\\\\.\\\\d+)?)"
      fileSystemPath:
        path: "/var/log/katib/mnist.log"
        kind: File
    collector:
      kind: File

모델 코드 만들기

텐서플로우 케라스로 작성한 mnist 숫자를 판별하는 모델입니다. logging 패키지를 사용하여 파일로 로그를 출력하고 있습니다.

mnist-with-log.py

from __future__ import absolute_import, division, print_function, unicode_literals

import tensorflow as tf
import argparse
import numpy as np
from datetime import datetime, timezone

import logging

logging.basicConfig(filename='/var/log/katib/mnist.log', level=logging.DEBUG)


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

    parser = argparse.ArgumentParser()
    parser.add_argument('--learning_rate', default=0.01, type=float)
    parser.add_argument('--dropout', default=0.2, type=float)
    args = parser.parse_args()

    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

    # Reserve 10,000 samples for validation
    x_val = x_train[-10000:]
    y_val = y_train[-10000:]
    x_train = x_train[:-10000]
    y_train = y_train[:-10000]

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

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

    print("Training...")
    katib_metric_log_callback = KatibMetricLog()
    training_history = model.fit(x_train, y_train, batch_size=64, epochs=10,
                                 validation_data=(x_val, y_val),
                                 callbacks=[katib_metric_log_callback])
    print("Average test loss: ", np.average(training_history.history['loss']))


class KatibMetricLog(tf.keras.callbacks.Callback):
    def on_epoch_end(self, epoch, logs=None):
        # RFC 3339
        local_time = datetime.now(timezone.utc).astimezone().isoformat()
        logging.info("\\n{} accuracy={:.4f} loss={:.4f} Validation-accuracy={:.4f} Validation-loss={:.4f}"
                     .format(local_time, logs['acc'], logs['loss'], logs['val_acc'], logs['val_loss']))


if __name__ == '__main__':
    train()

모델 컨테이너 이미지 만들기

모델 학습용 컨테이너 이미지를 만들기 위해서 Dockerfile을 생성하겠습니다.

다음은 텐서플로우 2.1을 기반 이미지로 해서, 모델 파일을 추가하는 Dockerfile 입니다.

Dockerfile

FROM tensorflow/tensorflow:2.1.0-py3

RUN mkdir -p /app
ADD mnist-with-log.py /app/

다음 명령어로 “kangwoo/mnist-with-log:katib” 라는 이름으로 컨테이너 이미지를 빌드할 수 있습니다.

docker build -t kangwoo/mnist-with-log:katib.

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

docker push kangwoo/mnist-with-log:katib

Experiment 생성하기

Experiment라는 사용자 리소스를 정의합니다. metricsCollectorSpec 필드에 filter가 추가되어 있습니다.

random-tf-event-example.yaml

apiVersion: "kubeflow.org/v1alpha3"
kind: Experiment
metadata:
  namespace: admin
  name: random-file-example
spec:
  metricsCollectorSpec:
    source:
      fileSystemPath:
        path: "/var/log/katib/mnist.log"
        kind: File
    collector:
      kind: File
  parallelTrialCount: 1
  maxTrialCount: 12
  maxFailedTrialCount: 3
  objective:
    type: maximize
    goal: 0.99
    objectiveMetricName: Validation-accuracy
    additionalMetricNames:
      - accuracy
  algorithm:
    algorithmName: random
  parameters:
    - name: --learning_rate
      parameterType: double
      feasibleSpace:
        min: "0.01"
        max: "0.2"
    - name: --dropout
      parameterType: double
      feasibleSpace:
        min: "0.1"
        max: "0.5"
  trialTemplate:
    goTemplate:
        rawTemplate: |-
          apiVersion: batch/v1
          kind: Job
          metadata:
            name: {{.Trial}}
            namespace: {{.NameSpace}}
          spec:
            template:
              spec:
                containers:
                - name: {{.Trial}}
                  image: kangwoo/mnist-with-log:katib
                  imagePullPolicy: Always
                  command:
                  - "python3"
                  - "/app/mnist-with-log.py"
                  {{- with .HyperParameters}}
                  {{- range .}}
                  - "{{.Name}}={{.Value}}"
                  {{- end}}
                  {{- end}}
                restartPolicy: Never

정의한 Experiment 사용자 리소스를 쿠버네티스 클러스터에 생성합니다.

kubectl apply -f random-tf-event-example.yaml

Experiment 결과 보기

Katib UI를 통해서 다음과 같은 결과를 확인할 수 있습니다.

https://s3-us-west-2.amazonaws.com/secure.notion-static.com/2d0f4811-7203-4ef7-bfd2-8dd2123f35d3/Untitled.png

작성일자

Kubeflow – Katib 하이퍼 파라미터 튜닝

하이퍼 파라미터 및 하이퍼 파라미터 튜닝

하이퍼 파라미터는 모델 학습 프로세스를 제어하는 ​​변수로서, 학습을 수행하기 위해 사전에 설정해야 하는 값들입니다. 예를 든다면 Learning rate, Batch Size, Regularization Strength 등이 있습니다.

하이퍼 파라미터 값은 학습되지 않습니다. 즉, 가중치 같은 학습 매개 변수와는 달리, 모델 학습 프로세스에서 하이퍼 파라미터 값을 조정하지 않습니다. 그래서 휴리스틱한 방법이나 경험 법칙에 의해서 결정하는 경우가 많습니다.

하이퍼 파라미터 튜닝은 최적의 하이퍼 파라미터 값을 탐색하여, 모델의 예측 정확도를 최대화하는 프로세스입니다. 만일 Katib 같은 자동화된 하이퍼 파라미터 튜닝 시스템이 없다면, 최적의 값을 찾기 위해 하이퍼 파라미터를 수동으로 조정하여, 많은 학습 작업을 사림이 직접 실행해야할것

자동화된 하이퍼 파라미터 튜닝 시스템은 대상의 목표 값을 이루기 위한 최적의 변수 값을 찾기 위해서 노력합니다. 일반적으로 모델의 정확성(accuracy)을 대상으로 사용합니다.

예를 들어 Katib의 다음 그래프는 다양한 하이퍼 파라미터 값의 조합 (learning_rate, dropout)에 따른 정확도를 보여 줍니다.

Katib는 Experiment이라 부르는 하이퍼 파라미터 튜닝 작업을 실행합니다. 실행된 Experiment는 Trial 이라고 부르는 학습 작업을 여러번 실행합니다.

random 알고리즘과 job을 이용한 하이퍼 파라미터 튜닝

하이퍼 파라미터 튜닝에 사용할 학습 모델 컨테이너 이미지를 만들어 보겠습니다.

모델 코드 작성하기

mnist 숫자를 판별하는 모델을 텐서플로우 케라스로 작성해 보겠습니다.

  1. 하이퍼 파라메터를 입력 받기 위해서 argparse 라이브러를 이용하였습니다. learning_rate와 dropout 값을 입력할 수 있습니다. parser = argparse.ArgumentParser() parser.add_argument('--learning_rate', default=0.01, type=float) parser.add_argument('--dropout', default=0.2, type=float) args = parser.parse_args()
  2. 케라스의 콜백을 이용해서, 매 에폭(epoch)마다 accuracy, loss, Validation-accuracy 그리고 Validation-loss를 StdOut 으로 출력하도록 하였습니다. Katib의 StdOutCollector를 사용해서 메트릭을 수집할 것이기 때문에, StdOut으로 {{MetricsName}}={{MetricsValue}} 형태로 메트릭을 StdOut 으로 출력하면 됩니다. 그리고 라인의 맨 앞부분에 RFC-3339 형식의 시간을 출력하면, 메트릭의 시간도 같이 수집이 됩니다. katib_metric_log_callback = KatibMetricLog() training_history = model.fit(x_train, y_train, batch_size=64, epochs=10, validation_data=(x_val, y_val), callbacks=[katib_metric_log_callback]) … class KatibMetricLog(tf.keras.callbacks.Callback): def on_epoch_end(self, epoch, logs=None): # RFC 3339 local_time = datetime.now(timezone.utc).astimezone().isoformat() print(“\nEpoch {}”.format(epoch+1)) print(“{} accuracy={:.4f}”.format(local_time, logs[‘acc’])) print(“{} loss={:.4f}”.format(local_time, logs[‘loss’])) print(“{} Validation-accuracy={:.4f}”.format(local_time, logs[‘val_acc’])) print(“{} Validation-loss={:.4f}”.format(local_time, logs[‘val_loss’]))

다음 코드는 텐서플로우 케라스로 작성한 mnist 숫자를 판별하는 모델입니다.

katib-mnist-random-job.py

from __future__ import absolute_import, division, print_function, unicode_literals

import tensorflow as tf
import numpy as np
import argparse
from datetime import datetime, timezone

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

    parser = argparse.ArgumentParser()
    parser.add_argument('--learning_rate', default=0.01, type=float)
    parser.add_argument('--dropout', default=0.2, type=float)
    args = parser.parse_args()

    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

    # Reserve 10,000 samples for validation
    x_val = x_train[-10000:]
    y_val = y_train[-10000:]
    x_train = x_train[:-10000]
    y_train = y_train[:-10000]

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

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

    print("Training...")

    katib_metric_log_callback = KatibMetricLog()
    training_history = model.fit(x_train, y_train, batch_size=64, epochs=10,
                                 validation_data=(x_val, y_val),
                                 callbacks=[katib_metric_log_callback])

    print("\\ntraining_history:", training_history.history)

    # Evaluate the model on the test data using `evaluate`
    print('\\n# Evaluate on test data')
    results = model.evaluate(x_test, y_test, batch_size=128)
    print('test loss, test acc:', results)


class KatibMetricLog(tf.keras.callbacks.Callback):
    def on_epoch_end(self, epoch, logs=None):
        # RFC 3339
        local_time = datetime.now(timezone.utc).astimezone().isoformat()
        print("\\nEpoch {}".format(epoch+1))
        print("{} accuracy={:.4f}".format(local_time, logs['acc']))
        print("{} loss={:.4f}".format(local_time, logs['loss']))
        print("{} Validation-accuracy={:.4f}".format(local_time, logs['val_acc']))
        print("{} Validation-loss={:.4f}".format(local_time, logs['val_loss']))


if __name__ == '__main__':
    train()

모델 컨테이너 이미지 만들기

모델 학습용 컨테이너 이미지를 만들기 위해서 Dockerfile을 생성하겠습니다.

다음은 텐서플로우 2.1을 기반 이미지로 해서, 모델 파일을 추가하는 Dockerfile 입니다.

Dockerfile

FROM tensorflow/tensorflow:2.1.0-py3

RUN mkdir -p /app
ADD katib-mnist-random-job.py /app/

ENTRYPOINT ["python", "/app/katib-mnist-random-job.py"]

다음 명령어로 “kangwoo/katib-mnist-job:0.0.1” 라는 이름으로 컨테이너 이미지를 빌드할 수 있습니다.

docker build -t kangwoo/katib-mnist-job:0.0.1 .

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

docker push kangwoo/katib-mnist-job:0.0.1

Experiment 생성하기

Katib를 사용하여 하이퍼 파라미터를 자동으로 튜닝하려면 Experiment라는 사용자 리소스를 정의해야합니다. Experiment에는 다음과 같은 내용이 포함되어 있습니다.

  • Objective: 최적화하려는 측정 항목.
  • Search algorithm: 최적의 하이퍼 파라미터를 찾는 데 사용하는는 알고리즘.
  • Configuration about parallelism: 병렬 처리에 대한 구성.
  • Search space: 탐색해야 하는 모든 하이퍼 파라미터의 이름 및 분포 (개별 값 또는 연속 값).
  • Trial Template: Trial을 정의하는 데 사용되는 템플릿.
  • Metrics Collection: 메트릭 수집 방법에 대한 정의

병렬 처리에 대한 구성 : 병렬 처리에 대한 설정할 수 있습니다.

  • parallelTrialCount : 병렬로 처리 할 수 있는 Trial 개수입니다.
  • maxTrialCount : Trial이 실행되는 최대 개수입니다.
  • maxFailedTrialCount : 최대 Trial 실패 개수를 넘으면 experiment은 실패하게 됩니다.
  parallelTrialCount: 1
  maxTrialCount: 12
  maxFailedTrialCount: 3

목표 : 최적화하려는 측정 항목을 설정할 수 있습니다.

“Validation-accuracy” 라는 이름의 메트릭의 최대값이 0.99에 도달하는 것을 목표로 합니다. 그리고 추가로 “accuracy” 라는 이름의 메트릭도 같이 수집합니다.

type은 maximize 나 minimize 를 사용할 수 있습니다.

  objective:
    type: maximize
    goal: 0.99
    objectiveMetricName: Validation-accuracy
    additionalMetricNames:
      - accuracy

검색 알고리즘 : 최적의 하이퍼 파라미터를 찾는 데 사용할 알고리즘을 설정할 수 있습니다.

하이퍼 파라미터 튜닝 알고리즘은 “random”을 사용합니다. 알고리즘 이름은 grid, random, hyperband, bayesianoptimization 을 사용할 수 있습니다.

  algorithm:
    algorithmName: random

탐색 공간 : 탐색해야하는 모든 하이퍼 파라미터의 이름과 범위(개별 값 또는 연속 값)에 대해 설정할 수 있습니다.

모델 학습에서 사용할 하이퍼 파라미터 목록입니다. learning_rate와, dropout을 파라미터로 정의합니다.

parameterType은 int, double, categorical 을 사용할 수 있습니다. 파라미터 값은 feasibleSpace, list를 사용할 수 있습니다.

  parameters:
    - name: --learning_rate
      parameterType: double
      feasibleSpace:
        min: "0.01"
        max: "0.2"
    - name: --dropout
      parameterType: double
      feasibleSpace:
        min: "0.1"
        max: "0.5"

Trial Template: Trial에서 생성할 Worker Job을 정의할 수 있습니다.

쿠버네티스의 Job을 생성해서 모델 학습 작업을 합니다.

  trialTemplate:
    goTemplate:
        rawTemplate: |-
          apiVersion: batch/v1
          kind: Job
          metadata:
            name: {{.Trial}}
            namespace: {{.NameSpace}}
          spec:
            template:
              spec:
                containers:
                - name: {{.Trial}}
                  image: kangwoo/katib-mnist-job:0.0.1
                  command:
                  - "python3"
                  - "/app/katib-mnist-random-job.py"
                  {{- with .HyperParameters}}
                  {{- range .}}
                  - "{{.Name}}={{.Value}}"
                  {{- end}}
                  {{- end}}
                restartPolicy: Never
  1. 메트릭 수집 방법에 대해서 정의합니다. 별도로 정의하지 않을 경우 StdOut 메트릭 수집기가 사용됩니다. metricsCollectorSpec: collector: kind: StdOut

다음은 admin에라는 네임스페이스에 생성할 Experiment 매니페스트입니다.

random-job-example.yaml

apiVersion: "kubeflow.org/v1alpha3"
kind: Experiment
metadata:
  namespace: admin
  name: random-job-example
spec:
  parallelTrialCount: 1
  maxTrialCount: 12
  maxFailedTrialCount: 3
  objective:
    type: maximize
    goal: 0.99
    objectiveMetricName: Validation-accuracy
    additionalMetricNames:
      - accuracy
  algorithm:
    algorithmName: random
  parameters:
    - name: --learning_rate
      parameterType: double
      feasibleSpace:
        min: "0.01"
        max: "0.2"
    - name: --dropout
      parameterType: double
      feasibleSpace:
        min: "0.1"
        max: "0.5"
  trialTemplate:
    goTemplate:
        rawTemplate: |-
          apiVersion: batch/v1
          kind: Job
          metadata:
            name: {{.Trial}}
            namespace: {{.NameSpace}}
          spec:
            template:
              spec:
                containers:
                - name: {{.Trial}}
                  image: kangwoo/katib-mnist-job:0.0.1
                  command:
                  - "python3"
                  - "/app/katib-mnist-random-job.py"
                  {{- with .HyperParameters}}
                  {{- range .}}
                  - "{{.Name}}={{.Value}}"
                  {{- end}}
                  {{- end}}
                restartPolicy: Never

Experiment 은 katib UI 화면이나, kubectl을 사용해서 생성할 수 있습니다.

Katib UI 화면에서 Experiment 생성하기

Kubeflow 대시보드의 왼쪽 메뉴에서 Katib를 클릭합니다.

“Hyperparameter Tuning”을 클릭합니다.

YAML File 탭에서 작성한 Experiment 매니페스트를 입력하고, 맨 아래에 있는 DEPLOY 버튼을 클릭하면 Experiment가 생성됩니다.

kubectl을 사용해서 Experiment 생성하기

kubectl을 사용해서 Experiment를 생성할 수 있습니다.

Experiment 매니페스트를 random-job-example.yaml 파일로 저정한 후, 다음 명령어를 사용하면, Experiment 를 생성할 수 있습니다.

kubectl apply -f random-job-example.yaml

Experiment 결과 보기

Katib UI 화면에서 Experiment 결과 보기

Katib UI 화면의 좌측 상단에 있는 메뉴를 선택한 후, HP > Monitor 를 선택하면, Experiment Monitor 화면으로 이동할 수 있습니다.

다음은 Experiment Monitor 화면입니다. 생성한 Experiment 목록을 확인할 수 있습니다.

Experiment 이름을 클릭하면 Trial의 측정 결과가 표시됩니다.

Trial 이름을 클릭하면 측정 항목에 대한 세부 정보가 표시됩니다.

kubectl을 사용해서 Experiment 결과 보기

kubectl 사용해서 결과를 조회 할 수 있습니다.

다음 명령을 사용하면, Trial 측정 결과를 조회 할 수 있습니다. (JSON을 구문 분석 때문에, jq를 설치해야 합니다)

kubectl -n admin get trials -l experiment=random-job-example -o json | jq ".items[] | {assignments: .spec.parameterAssignments, observation: .status.observation}"

{
  "assignments": [
    {
      "name": "--learning_rate",
      "value": "0.08177734351368438"
    },
    {
      "name": "--dropout",
      "value": "0.4439382425122721"
    }
  ],
  "observation": {
    "metrics": [
      {
        "name": "Validation-accuracy",
        "value": 0.9712
      }
    ]
  }
}
{
  "assignments": [
    {
      "name": "--learning_rate",
      "value": "0.13167199355992532"
    },
    {
      "name": "--dropout",
      "value": "0.36691549333903695"
    }
  ],
  "observation": {
    "metrics": [
      {
        "name": "Validation-accuracy",
        "value": 0.9752
      }
    ]
  }
}
...

grid 알고리즘을 이용한 하이퍼 파라미터 튜닝

grid 알고리즘을 사용하여 하이퍼 파리미터 튜닝을 해 보겠습니다. 모델 컨테이너 이미지는 radnom 하이퍼 파라미터 튜닝에서 사용한 kangwoo/mnist:katib 을 그대로 사용하겠습니다.

grid 알고리즘을 사용하려면 algorithmName 필드에 grid 라고 설정하면 됩니다.

algorithm:
    algorithmName: grid

주의해할 점은 categorical 타입의 파라미터는 지원하지 않습니다. 그래서 다음과 같은 파라미터는 사용할 수 없습니다

parameters:
    # Grid doesn't support categorical, refer to <https://chocolate.readthedocs.io/api/sample.html#chocolate.Grid>
    - name: --optimizer
      parameterType: categorical
      feasibleSpace:
        list:
        - sgd
        - adam
        - ftrl

그리고 double 타입의 파라미터를 사용할 때는 step을 정의해줘야 합니다. 값을 얼마만큼의 간격으로 증가시킬지를 지정하는 것입니다. int 타입의 파라미터인 경우에도 setup 값을 정의할 수 있습니다. int 타입인 경우 별도로 정의하지 않으면 기본값인 1일 사용합니다.

parameters:
    - name: --learning_rate
      parameterType: double
      feasibleSpace:
        min: "0.01"
        max: "0.2"
        step: "0.01"

Experiment 생성하기

Experiment라는 사용자 리소스를 정의합니다. metricsCollectorSpec 필드에 filter가 추가되어 있습니다.

grid-stdout-example.yaml

apiVersion: "kubeflow.org/v1alpha3"
kind: Experiment
metadata:
  namespace: admin
  name: grid-stdout-example
spec:
  parallelTrialCount: 1
  maxTrialCount: 12
  maxFailedTrialCount: 3
  objective:
    type: maximize
    goal: 0.99
    objectiveMetricName: Validation-accuracy
    additionalMetricNames:
      - accuracy
  algorithm:
    algorithmName: grid
  parameters:
    - name: --learning_rate
      parameterType: double
      feasibleSpace:
        min: "0.01"
        max: "0.2"
        step: "0.01"
    - name: --dropout
      parameterType: double
      feasibleSpace:
        min: "0.1"
        max: "0.5"
        step: "0.05"
  trialTemplate:
    goTemplate:
        rawTemplate: |-
          apiVersion: batch/v1
          kind: Job
          metadata:
            name: {{.Trial}}
            namespace: {{.NameSpace}}
          spec:
            template:
              spec:
                containers:
                - name: {{.Trial}}
                  image: kangwoo/mnist:katib
                  command:
                  - "python3"
                  - "/app/mnist.py"
                  {{- with .HyperParameters}}
                  {{- range .}}
                  - "{{.Name}}={{.Value}}"
                  {{- end}}
                  {{- end}}
                restartPolicy: Never

정의한 Experiment 사용자 리소스를 쿠버네티스 클러스터에 생성합니다.

kubectl apply -f grid-stdout-example.yaml

Experiment 결과 보기

Katib UI를 통해서 다음과 같은 결과를 확인할 수 있습니다.

bayesianoptimization 알고리즘을 이용한 하이퍼 파라미터 튜닝

bayesianoptimization 알고리즘을 사용하여 하이퍼 파리미터 튜닝을 해 보겠습니다. 모델 컨테이너 이미지는 radnom 하이퍼 파라미터 튜닝에서 사용한 kangwoo/mnist:katib 을 그대로 사용하겠습니다.

bayesianoptimization 알고리즘을 사용하려면 algorithmName 필드에 bayesianoptimization 라고 설정하면 됩니다. 그리고 algorithmSettings 필드를 사용해서 알고리즘을 설정할 수 있습니다.

algorithm:
    algorithmName: bayesianoptimization
    algorithmSettings:
      - name: "random_state"
        value: "10"

Experiment 생성하기

Experiment라는 사용자 리소스를 정의합니다. metricsCollectorSpec 필드에 filter가 추가되어 있습니다.

bayesianoptimization-stdout-example.yaml

apiVersion: "kubeflow.org/v1alpha3"
kind: Experiment
metadata:
  namespace: admin
  name: bayesianoptimization-stdout-example
spec:
  parallelTrialCount: 1
  maxTrialCount: 12
  maxFailedTrialCount: 3
  objective:
    type: maximize
    goal: 0.99
    objectiveMetricName: Validation-accuracy
    additionalMetricNames:
      - accuracy
  algorithm:
    algorithmName: bayesianoptimization
    algorithmSettings:
      - name: "random_state"
        value: "10"
  parameters:
    - name: --learning_rate
      parameterType: double
      feasibleSpace:
        min: "0.01"
        max: "0.2"
    - name: --dropout
      parameterType: double
      feasibleSpace:
        min: "0.1"
        max: "0.5"
  trialTemplate:
    goTemplate:
        rawTemplate: |-
          apiVersion: batch/v1
          kind: Job
          metadata:
            name: {{.Trial}}
            namespace: {{.NameSpace}}
          spec:
            template:
              spec:
                containers:
                - name: {{.Trial}}
                  image: kangwoo/mnist:katib
                  command:
                  - "python3"
                  - "/app/mnist.py"
                  {{- with .HyperParameters}}
                  {{- range .}}
                  - "{{.Name}}={{.Value}}"
                  {{- end}}
                  {{- end}}
                restartPolicy: Never

정의한 Experiment 사용자 리소스를 쿠버네티스 클러스터에 생성합니다.

kubectl apply -f bayesianoptimization-stdout-example.yaml

Experiment 결과 보기

Katib UI를 통해서 다음과 같은 결과를 확인할 수 있습니다.

hyperband 알고리즘을 이용한 하이퍼 파라미터 튜닝

hyperband 알고리즘을 사용하여 하이퍼 파리미터 튜닝을 해 보겠습니다. 모델 컨테이너 이미지는 radnom 하이퍼 파라미터 튜닝에서 사용한 kangwoo/mnist:katib 을 그대로 사용하겠습니다.

r_l and resource_name must be set.

r_l must be a positive float number.

if "eta" in setting_dict:
            eta = int(float(setting_dict["eta"]))
            if eta <= 0:
                eta = 3
        else:
            eta = 3


smax = int(math.log(rl)/math.log(eta))
        max_parallel = int(math.ceil(eta**smax))
        if request.experiment.spec.parallel_trial_count < max_parallel:
            return self._set_validate_context_error(context,
                                                    "parallelTrialCount must be not less than %d." % max_parallel)

parallel_trial_count 는 eta**log(rl)/log(eta) 값보다 커야 합니다.

hyperband 알고리즘을 사용하려면 algorithmName 필드에 hyperband 라고 설정하면 됩니다. 그리고 algorithmSettings 필드를 사용해서 알고리즘을 설정할 수 있습니다.

algorithm:
    algorithmName: hyperband
    algorithmSettings:
      - name: "resource_name"
        value: "--num-epochs"
      - name: "eta"
        value: "3"
      - name: "r_l"
        value: "9"

Experiment 생성하기

Experiment라는 사용자 리소스를 정의합니다. metricsCollectorSpec 필드에 filter가 추가되어 있습니다.

hyperband-stdout-example.yaml

apiVersion: "kubeflow.org/v1alpha3"
kind: Experiment
metadata:
  namespace: admin
  name: hyperband-stdout-example
spec:
  parallelTrialCount: 9
  maxTrialCount: 9
  maxFailedTrialCount: 9
  objective:
    type: maximize
    goal: 0.99
    objectiveMetricName: Validation-accuracy
    additionalMetricNames:
      - accuracy
  algorithm:
    algorithmName: hyperband
    algorithmSettings:
      - name: "resource_name"
        value: "--epochs"
      - name: "eta"
        value: "3"
      - name: "r_l"
        value: "9"
  parameters:
    - name: --learning_rate
      parameterType: double
      feasibleSpace:
        min: "0.01"
        max: "0.2"
    - name: --dropout
      parameterType: double
      feasibleSpace:
        min: "0.1"
        max: "0.5"
    - name: --epochs
      parameterType: int
      feasibleSpace:
        min: "10"
        max: "10"
  trialTemplate:
    goTemplate:
        rawTemplate: |-
          apiVersion: batch/v1
          kind: Job
          metadata:
            name: {{.Trial}}
            namespace: {{.NameSpace}}
          spec:
            template:
              spec:
                containers:
                - name: {{.Trial}}
                  image: kangwoo/mnist:katib
                  command:
                  - "python3"
                  - "/app/mnist.py"
                  {{- with .HyperParameters}}
                  {{- range .}}
                  - "{{.Name}}={{.Value}}"
                  {{- end}}
                  {{- end}}
                restartPolicy: Never

정의한 Experiment 사용자 리소스를 쿠버네티스 클러스터에 생성합니다.

kubectl apply -f hyperband-stdout-example.yaml

Experiment 결과 보기

Katib UI를 통해서 다음과 같은 결과를 확인할 수 있습니다.

tpe 알고리즘을 이용한 하이퍼 파라미터 튜닝

tpe 알고리즘을 사용하여 하이퍼 파리미터 튜닝을 해 보겠습니다. 모델 컨테이너 이미지는 radnom 하이퍼 파라미터 튜닝에서 사용한 kangwoo/mnist:katib 을 그대로 사용하겠습니다.

tpe 알고리즘을 사용하려면 algorithmName 필드에 tpe 라고 설정하면 됩니다.

algorithm:
    algorithmName: tpe

Experiment 생성하기

Experiment라는 사용자 리소스를 정의합니다. metricsCollectorSpec 필드에 filter가 추가되어 있습니다.

tpe-stdout-example.yaml

apiVersion: "kubeflow.org/v1alpha3"
kind: Experiment
metadata:
  namespace: admin
  name: tpe-stdout-example
spec:
  parallelTrialCount: 1
  maxTrialCount: 12
  maxFailedTrialCount: 3
  objective:
    type: maximize
    goal: 0.99
    objectiveMetricName: Validation-accuracy
    additionalMetricNames:
      - accuracy
  algorithm:
    algorithmName: tpe
  parameters:
    - name: --learning_rate
      parameterType: double
      feasibleSpace:
        min: "0.01"
        max: "0.2"
    - name: --dropout
      parameterType: double
      feasibleSpace:
        min: "0.1"
        max: "0.5"
  trialTemplate:
    goTemplate:
        rawTemplate: |-
          apiVersion: batch/v1
          kind: Job
          metadata:
            name: {{.Trial}}
            namespace: {{.NameSpace}}
          spec:
            template:
              spec:
                containers:
                - name: {{.Trial}}
                  image: kangwoo/mnist:katib
                  command:
                  - "python3"
                  - "/app/mnist.py"
                  {{- with .HyperParameters}}
                  {{- range .}}
                  - "{{.Name}}={{.Value}}"
                  {{- end}}
                  {{- end}}
                restartPolicy: Never

정의한 Experiment 사용자 리소스를 쿠버네티스 클러스터에 생성합니다.

kubectl apply -f tpe-stdout-example.yaml

Experiment 결과 보기

Katib UI를 통해서 다음과 같은 결과를 확인할 수 있습니다.

이전 : Kubeflow – Katib 소개

다음 : Kubeflow – Katib : Metrics Collector

작성일자

Kubeflow – Katib 소개

Katib 살펴보기

Katib는 Kubeflow 컴포넌트로서, 하이퍼 파라미터(Hyperparameter) 튜닝 및 신경망 아키텍처 탐색(Neural Architecture Search)을 위한 쿠버네티스 기반의 시스템입니다. Katib는 TensorFlow, PyTorch, Apache MXNet, XGBoost 등 다양한 머신러닝 프레임워크를 지원합니다.

Kubeflow의 컴포넌트인 Katib 시스템에 대해서 이해하고, 하이퍼 파라미터 튜닝하는 방법에 대해서 알아 보도록 하겠습니다. 예제에 사용한 Katib 버전은 0.8 입니다.

Katib의 개념

Katib에는 실험(Experiment), 제안(Suggestion), 시도(Trial) 및 작업(Job) 이라는 개념이 있습니다.

Experiment

Experiment 란 목표로 하는 대상 값을 찾기 위해서, 하이퍼 파라미터 값들을 찾는 일련의 탐색 작업을 의미합니다. Experiment에는 다음과 같은 구성 요소가 포함되어 있습니다

  • 목표 (Objective) : 하이퍼 파라미터 튜닝 작업 통해서, 이루고자 하는 목표를 정의해야 합니다. 예를 든다면, 모델의 정확성(accuracy)의 최대값을 0.91 로 목표로 한다고 정의할 수 있습니다
  • 탐색 범위 (Search Space) : 하이퍼 파라미터 튜닝 작업시 사용해야 할 모든 하이퍼 파라미터 값과 하이퍼 파라미터의 제약 조건을 정의해야 합니다. 예를 든다면, Learning rate는 0.1부터 0.5까지의 값을 사용하고, optimizer는 sgd와 adam을 사용한다고 정의할 수 있습니다.
  • 탐색 알고리즘 (Search Algorithm) : 하이퍼 파라미터 튜닝 작업시 사용할 알고리즘을 정의해야 합니다. Random Search, Grid Search, Bayesian Optimization 등 다용한 알고리즘을 사용할 수 있습니다.

Katib를 사용해서 최적의 하이퍼 파라미터를 찾으려면, Experiment 라는 사용자 리소스를 생성하면 됩니다.

Suggestion

Katib는 각 Experiment 사용자 리소스 마다 하나의 Suggestion 사용자 리소스를 생성합니다. Suggestion 은 탐색 알고리즘이 제안한 하이퍼 파라미터 값들의 세트를 가지고 있습니다. Katib 는 제안된 하이퍼 파라미터 값들을 세트별로 평가하기 위한 Trial을 작성합니다.

Trial

Trial은 제안된 하이퍼 파리미터 값들을 평가하기 위한 하나의 작업을 의미하는 사용자 리소스입니다. 제안된 매개 변수 값들을 Woker Job 인스턴스 할당해서 실행합니다.

각 Experiment 는 여러 번의 Trial을 수행합니다. Experiment 는 목표나 설정한 최대 시도 횟수에 도달 할 때까지 Trial 을 계속 실행합니다.

Worker Job

Worker Job은 Trial을 평가하고 목표 값을 계산하는 프로세스를 의미합니다. 제안된 하이퍼 파라미터 값들을 넘겨 받아서 실제로 모델을 학습하게 됩니다.

다음은 사용 가능한 Worker Job의 유형입니다.

  • Kubernetes Job
  • Kubeflow TFJob (분산 처리 지원)
  • Kubeflow PyTorchJob (분산 처리 지원)

Metrics Collector

하이퍼 파라미터 튜닝 작업 통해서, 목표로 하는 대상 값을 찾기 위해서는 대상 값을 수집하고 저장해야 합니다. Katib에서는 이러한 메트릭들을 저장하기 위해서 Metrics Collector를 사용합니다.

Job, TFJob, PytorchJob 등과 같은 실제 모델 학습을 진행하는 포드가 실행 될 때, 학습에 관련된 결과 값들을 수집하기 위해서 Metrics Collector가 포함된 Collector 컨테이너를 사이드카로 포드에 주입합니다. Collector 컨테이너는 메트릭 소스의 구문을 분석하여, Worker 컨테이너의 메트릭을 수집하고 Katib-manager의 katib-db 와 같은 영구 저장소에 메트릭을 저장합니다.

Katib에서 지원하고 있는 Metrics Collector는 다음과 같습니다.

  • StdOut : 운영 체제의 기본 출력인 StdOut으로 출력되는 메트릭을 수집합니다. 별도의 수집기를 정의하지 않으면 StdOut가 사용됩니다.
  • File : 지정한 파일을 이용해서 메트릭을 수집합니다. source 필드에 경로를 지정해야합니다.
  • TensorFlowEvent : 지정한 디렉토리에 저장된 tf.Event 를 이용해서 메트릭을 수집합니다. 현재는 텐서플로우 1 버전만 지원합니다. source 필드에 경로를 지정해야합니다.
  • Custom : 사용자가 정의한 메트릭 수집기를 사용합니다.
  • None : Katib의 메트릭 수집기를 사용하지 않을 때 사용합니다.

탐색 알고리즘

Katib에서 제공하는 탐색 알고리즘은 다음과 같습니다.

Hyperparameter Tuning

  • Grid Search (grid) : 그리드 탐색은 하이퍼 파라미터 최적화를 수행하는 전통적인 방법 중 하나로서, 하이퍼 파라미터 공간에서 수동으로 지정한 하위 집합을 모두 조합해서 전부 탐색하는 것을 말합니다. 이러한 작업은 학습 세트에 대한 교차 검증(cross-validation)이나 보류(held-out) 된 검증 세트에 대한 평가에 따라 진행됩니다. 균등한 공간의 시작점들로부터 시작해서, 이 점들의 목적 함수 값(objective functions)을 계산하여 최적의 조합을 선택하게 됩니다. 그리드 탐색은 모든 가능성에 대해 탐색을 수행하기 때문에, 중간 규모의 문제에 대해서도 탐색 프로세스를 매우 길게 만듭니다. 그래서 그리드 탐색은 만들어낼 수 있는 파라미터들의 탐색 조합이 적은 경우에만 유용하게 사용할 수 있습니다.
  • Random Search (random) : 무작위 탐색은 그리드 탐색의 대안으로서, 조합할 수 있는 파라미터의 수가 많을 때 사용하면 좋습니다. 무작위 탐색은 무작위로 파라미터를 선택하여 조합을 만들어냅니다. 하이퍼 파라미터 공간에서 수동으로 하위 집합을 지정할 필요가 없기 때문에 간단하게 적용 할 수 있습니다. 그렇기 때문에 무작위 탐색은 모든 가능성에 대한 탐색이 불가능할 때 사용하기 좋은 알고리즘입니다. Katib는 hyperopt 라는 최적화 프레임워크를 사용해서 무작위 탐색 알고리즘을 지원합니다.
  • Tree of Parzen Estimators (tpe) : Katib 는 hyperopt 를 사용해서 Tree of Parzen Estimators (TPE) 알고리즘을 지원합니다 . 이 방법은 “정방향 및 역방향 그라디언트 기반” 탐색을 제공합니다.
  • Hyperband (hyperbadn): 하이퍼밴드는 반복 알고리즘을 조정하는 비교적 새로운 방법으로서, 최적화 탐색 속도에 중점을 두었습니다. 리소스 할당을 최적화하여 평가 할 수 있는 조합의 수를 최대화 합니다. 그래서 빠르게 목적에 도달해서 해서 조기 중지(early stopping)에 이르게 하고 있습니다.
  • Bayesian Optimization (skopt-bayesian-optimization) : ‘베이지안 최적화’방법은 가우시안 프로세스 회귀를 사용하여 탐색 공간을 모델링합니다. 이 기법은 탐색 공간의 모든 지점에서 손실 함수의 추정치와 해당 추정치의 불확실성을 계산합니다. 즉, 현재 모델을 기반으로 유망한 하이퍼 파라미터 구성을 반복적으로 평가해서, 최적의 위치에 대한 정보를 나타내는 관측치를 수집하는 등의 확률적 추정 결과를 바탕으로 최적의 값을 찾습니다. 이 방법은 탐색 공간의 차원 수가 적은 경우에 적합합니다. 이 방법은 예상 손실과 불확실성을 모두 모델링하므로 탐색 알고리즘이 몇 단계로 수렴되므로 , 매개 변수 구성 평가를 완료하는 데 시간이 오래 걸릴 경우 사용하면 좋습니다. Katib는  Scikit-Optimize (skopt) 라는 라이브러리를 사용해서 베이지안 탐색을 지원합니다.

Neural Architecture Search

Katib 구성 요소

Katib는 다음과 같은 구성 요소로 이루어져 있습니다.

  • katib-ui : 하이퍼 파라미터 튜닝을 실행하고 관리하기 위한 사용자 인터페이스 (UI).
  • katib-controller : Katib 사용자 리소스를 제어하기 위한 쿠버네티스 컨트롤러.
  • katib-db-manager: DB 인터페이스인 Katib의 GRPC API 서버.
  • katib-mysql : Katib의 데이터를 저장하기 위한 MySql 데이터베이스.

Katib UI 접속하기

Katib 사용자 인터페이스를 사용하면, Experiment 을 제출하고 결과를 조회 해 볼 수 수 있습니다.

다음은 Kubeflow 에 있는 Katib UI 화면입니다.

Kubeflow 대시보드 화면의 왼쪽 메뉴에서 Katib를 클릭하면 접속할 수 있습니다.

다음 : Kubeflow – Katib 하이퍼 파라미터 튜닝

작성일자

Kubeflow – Fairing

Kubeflow Fairing 살펴보기

Kubeflow Fairing은 머신러닝 모델을 쉽게 학습하고, 배포할 수 있도록 도와주는 파이썬(Python) 패키지입니다. fairing을 사용하면, 파이썬 코드나 주피터 노트북에서 원격으로 Kubeflow 클러스터에 학습(training) 작업을 실행시킬 수 있습니다. 그리고 학습 완료된 모델을 배포할 수 있도록 도와줍니다.

다음은 kubeflow 사이트에 올라와 있는,  Kubeflow Fairing project가 추구하는 목표입니다.

  • Easily package ML training jobs: ML 실무자가 ML 모델 학습 코드와 코드의 종속성을 도커 이미지로 쉽게 패키징 할 수 있습니다.
  • Easily train ML models in a hybrid cloud environment: 기본 인프라를 이해할 필요없이 클라우드에서 교육 작업을 쉽게 실행할 수 있도록 ML 모델 교육을 위한 고급 API 제공합니다.
  • Streamline the process of deploying a trained model: ML 실무자가 학습된 ML 모델을 하이브리드 클라우드 환경에 쉽게 배포 할 수 있도록 합니다.

Fairing은 파이썬 파일이나, 주피터 노트북 등을 컨테이너 이미지로 만든 다음, 해당 이미지를 kubeflow 클러스터에 배포해서 모델 학습 작업을 실행할 수 있도록 도와줍니다. 그리고 학습 완료된 모델은 쉽게 서빙할 수 있는 기능독 제공하고 있습니다.

Kubeflow Fairing 개념

Kubeflow Fairing 에는 세 가지의 중요한 개념이 있습니다. 바로 preprocessor, builder와 deployer 입니다.

Preprocessor (전처리기)

전처리기는 Kubeflow Fairing이 학습 작업을 위한 컨테이너 이미지를 만들 때, 이미지 생성에 필요한 일련의 정보들을 정의하는 역할을 합니다. 전처리기를 사용하면 컨테이너 이미지에 들어갈 입력 파일을 선택할 수 있고, 변환할 수 있으며, 불필요한 파일을 제외 시킬 수도 있습니다. 그리고 컨테이너 이미지의 엔트리포인터(entrypoint) 같은 설정 정보도 변경할 수 있습니다.

Kubeflow Fairing 에는 제공하는 전처리기는 다음과 같습니다.

  • python : 입력 파일을 컨테이너 이미지에 직접 복사합니다.
  • notebook : 노트북을 실행 가능한 파이썬 파일로 변환합니다. 그리고 노트북 코드에서 파이썬 코드가 아닌 부분을 제거합니다.
  • full_notebook : 파이썬 코드가 아닌 부분들을 포함해서 전체 노트북을 그대로 실행합니다. 별다른 설정이 없다면, 노트북 실행에 papermill을 사용합니다.
  • function : FunctionPreProcessor는 단일 함수를 전처리합니다. function_shim.py을 사용하여 함수를 직접 호출합니다.

Builder (빌더)

빌더는 Kubeflow Fairing이 학습 작업에 사용할 컨테이너 이미지를 빌드하는 방법 및 컨테이너 이미지를 저장할 컨테이너 레지스트리의 위치를 ​​정의하는 역할을 합니다.

Kubeflow Fairing 에는 제공하는 빌더는 다음과 같습니다.

  • append : 기존 컨테이너 이미지를 바탕으로, 코드를 새 레이어로 추가합니다. 이 빌더는 기본 이미지를 가져 와서 이미지를 작성하지 않고, 추가된 부분만 컨테이너 이미지 레지스트리에 푸시합니다. 그래서 학습 작업을 위한 컨테이너 이미지를 작성하는 데 시간이 상대적으로 적게 소모됩니다. 그리고 파이썬 라이브러인 containerregistry을 사용하기 때문에, 도커 데몬이 필요 없습니다.
  • docker : 로컬 도커 데몬을 사용하여, 학습 작업에 사용할 컨테이너 이미지를 빌드하고, 컨테이너 이미지 레지스트리에 푸시합니다
  • cluster : 쿠버네티스 클러스터에서 학습 작업에 사용할 컨테이너 이미지를 빌드하고, 컨테이너 이미지 레지스트리에 푸시합니다

Deployer (배포자)

배포자는 Kubeflow Fairing이 학습 작업에 사용할 컨테이너 이미지를 배포하고 실행할 위치를 정의하는 역할을 합니다. 배포자는 빌더가 생성한 이미지를 사용하여 Kubeflow 클러스터에 학습 작업을 배포하고 실행합니다

Kubeflow Fairing 에는 제공하는 배포자는 다음과 같습니다.

  • Job : 쿠버네티스 Job 리소스를 사용하여 학습 작업을 시작합니다.
  • TfJob : Kubeflow의 TFJob 컴포넌트를 사용하여 텐서플로우 학습 작업을 시작합니다.
  • PyTorchJob : Kubeflow의 PyTorchJob 컴포넌트를 사용하여 PyTorch 학습 작업을 시작합니다.
  • GCPJob : GCP에게 학습 작업 보냅니다.
  • Serving : 쿠버네티스의 디플로이먼트(deployment)와 서비스(service)를 사용하여, 예측(prediction) 엔드포인트를 서빙합니다.
  • KFServing : KFServing을 사용하여, 예측(prediction) 엔드포인트를 서빙합니다.

Kubeflow Fairing 사용하기

Kubeflow Fairing을 사용하여, ML 모델을 학습하고 배포하는 방법에 대해서 알아보도록 하겠습니다. Kubeflow Fairing은 파이썬 패키지이므로 로컬 개발 환경 및 주피터 노트북 어디서나 사용할 수 있습니다.

로컬 개발 환경에 Kubeflow Fairing 설치하기

다음은 로컬 개발 환경에서 Kubeflow Fairing 을 설치하는 방법에 대해 알아보도록 하겠습니다.

파이썬 설치하기

Kubeflow Fairing을 사용하려면 Python 3.6 이상이 필요합니다. 다음 명령을 실행해서, Python 3.6 이상이 설치되어 있는지 확인해 보겠습니다.

python3 -V

파이썬이 설치되어 있다면, 다음과 같은 응답 결과를 얻을 수 있습니다.

Python 3.6.9

Python 3.6 이상이 없는 경우 Python Software Foundation에서 Python을 다운로드 할 수 있습니다.

가상 환경 만들기

가상 환경을 만들어서 Kubeflow Fairing을 설치를 진행할 것입니다. 가상 환경을 만들기 위해서 virtualenv가 설치되어 있어야합니다. 다음 명령을 실행해서, virtualenv가 설치되어 있는지 확인해 보겠습니다.

which virtualenv

virtualenv가 설치되어 있다면, 다음과 같은 응답 결과를 얻을 수 있습니다.

/usr/local/bin/virtualenv

virtualenv가 설치되어 있지 않다면, pip3 를 이용해서 virtualenv를 설치합니다.

pip3 install virtualenv

가상 환경을 새로 만들겠습니다.

virtualenv venv --python=python3

그리고, 새로 만든 가상 환경을 활성화 하겠습니다.

source venv/bin/activate

Kubeflow Fairing 설치하기

활성화한 가상 환경에, Kubeflow Fairing을 설치하겠습니다.

pip install kubeflow-fairing

설치가 완료되면 Fairing 파이썬 패키지를 사용할 수 있습니다. 다음 명령을 실행해서, Kubeflow Fairing이 설치되어 있는지 확인해 보겠습니다.

pip show kubeflow-fairing

Kubeflow Fairing이 설치되어 있다면, 다음과 같은 응답 결과를 얻을 수 있습니다.

Name: kubeflow-fairing
Version: 0.7.1
Summary: Kubeflow Fairing Python SDK.
Home-page: <https://github.com/kubeflow/fairing>
Author: Kubeflow Authors
Author-email: hejinchi@cn.ibm.com
License: Apache License Version 2.0
Location: /Users/kangwoo/Documents/kubeflow/fairing/venv/lib/python3.7/site-packages
Requires: python-dateutil, six, google-api-python-client, google-cloud-storage, notebook, kubernetes, kfserving, boto3, httplib2, requests, google-auth, numpy, azure, docker, future, kubeflow-tfjob, kubeflow-pytorchjob, setuptools, urllib3, cloudpickle, retrying, tornado, google-cloud-logging, oauth2client
Required-by:

Docker 설치

Kubeflow Fairing의 docker 빌더를 사용하기 위해서는 도커가 필요합니다. 다른 빌더를 사용하시리면, 설치하지 않으셔도 됩니다.

다음 명령을 실행해서, Docker가 설치되어 있는지 확인해 보겠습니다.

docker version

docker가 설치되어 있지 않다면, docker를 설치합니다.

로컬 개발 환경에 Kubeflow Fairing 설정하기

로컬 개발 환경에서 Kubeflow Fairing을 사용하여 Kubeflow 클러스터에서 머신러닝 모델을 학습하거나 배포하려면, 컨테이너 이미지 레지스트리와 Kubeflow 클러스터에 접근 할 수 있는 환경을 구성해야합니다.

로컬 개발 환경에서 컨테이너 이미지 레지스트리에 접근할 수 있도록 Docker 설정하기

로컬 개발 환경에 있는 Kubeflow Fairing에서 생성한 컨테이너 이미지를 컨테이너 이미지 레지스트리에 푸시 하려면 접근 권한이 필요합니다.

컨테이너 이미지 레지스트리를 기본 저장소인 도커허브(Docker Hub)를 사용하고 있다면, docker login 명령어를 실행해서 도커가 컨테이너 이미지 레지스트리에 액세스 할 수 있도록 권한을 부여할 수 있습니다. 물론 로그인 하기 전에 도커허브 계정을 가지고 있어야 합니다.

docker login
Login with your Docker ID to push and pull images from Docker Hub. If you don't have a Docker ID, head over to <https://hub.docker.com> to create one.
Username: kangwoo
Password: 
Login Succeeded

만약 다른 개인 레지스트리를 사용하고 있다면, 로그인을 할 때 레지스트리 주소를 입력하시면 됩니다.

다음은 docker-registry.foo.bar 라고 하는 개인 레지스트리에 로그인 하는 명령어입니다.

docker login [docker-registry.foo.bar](<http://docker-registry.foo.bar>)

로그인이 성공적으로 되면, ~/.docker/config.json 에 파일이 생성된 것을 확인할 수 있습니다.

Kubeflow 클러스터에서 컨테이너 이미지 레지스트리에 접근할 수 있도록 쿠버네티스 설정하기

Kubeflow가 설치된 쿠버네티스에서 Kubeflow Fairing에서 생성한 컨테이너 이미지를 내려 받아야합니다. 도커허브의 공개 저장소 같은 경우에는 이미지를 내려 받는데에 별도의 권한이 필요없습니다. 하지만 개인 레지스트리를 사용하는 경우라면, 접근 권한이 필요합니다.

만약 config.json 파일이 아래와 같은 형식이라면, 이 파일을 그대로 사용할 수 있습니다.

{
  "auths": {
      "<https://index.docker.io/v1/>": {
          "auth": "XXXXX"
      }
  }
}

하지만 아래처럼 인증 정보를 별도로 저장하는 형식이라면, config.json 파일을 직접 만들어야 합니다.

{
	"auths": {
		"<https://index.docker.io/v1/>": {}
	},
	"HttpHeaders": {
		"User-Agent": "Docker-Client/19.03.5 (darwin)"
	},
	"credsStore": "desktop",
	"experimental": "disabled",
	"stackOrchestrator": "swarm"
}

직접 config.json 파일을 만들기

직접 config.json 파일을 만들려면 auth 필드에 base64 인코딩된 username:password 문자열 값을 설정하면 됩니다.

다음 명령어를 사용해서 auth 값을 생성할 수 있습니다.

echo -n 'username:password' | base64

만약 도커 허브가 아니라 별도의 컨테이너 이미지 레지스트리를 사용한다면 “https://index.docker.io/v1/” 값 대신에 해당 레지스트리 주소를 설정하면 됩니다.

다음은 도커 허브를 사용하고, 사용자명이 username, 비밀번호가 password 인 config.json 예제 파일입니다.

{
  "auths": {
      "<https://index.docker.io/v1/>": {
          "auth": "dXNlcm5hbWU6cGFzc3dvcmQ="
      }
  }
}

컨테이너 레지스트리 접속을 위한 secret 리소스 만들기

쿠버네티스 클러스터는 개인 컨테이너 레지스트리로 접근하기 위해서, 도커-레지스트리의 인증 형식을 사용합니다.

docker login을 실행 또는 직접 $HOME/.docker/config.json 파일을 생성했다면, 해당 자격 증명을 Kubernetes에 복사 할 수 있습니다

다음은 admin이라는 네임스페이스에 ~/.docker/config.json 의 파일을 사용해서 regcred 이름의 secret 리소스를 생성하는 명령어입니다.

kubectl -n admin create secret generic regcred \\
    --from-file=.dockerconfigjson=~/.docker/config.json \\
    --type=kubernetes.io/dockerconfigjson

생성한 secret을 이미지를 내려 받을 때 사용하도록 네임스페이스의 서비스 계정을 수정하겠습니다.

쿠버네티스 네임스페이스 있는 default 서비스 계정과, Kubeflow에서 생성한 default-editor, default-viewer 계정을 모두 수정하겠습니다.

다음은 admin이라는 네임스페이스에 있는 default, default-editor, default-viewer 계정에, 컨테이너 이미지를 내려 받을 때 regcred 이름의 secret 리소스를 사용하라고 수정하는 명령어입니다.

kubectl -n admin patch serviceaccount default -p '{"imagePullSecrets": [{"name": "regcred"}]}'
kubectl -n admin patch serviceaccount default-editor -p '{"imagePullSecrets": [{"name": "regcred"}]}'
kubectl -n admin patch serviceaccount default-viewer -p '{"imagePullSecrets": [{"name": "regcred"}]}'

로컬 환경에서 Kubeflow 클러스터에 접근할 있도록 kubeconfig 설정하기

로컬 개발 환경에 있는 Kubeflow Fairing에서 Kubeflow가 설치된 쿠버네티스 클러스터에 작업을 배포하려면, 클러스터 접근 권한이 필요합니다. 설정이 되어 있지 않다면, 쿠버네티스 클러스의 ~/.kube/config 파일을 로컬 환경의 ~/.kube/config로 복사하면 됩니다.

원칙적으로는 필요한 권한만 부여된 계정을 만들어서 사용해야하지만, 편의를 위해서 cluster-admin 권한이 있는 인증 파일을 그대로 사용한 것입니다.

로컬 개발 환경에 Kubeflow Fairing 사용하기

로컬 개발 환경서 Kubeflow Fairing을 사용해보도록 하겠습니다. mnist 숫자를 분류하는 간단한 모델을 컨테이너 이미지로 빌드하고, Kubeflow 클러스터에 job 형태로 배포해서 학습하는 것을 하도록 하겠습니다.

“docker” 빌더를 사용해서 모델 학습하기

먼저 학습에 사용할 모델 파일을 만들겠습니다.

다음은 mnist 숫자를 분류하는 모델을 텐서플로우 케라스로 작성한 코드입니다.

mnist-simple.py

from __future__ import absolute_import, division, print_function, unicode_literals

import tensorflow as tf
import numpy as np


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)

    print("Average test loss: ", np.average(training_history.history['loss']))


if __name__ == '__main__':
    train()

컨테이너 이미지를 만들기 위한 Dockerfile을 생성하겠습니다.

다음은 텐서플로우 2.1을 기반 이미지로 해서, 모델 파일을 추가하는 Dockerfile 입니다.

Dockerfile

FROM tensorflow/tensorflow:2.1.0-py3

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

컨테이너 이미지를 빌드하고, 배포하기 위한 fairing 코드 파일을 생성하겠습니다.

fairing-local-docker.py

import uuid
from kubeflow import fairing

CONTAINER_REGISTRY = 'kangwoo'

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

command=["python", "mnist-simple.py"]
output_map = {
    "Dockerfile": "Dockerfile",
    "mnist-simple.py": "mnist-simple.py"
}

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

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

fairing.config.set_deployer('job', namespace=namespace, job_name=job_name, cleanup=False, stream_log=False)

fairing.config.run()

“CONTAINER_REGISTRY”에는 사용할 컨테이너 이미지 레지스트리를 입력하시면 됩니다. 예제에서는 도커허브를 사용하기 때문에 계정명인 kangwoo 를 사용하였습니다.

“namespace” 에는 모델 작업을 배포한 쿠버네티스 네임스페이스를 입력하시면 됩니다. 예제에서는 admin 이라는 네임스페이스를 사용하였습니다.

다음은 전처리기입니다.

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

파이썬 파일을 사용하기 때문에, python 전처리기를 사용하였습니다. 그리고 컨테이너 이미지를 빌드하기 위한 파일들을 빌더로 넘겨주기 위해서 output_map을 설정하였습니다. Dockerfile과 mnist-simple.py 파일이 빌더에서 사용됩니다. 학습 작업을 실행하기 위한 명령어를 command로 설정하였습니다.

다음은 빌더입니다.

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

docker 빌더를 사용하였습니다. 사용할 컨테이너 이미지 레지스트리 주소를 registry 에 설정합니다. 이미지 이름을 image_name에 설정합니다. 그리고 빌드에 사용할 Dockerfile을 dockerfile_path에 설정합니다.

다음은 배포자입니다

fairing.config.set_deployer('job', namespace=namespace, job_name=job_name, cleanup=False, stream_log=False)

job 배포자를 사용하였습니다. 작업이 배포되는 네임스페이스 이름을 namespace에 설정합니다. 그리고 작업 이름을 job_name에 설정합니다.

cleanup 파라메터는 작업을 완료하였을때, 작업에 사용한 포드들을 삭제할 여부를 결정합니다. 기본값을 True 입니다. 예제에서는 포드를 조회하기 위해서 False로 설정하였습니다.

stream_log 파라메터는 쿠버네티스에서 실행되는 작업들의 포드 로그를 fairing을 실행하는 터미널에 출력할지를 결정합니다. 기본값을 True 입니다. 예제에서는 False로 설정하였습니다.

이제 작성한 fairing 코드를 실행해 보겠습니다.

python fairing-local-docker.py

정상적으로 실행되면 다음과 같은 응답 화면을 볼 수 있습니다.

[I 200307 10:29:50 config:125] Using preprocessor: <kubeflow.fairing.preprocessors.base.BasePreProcessor object at 0x1082a9b10>
[I 200307 10:29:50 config:127] Using builder: <kubeflow.fairing.builders.docker.docker.DockerBuilder object at 0x10d4223d0>
[I 200307 10:29:50 config:129] Using deployer: <kubeflow.fairing.deployers.job.job.Job object at 0x1082a9b50>
[I 200307 10:29:50 docker:32] Building image using docker
[W 200307 10:29:50 docker:41] Docker command: ['python', 'mnist-simple.py']
[I 200307 10:29:50 base:107] Creating docker context: /tmp/fairing_context_zq4eaqjt
[W 200307 10:29:50 docker:56] Building docker image kangwoo/mnist-simple:501087B0...
...
[I 200307 10:30:09 docker:103] Push output: 501087B0: digest: sha256:dd3f7146d3b66ff44d0c7f5670e07fb9fd621b9da136987cb2eb95fe8eb9795c size: 2831 None
[I 200307 10:30:09 docker:103] Push finished: {'Tag': '501087B0', 'Digest': 'sha256:dd3f7146d3b66ff44d0c7f5670e07fb9fd621b9da136987cb2eb95fe8eb9795c', 'Size': 2831}
[W 200307 10:30:09 job:90] The job mnist-job-0a3bd86kp launched.

Job 이름은 “mnist-job-0a3bd86kp” 라는 것을 확인 할 수 있습니다.

다음 명령어를 실행하면 admin 네임스페이스의 Job 목록을 확인할 수 있습니다.

kubectl -n admin get job

정상적으로 실행되면 다음과 같은 응답 화면을 볼 수 있습니다.

NAME                  COMPLETIONS   DURATION   AGE
mnist-job-0a3bd86kp   1/1           18s        7m25s

다음 명령어를 실행하면 mnist-job-0a3bd86kp 라는 이름을 가진 Job이 생성한 포드를 확인할 수 있습니다.

kubectl -n admin get pod -l job-name=mnist-job-0a3bd86kp

정상적으로 실행되면 다음과 같은 응답 화면을 볼 수 있습니다.

NAME                        READY   STATUS      RESTARTS   AGE
mnist-job-0a3bd86kp-6vst8   0/1     Completed   0          8m39s

학습 작업 삭제하기

작업이 완료되어도 Job은 삭제되지 않습니다.

다음 명령어를 실행하면 admin 네임스페이스의 mnist-job-0a3bd86kp 라는 이름의 Job을 삭제할 수 있습니다.

kubectl -n admin delete job mnist-job-0a3bd86kp

“append” 빌더를 사용해서 모델 학습하기

먼저 학습에 사용할 모델 파일을 만들겠습니다.

“docker” 빌더를 사용해서 모델 학습하기 에서 사용한것과 동일한 코드입니다.

mnist-simple.py

from __future__ import absolute_import, division, print_function, unicode_literals

import tensorflow as tf
import numpy as np


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)

    print("Average test loss: ", np.average(training_history.history['loss']))


if __name__ == '__main__':
    train()

컨테이너 이미지를 빌드하고, 배포하기 위한 fairing 코드 파일을 생성하겠습니다.

fairing-local-docker.py

import uuid
from kubeflow import fairing

CONTAINER_REGISTRY = 'kangwoo'

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

fairing.config.set_preprocessor('python', executable="mnist-simple.py")

fairing.config.set_builder('append', registry=CONTAINER_REGISTRY, image_name="mnist-simple",
                           base_image="tensorflow/tensorflow:2.1.0-py3")

fairing.config.set_deployer('job', namespace=namespace, job_name=job_name, cleanup=False, stream_log=True)

fairing.config.run()

가장 큰 변경 사항은 Dockerfile이 없어진 것과, 빌더 부분에 “base_image” 파라메터가 추가 된 것입니다.

다음은 전처리기입니다.

fairing.config.set_preprocessor('python', executable="mnist-simple.py")

파이썬 파일을 사용하기 때문에, python 전처리기를 사용하였습니다. 그리고 실행할 파이썬 파일을 executable에 설정하였습니다. executable에 추가한 파일은 자동으로 input_files에 추가 되고 컨테이너 이미지에 추가됩니다. 그리고 학습 작업을 실행하기 위한 명령어에 자동으로 추가됩니다.

다음은 빌더입니다.

fairing.config.set_builder('append', registry=CONTAINER_REGISTRY, image_name="mnist-simple",
                           base_image="tensorflow/tensorflow:2.1.0-py3")

appender 빌더를 사용하였습니다. docker 빌더와는 다른게 Dockerfile을 정의해 줄 필요가 없습니다. 대신 기본 이미지를 base_image 에서 설정해주어야합니다..

다음은 배포자입니다

fairing.config.set_deployer('job', namespace=namespace, job_name=job_name, cleanup=False, stream_log=True)

배포자는 변경된 부분이 없습니다.

이제 작성한 fairing 코드를 실행해 보겠습니다.

python fairing-local-append.py

정상적으로 실행되면 다음과 같은 응답 화면을 볼 수 있습니다.

[I 200307 12:52:04 config:125] Using preprocessor: <kubeflow.fairing.preprocessors.base.BasePreProcessor object at 0x100597950>
[I 200307 12:52:04 config:127] Using builder: <kubeflow.fairing.builders.append.append.AppendBuilder object at 0x100597990>
[I 200307 12:52:04 config:129] Using deployer: <kubeflow.fairing.deployers.job.job.Job object at 0x100597ad0>
[W 200307 12:52:04 append:50] Building image using Append builder...
[I 200307 12:52:04 base:107] Creating docker context: /tmp/fairing_context_9qe8ujn6
[I 200307 12:52:04 docker_creds_:234] Loading Docker credentials for repository 'tensorflow/tensorflow:2.1.0-py3'
[I 200307 12:52:05 docker_creds_:152] Invoking 'docker-credential-desktop' to obtain Docker credentials.
[I 200307 12:52:05 docker_creds_:175] Successfully obtained Docker credentials.
[W 200307 12:52:07 append:54] Image successfully built in 2.6276446930000006s.
[W 200307 12:52:07 append:94] Pushing image kangwoo/mnist-simple:645ECCC9...
...
[W 200307 12:52:12 append:99] Pushed image kangwoo/mnist-simple:645ECCC9 in 5.018134577s.
[W 200307 12:52:12 job:90] The job mnist-job-f7f8mlrr5 launched.

Job 이름은 “mnist-job-f7f8mlrr5” 라는 것을 확인 할 수 있습니다.

다음 명령어를 실행하면 admin 네임스페이스의 Job 목록을 확인할 수 있습니다.

kubectl -n admin get job

정상적으로 실행되면 다음과 같은 응답 화면을 볼 수 있습니다.

NAME                  COMPLETIONS   DURATION   AGE
mnist-job-f7f8mlrr5   1/1           18s        7m25s

다음 명령어를 실행하면 mnist-job-f7f8mlrr5 라는 이름을 가진 Job이 생성한 포드를 확인할 수 있습니다.

kubectl -n admin get pod -l job-name=mnist-job-f7f8mlrr5

정상적으로 실행되면 다음과 같은 응답 화면을 볼 수 있습니다.

NAME                        READY   STATUS      RESTARTS   AGE
mnist-job-f7f8mlrr5-hjkx2   0/1     Completed   0          93s

학습 작업 삭제하기

작업이 완료되어도 Job은 삭제되지 않습니다.

다음 명령어를 실행하면 admin 네임스페이스의 mnist-job-f7f8mlrr5 라는 이름의 Job을 삭제할 수 있습니다.

kubectl -n admin delete job mnist-job-f7f8mlrr5

모델 코드 파일에서 Kubeflow Fairing 사용하기

지금까지 살펴본 예제는 모델 코드 파일과 fairing 코드 파일이 따로 분리 되어 있었습니다. 이번에는 모델 코드 안에서 fairing을 사용하는 방법을 알아보도록 하겠습니다.

다음은 모델 코드와 fairing 코드가 같이 들어있는 코드입니다.

mnist-with-fairing.py

from __future__ import absolute_import, division, print_function, unicode_literals

import os

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)

    print("Average test loss: ", np.average(training_history.history['loss']))


def fairing_run():
    CONTAINER_REGISTRY = 'kangwoo'

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

		fairing.config.set_preprocessor('python', executable="mnist-simple.py")

    fairing.config.set_builder('append', registry=CONTAINER_REGISTRY, image_name="mnist-simple",
                               base_image="tensorflow/tensorflow:2.1.0-py3")

    fairing.config.set_deployer('job', namespace=namespace, job_name=job_name, cleanup=False, stream_log=True)

    fairing.config.run()


if __name__ == '__main__':
    if os.getenv('FAIRING_RUNTIME', None) is None:
        import uuid
        from kubeflow import fairing
        fairing_run()
    else:
        import tensorflow as tf
        import numpy as np

        train()

다음은 “FAIRING_RUNTIME”으로 분기를 타는 부분입니다.

if __name__ == '__main__':
    if os.getenv('FAIRING_RUNTIME', None) is None:
        import uuid
        from kubeflow import fairing
        fairing_run()
    else:
        import tensorflow as tf
        import numpy as np

        train()

모델 코드와 fairing 코드가 같이 들어 있을 경우에는, 모델 학습을 위해서 코드가 실행된 것인지, fairing을 사용하기 위해서 실행된 것인지를 구별해 줄 필요가 있습니다. 그래서 fairing에서는 FAIRING_RUNTIME 라는 환경 변수를 제공하고 있습니다. fairing에서 실행하는 작업은 컨테이너의 환경 변수에 아래처럼 FAIRING_RUNTIME 환경 변수가 추가 됩니다. 이 값을 가지고 모델 학습을 위해서 코드가 실행된 것인지, fairing을 사용하기 위해서 실행된 것인지를 구별할 수 있습니다.

env:
    - name: FAIRING_RUNTIME
      value: "1"

이제 작성한 fairing 코드를 실행해 보겠습니다.

python mnist-with-fairing.py

정상적으로 실행되면 다음과 같은 응답 화면을 볼 수 있습니다.

[I 200307 13:07:20 config:125] Using preprocessor: <kubeflow.fairing.preprocessors.base.BasePreProcessor object at 0x10b7d2c10>
[I 200307 13:07:20 config:127] Using builder: <kubeflow.fairing.builders.append.append.AppendBuilder object at 0x10b815490>
[I 200307 13:07:20 config:129] Using deployer: <kubeflow.fairing.deployers.job.job.Job object at 0x10b815a90>
[W 200307 13:07:20 append:50] Building image using Append builder...
[I 200307 13:07:20 base:107] Creating docker context: /tmp/fairing_context_1pz0hi3k
[I 200307 13:07:20 docker_creds_:234] Loading Docker credentials for repository 'tensorflow/tensorflow:2.1.0-py3'
[I 200307 13:07:21 docker_creds_:152] Invoking 'docker-credential-desktop' to obtain Docker credentials.
[I 200307 13:07:21 docker_creds_:175] Successfully obtained Docker credentials.
[W 200307 13:07:23 append:54] Image successfully built in 2.8729543819999996s.
[W 200307 13:07:23 append:94] Pushing image kangwoo/mnist-simple:83A2C3C4...
...
[W 200307 13:07:28 append:99] Pushed image kangwoo/mnist-simple:83A2C3C4 in 5.097283148000001s.
[W 200307 13:07:28 job:90] The job mnist-job-139bf5rrm launched.

Job 이름은 “mnist-job-f7f8mlrr5” 라는 것을 확인 할 수 있습니다.

주피터 노트북에서 Kubeflow Fairing 설치하기

다음은 주피터 노트북 환경에서 Kubeflow Fairing 을 설치하는 방법에 대해 알아보도록 하겠습니다.

Kubeflow 에서 제공하는 기본 주피터 컨테이너 이미지를 사용하는 경우, Kubeflow Fairing 환경이 사전 구성되어 있으므로, 추가 설치를 진행 하지 않아도 됩니다.

하지만 불행히도 Kubeflow 1.0 버전을 릴리즈 할 때 만들어진, 기본 주피터 컨테이너 이미지에는 예전 버전의 Fairing (fairing 0.5)을 사용합니다. 패키지 이름도 kubeflow-fairing이 아니라 fairing 입니다. 그래서 fairing이 설치되어 있다면, kubeflow-fairing을 설치해주시기 바랍니다.

파이썬 버전 확인하기

Kubeflow Fairing을 사용하려면 Python 3.6 이상이 필요합니다. 주피터 노트북에서 Python 3.6 이상이 설치되어 있는지 확인해 보겠습니다.

Jupyter 노트북 사용자 인터페이스의 메뉴에서 File > New > Terminal 을 클릭하여 노트북 환경에서 새 터미널 세션을 시작하십시오.

python3 -V

파이썬이 설치되어 있다면, 다음과 같은 응답 결과를 얻을 수 있습니다.

Python 3.6.9

Python 3.6 이상이 없는 경우, Python 3.6 이상이 설치된 주피터 이미지 사용하시거나, 별도로 생성하셔야 합니다.

Kubeflow Fairing 설치하기

노트북 터미널 세션에서, Kubeflow Fairing을 설치하겠습니다.

pip install kubeflow-fairing

설치가 완료되면 Fairing 파이썬 패키지를 사용할 수 있습니다. 다음 명령을 실행해서, Kubeflow Fairing이 설치되어 있는지 확인해 보겠습니다.

pip show kubeflow-fairing

Kubeflow Fairing이 설치되어 있다면, 다음과 같은 응답 결과를 얻을 수 있습니다.

Name: kubeflow-fairing
Version: 0.7.1
Summary: Kubeflow Fairing Python SDK.
Home-page: <https://github.com/kubeflow/fairing>
Author: Kubeflow Authors
Author-email: hejinchi@cn.ibm.com
License: Apache License Version 2.0
Location: /Users/kangwoo/Documents/kubeflow/fairing/venv/lib/python3.7/site-packages
Requires: python-dateutil, six, google-api-python-client, google-cloud-storage, notebook, kubernetes, kfserving, boto3, httplib2, requests, google-auth, numpy, azure, docker, future, kubeflow-tfjob, kubeflow-pytorchjob, setuptools, urllib3, cloudpickle, retrying, tornado, google-cloud-logging, oauth2client
Required-by:

주피터 노트북에서 Kubeflow Fairing 설정하기

Kubeflow 주피터 노트북 환경에서 Kubeflow Fairing을 사용하여 Kubeflow 클러스터에서 머신러닝 모델을 학습하거나 배포하려면, 컨테이너 이미지 레지스트리에 접근 할 수 있는 환경을 구성해야합니다.

Kubeflow 주피터 노트북 환경에서 컨테이너 이미지 레지스트리에 접근할 수 있도록 Docker 설정하기

주피터 노트북 환경에서 Kubeflow Fairing이 생성한 컨테이너 이미지를 컨테이너 이미지 레지스트리에 푸시 하려면 접근 권한이 필요합니다.

생성된 config.json 사용하기

로컬 개발 환경에서 도커 로그인을 한 적이 있으면, 로컬 개발 환경의 ~/.docker/config.json 파일을 조회해 봅니다.

만약 config.json 파일이 아래와 같은 형식이라면, 이 파일을 주피터 노트북의 ~/.docker/config.json 으로 그대로 복사하면 됩니다.

{
  "auths": {
      "<https://index.docker.io/v1/>": {
          "auth": "XXXXX"
      }
  }
}

하지만 아래처럼 인증 정보를 별도로 저장하는 형식이라면, config.json 파일을 직접 만들어야 합니다.

{
	"auths": {
		"<https://index.docker.io/v1/>": {}
	},
	"HttpHeaders": {
		"User-Agent": "Docker-Client/19.03.5 (darwin)"
	},
	"credsStore": "desktop",
	"experimental": "disabled",
	"stackOrchestrator": "swarm"
}

직접 config.json 파일을 만들기

직접 config.json 파일을 만들려면 auth 필드에 base64 인코딩된 username:password 문자열 값을 설정하면 됩니다.

다음 명령어를 사용해서 auth 값을 생성할 수 있습니다.

echo -n 'username:password' | base64

만약 도커 허브가 아니라 별도의 컨테이너 이미지 레지스트리를 사용한다면 “https://index.docker.io/v1/” 값 대신에 해당 레지스트리 주소를 설정하면 됩니다.

다음은 도커 허브를 사용하고, 사용자명이 username, 비밀번호가 password 인 config.json 예제 파일입니다.

{
  "auths": {
      "<https://index.docker.io/v1/>": {
          "auth": "dXNlcm5hbWU6cGFzc3dvcmQ="
      }
  }
}

config.json 파일을 주피터 노트북으로 복사하기

주피터 노트북 사용자 인터페이스의 메뉴에서 File > New > Terminal 을 클릭하여 노트북 환경에서 새 터미널 세션을 시작하십시오.

bash
mkdir ~/.docker
cd ~/.docker
cat << EOF > config.json
{
  "auths": {
      "<https://index.docker.io/v1/>": {
          "auth": "a2FuZ3dvbzpnZWVuYTEx"
      }
  }
}
EOF

Kubeflow 클러스터에서 컨테이너 이미지 레지스트리에 접근할 수 있도록 쿠버네티스 설정하기

Kubeflow가 설치된 쿠버네티스에서 Kubeflow Fairing에서 생성한 컨테이너 이미지를 내려 받아야합니다. 도커허브의 공개 저장소 같은 경우에는 이미지를 내려 받는데에 별도의 권한이 필요없습니다. 하지만 개인 레지스트리를 사용하는 경우라면, 접근 권한이 필요합니다. 설정이 되어 있지 않다면, “LINK-쿠버네티스 클러스터에서 컨테이너 이미지 레지스트리에 접근할 수 있도록 설정하기”를 참고하기 바랍니다.

주피터 노트북 환경에서 Kubeflow Fairing 사용하기

Kubeflow 주피터 노트북 환경에서 Kubeflow Fairing을 사용해보도록 하겠습니다. mnist 숫자를 분류하는 간단한 모델을 컨테이너 이미지로 빌드하고, Kubeflow 클러스터에 job 형태로 배포해서 학습하는 것을 하도록 하겠습니다.

“notebook” 전처리기를 사용해서 모델 학습하기

주피터 노트북 사용자 인터페이스의 메뉴에서 File > New > Notebook 을 클릭하여 노트북 환경에서 새 노트북을 시작하십시오.

노트북 컬럼에 모델 코드와 faring 코드를 입력하겠습니다.

다음은 mnist 숫자를 분류하는 모델을 텐서플로우 케라스로 작성한 코드입니다.

import os

import tensorflow as tf
import numpy as np

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)

    print("Average test loss: ", np.average(training_history.history['loss']))

다음은 fairing 코드 입니다.

def fairing_run():
    CONTAINER_REGISTRY = 'kangwoo'

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


    fairing.config.set_builder('append', registry=CONTAINER_REGISTRY, image_name="mnist-simple",
                               base_image="tensorflow/tensorflow:2.1.0-py3")

    fairing.config.set_deployer('job', namespace=namespace, job_name=job_name, cleanup=False, stream_log=True)

    fairing.config.run()

특이하게도 전처리기를 설정하는 set_preprocessor() 구문이 없습니다. 노트북에서는 별도의 전처리기를 설정하지 않으면 “notebook” 전처리기가 사용됩니다. “notebook” 전처리기는 노트북의 파이썬 코드 셀을 파이썬 파일로 변경해 줍니다.

다음은 환경 변수에 따라 모델 훈련을 실행하거나, fairing을 실행하는 부분입니다.

if __name__ == '__main__':
    if os.getenv('FAIRING_RUNTIME', None) is None:
        import uuid
        from kubeflow import fairing
        fairing_run()
    else:
        train()

shift + enter 키를 눌러 셀 들을 실행시키면, 다음과 같은 응답 결과를 얻을 수 있습니다.

[I 200307 06:22:20 config:125] Using preprocessor: <kubeflow.fairing.preprocessors.converted_notebook.ConvertNotebookPreprocessor object at 0x7fc924284fd0>
[I 200307 06:22:20 config:127] Using builder: <kubeflow.fairing.builders.append.append.AppendBuilder object at 0x7fc89c5c2c88>
[I 200307 06:22:20 config:129] Using deployer: <kubeflow.fairing.deployers.job.job.Job object at 0x7fc8a9d78898>
[W 200307 06:22:20 append:50] Building image using Append builder...
[I 200307 06:22:20 base:107] Creating docker context: /tmp/fairing_context_n04p7lim
[I 200307 06:22:20 converted_notebook:127] Converting test.ipynb to test.py
[I 200307 06:22:20 docker_creds_:234] Loading Docker credentials for repository 'tensorflow/tensorflow:2.1.0-py3'
[W 200307 06:22:23 append:54] Image successfully built in 2.5187153209990356s.
[W 200307 06:22:23 append:94] Pushing image kangwoo/mnist-simple:5D22B676...
...
[W 200307 06:22:28 append:99] Pushed image kangwoo/mnist-simple:5D22B676 in 5.467247595999652s.
[W 200307 06:22:28 job:90] The job mnist-job-b445qfblz launched.

“function” 전처리기를 사용해서 모델 학습하기

주피터 노트북 사용자 인터페이스의 메뉴에서 File > New > Notebook 을 클릭하여 노트북 환경에서 새 노트북을 시작하십시오.

노트북 컬럼에 모델 코드와 faring 코드를 입력하겠습니다.

다음은 mnist 숫자를 분류하는 모델을 텐서플로우 케라스로 작성한 코드입니다.

import os

import tensorflow as tf
import numpy as np

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)

    print("Average test loss: ", np.average(training_history.history['loss']))

다음은 fairing 코드 입니다.

def fairing_fn(fn):
    CONTAINER_REGISTRY = 'kangwoo'

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


    fairing.config.set_builder('append', registry=CONTAINER_REGISTRY, image_name="mnist-simple",
                               base_image="tensorflow/tensorflow:2.1.0-py3")

    fairing.config.set_deployer('job', namespace=namespace, job_name=job_name, cleanup=False, stream_log=True)

    return fairing.config.fn(fn)

전처리기를 설정하는 set_preprocessor() 구문이 없습니다. 그리고 마지막 줄에 fairing을 실행하는 fairing.config.run() 코드가 사라지고, return fairing.config.fn(fn) 코드가 추가되었습니다. fairing.config.fn(fn) 함수는 넘겨진 fn 함수를 직접 실행해주는 “function” 전처리기가 사용됩니다.

다음은 fairing에서 함수를 넘겨받아 실행하는 부분입니다.

if __name__ == '__main__':
    import uuid
    from kubeflow import fairing
    remote_train = fairing_fn(train)
    remote_train()

shift + enter 키를 눌러 셀 들을 실행시키면, 다음과 같은 응답 결과를 얻을 수 있습니다.

[I 200307 07:08:41 config:125] Using preprocessor: <kubeflow.fairing.preprocessors.function.FunctionPreProcessor object at 0x7fb1800bffd0>
[I 200307 07:08:41 config:127] Using builder: <kubeflow.fairing.builders.append.append.AppendBuilder object at 0x7fb1800bf320>
[I 200307 07:08:41 config:129] Using deployer: <kubeflow.fairing.deployers.job.job.Job object at 0x7fb1800bfa20>
[W 200307 07:08:41 append:50] Building image using Append builder...
[I 200307 07:08:41 base:107] Creating docker context: /tmp/fairing_context_c1g8ic6t
[W 200307 07:08:41 base:94] /usr/local/lib/python3.6/dist-packages/kubeflow/fairing/__init__.py already exists in Fairing context, skipping...
[I 200307 07:08:41 docker_creds_:234] Loading Docker credentials for repository 'tensorflow/tensorflow:2.1.0-py3'
[W 200307 07:08:44 append:54] Image successfully built in 2.49341115399875s.
[W 200307 07:08:44 append:94] Pushing image kangwoo/mnist-simple:88C9B5FC...
...
[W 200307 07:08:50 append:99] Pushed image kangwoo/mnist-simple:88C9B5FC in 5.64489735999814s.
[W 200307 07:08:50 job:90] The job mnist-job-c27ep7bs5 launched.

“full_notebook” 전처리기를 사용해서 노트북 실행하기

“full_notebook” 전처리기는 Papermill 을 사용해서 노트북을 실행시킵니다. 그래서 컨테이너 이미지 안에 Papermill 이 있어야합니다. Papermill은 노트북을 매개 변수화하고 실행할 수있는 도구입니다.

Dockerfile

FROM tensorflow/tensorflow:2.1.0-py3

RUN python -m pip install papermill \\
    && python -m pip install papermill s3 \\
    && python -m pip install papermill gcloud \\
    && python -m pip install papermill azure \\
    && python -m pip install ipykernel \\
    && python -m ipykernel install

주피터 노트북 사용자 인터페이스의 메뉴에서 File > New > Notebook 을 클릭하여 노트북 환경에서 새 노트북을 시작하십시오.

노트북 컬럼에 모델 코드와 faring 코드를 입력하겠습니다.

다음은 mnist 숫자를 분류하는 모델을 텐서플로우 케라스로 작성한 코드입니다.

import os

import tensorflow as tf
import numpy as np

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)

    print("Average test loss: ", np.average(training_history.history['loss']))

다음은 fairing 코드 입니다.

def fairing_run():
    CONTAINER_REGISTRY = 'kangwoo'

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

    fairing.config.set_preprocessor('full_notebook')

    fairing.config.set_builder('append', registry=CONTAINER_REGISTRY, image_name="mnist-simple",
                               base_image="kangwoo/tensorflow:2.1.0-papermill")

    fairing.config.set_deployer('job', namespace=namespace, job_name=job_name, cleanup=False, stream_log=True)

    fairing.config.run()

“full_notebook” 전처리기를 사용합니다. “full_notebook” 전처리기는 Papermill 을 사용해서 노트북을 실행시킵니다. 그래서 컨테이너 이미지 안에 Papermill 이 있어야합니다. Papermill은 노트북을 매개 변수화하고 실행할 수있는 도구입니다.

다음은 fairing에서 함수를 넘겨받아 실행하는 부분입니다.

if __name__ == '__main__':
    if os.getenv('FAIRING_RUNTIME', None) is None:
        import uuid
        from kubeflow import fairing
        fairing_run()
    else:
        train()

shift + enter 키를 눌러 셀 들을 실행시키면, 다음과 같은 응답 결과를 얻을 수 있습니다.

[I 200307 08:32:18 config:125] Using preprocessor: <kubeflow.fairing.preprocessors.full_notebook.FullNotebookPreProcessor object at 0x7fefb8236a58>
[I 200307 08:32:18 config:127] Using builder: <kubeflow.fairing.builders.append.append.AppendBuilder object at 0x7fef3e50f898>
[I 200307 08:32:18 config:129] Using deployer: <kubeflow.fairing.deployers.job.job.Job object at 0x7fef3e7aba58>
[W 200307 08:32:18 append:50] Building image using Append builder...
[I 200307 08:32:18 base:107] Creating docker context: /tmp/fairing_context_d3874211
[I 200307 08:32:18 docker_creds_:234] Loading Docker credentials for repository 'kangwoo/tensorflow:2.1.0-papermill'
[W 200307 08:32:20 append:54] Image successfully built in 2.5086942100024316s.
[W 200307 08:32:20 append:94] Pushing image kangwoo/mnist-simple:51B47A33...
...
[W 200307 08:32:25 append:99] Pushed image kangwoo/mnist-simple:51B47A33 in 4.993432313000085s.
[W 200307 08:32:25 job:90] The job mnist-job-ca13vb4zw launched.

작업 결과 만들어진 노트북은 전처리기의 output_file 파라메터에 지정한 곳에 생성됩니다. 값을 지정하지 않으면, 기본값인 fairing_output_notebook.ipynb 으로 설정됩니다. 그래서 컨테이너 안에 생성되서, 쉽게 확인해 볼 수 없습니다. PV나 S3, GCS 같은 것을 사용해야합니다.

“cluster” 빌더를 사용해서 모델 학습하기

cluster 빌더를 사용해 보겠습니다. cluster 빌더는 컨테이너 이미지 빌드 작업이 쿠버네티스 클러스터에서 실행됩니다. cluster 빌더는 kaniko 라는 것을 사용해서 클러스터에서 빌드 작업을 합니다. kaniko는 컨테이너 또는 쿠버네티스 클러스터 내부에서 Dockerfile을 이용한 컨테이너 이미지를 빌드하는 도구입니다. kaniko는 도커 데몬을 사용하지 않으며, 사용자 공간에서 Dockerfile 에 포함된 명령을 완벽히 실행합니다. 그래서 쿠버네티스 클러스터와 같은 도커 데몬을 쉽고 안전하게 실행할 수 없는 환경에서도 컨테이너 이미지를 빌드 할 수 있습니다.

cluster 빌더를 사용하려면 ContextSource를 정의해 주어야 합니다. s3, gcs, azurestorage 그리고 minio를 지원하고 있습니다. (minio는 kubeflow-fairing 0.7.1.1 이상에서 가능합니다.)

docker-config 생성하기

먼저 kaniko에서 컨테이너 이미지 레지스터리에 접근할 수 있도록 쿠버네티스 클러스터에 docker-config 라는 컨피그맵(ConfigMap) 리소스를 만들겠습니다. 설정 정보는 앞에서 생성한 ~/.docker/config.json 파일을 사용하 겠습니다.

Jupyter 노트북 사용자 인터페이스의 메뉴에서 File > New > Terminal 을 클릭하여 노트북 환경에서 새 터미널 세션을 시작하십시오.

다음 명령어를 실행해서 admin이라는 네임스페이스에 docker-config 컨피그맵을 생성합니다.

kubectl -n admin create configmap docker-config --from-file=/home/jovyan/.docker/config.json

모델 파일 생성하기

모델 코드인 mnist-simple.py 파일과 컨테이너 이미지 빌드를 위한 Dockerfile을 생성합니다.

다음은 mnist 숫자를 분류하는 모델을 텐서플로우 케라스로 작성한 코드입니다.

mnist-simple.py

from __future__ import absolute_import, division, print_function, unicode_literals

import tensorflow as tf
import numpy as np


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)

    print("Average test loss: ", np.average(training_history.history['loss']))


if __name__ == '__main__':
    train()

컨테이너 이미지 빌드를 Dockerfile을 생성하겠습니다.

Dockerfile

FROM tensorflow/tensorflow:2.1.0-py3

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

주피터 노트북 사용자 인터페이스의 메뉴에서 File > New > Notebook 을 클릭하여 노트북 환경에서 새 노트북을 시작하십시오.

노트북 컬럼에 faring 코드를 입력하겠습니다.

컨테이너 이미지를 빌드하고, 배포하기 위한 fairing 코드 파일을 생성하겠습니다. cloud 빌더를 사용하고, ContextSource로 minio를 사용하였습니다.

Kubeflow 설치시 파이프라인에서 사용하기 위해서 minio가 기본적으로 설치됩니다. 편의를 위해서 해당 minio를 사용하도록 하겠습니다.

minio의 접속 정보를 사용하여 MinioContextSource 를 생성합니다. 그리고 빌더 타입을 cluster로 사용하고, context_source 파라미티를 사용하여 생성한 MinioContextSource를 넘겨줍니다.

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"

from kubeflow.fairing.builders.cluster.minio_context import MinioContextSource
minio_context_source = MinioContextSource(endpoint_url=minio_endpoint, minio_secret=minio_username, minio_secret_key=minio_key, region_name=minio_region)

fairing.config.set_builder('cluster', registry=CONTAINER_REGISTRY, image_name="mnist-simple", dockerfile_path="Dockerfile",
                           context_source=minio_context_source)

fairing-cloud-minio.py

import uuid
from kubeflow import fairing

CONTAINER_REGISTRY = 'kangwoo'

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

command=["python", "mnist-simple.py"]
output_map = {
    "Dockerfile": "Dockerfile",
    "mnist-simple.py": "mnist-simple.py"
}

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

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"

from kubeflow.fairing.builders.cluster.minio_context import MinioContextSource

minio_context_source = MinioContextSource(endpoint_url=minio_endpoint, minio_secret=minio_username, minio_secret_key=minio_key, region_name=minio_region)

fairing.config.set_builder('cluster', registry=CONTAINER_REGISTRY, image_name="mnist-simple", dockerfile_path="Dockerfile",
                           context_source=minio_context_source)


fairing.config.set_deployer('job', namespace=namespace, job_name=job_name, cleanup=False, stream_log=True)

fairing.config.run()

이제 작성한 fairing 코드를 실행해 보겠습니다.

python fairing-local-append.py

정상적으로 실행되면 다음과 같은 응답 화면을 볼 수 있습니다.

[I 200308 01:42:47 config:125] Using preprocessor: <kubeflow.fairing.preprocessors.base.BasePreProcessor object at 0x7f346dd42080>
[I 200308 01:42:47 config:127] Using builder: <kubeflow.fairing.builders.cluster.cluster.ClusterBuilder object at 0x7f34a0542320>
[I 200308 01:42:47 config:129] Using deployer: <kubeflow.fairing.deployers.job.job.Job object at 0x7f3475df0f28>
[I 200308 01:42:47 cluster:46] Building image using cluster builder.
[I 200308 01:42:47 base:107] Creating docker context: /tmp/fairing_context_2ns16rsm
[W 200308 01:42:48 manager:296] Waiting for fairing-builder-9hvzb-8v7x9 to start...
[I 200308 01:42:50 manager:302] Pod started running True
INFO[0002] Resolved base name tensorflow/tensorflow:2.1.0-py3 to tensorflow/tensorflow:2.1.0-py3
...
INFO[0045] Taking snapshot of files...
[W 200308 01:43:40 job:101] The job mnist-job-508a launched.
[W 200308 01:43:40 manager:296] Waiting for mnist-job-508a-ps9tt to start...
[I 200308 01:43:46 manager:302] Pod started running True

Job 이름은 “mnist-job-508a” 라는 것을 확인 할 수 있습니다.

다음 명령어를 실행하면 admin 네임스페이스의 Job 목록을 확인할 수 있습니다.

kubectl -n admin get job

정상적으로 실행되면 다음과 같은 응답 화면을 볼 수 있습니다.

NAME                  COMPLETIONS   DURATION   AGE
mnist-job-508a        1/1           18s        7m25s

다음 명령어를 실행하면 mnist-job-508a 라는 이름을 가진 Job이 생성한 포드를 확인할 수 있습니다.

kubectl -n admin get pod -l job-name=mnist-job-508a

정상적으로 실행되면 다음과 같은 응답 화면을 볼 수 있습니다.

NAME                        READY   STATUS      RESTARTS   AGE
mnist-job-508a-ps9tt        0/1     Completed   0          93s

학습 작업 삭제하기

작업이 완료되어도 Job은 삭제되지 않습니다.

다음 명령어를 실행하면 admin 네임스페이스의 mnist-job-508 라는 이름의 Job을 삭제할 수 있습니다.

kubectl -n admin delete job mnist-job-508

“tfjob” 배포자를 사용해서 모델 학습하기

“tfjob” 배포자와 “notebook” 전처리기 그리고 “cluster”배포자를 사용해서 모델을 학습해 보겠습니다. 모델을 저장하기 위해서 퍼시스턴스 볼륨을 사용하겠습니다

PVC 생성하기

학습이 완료된 모델을 저장하기 위해서 별도의 볼륨을 만들어서 사용하겠습니다.

다음은 100메가의 저장 용량을 가진 볼륨 생성을 요청하는 PVC 매니페스트입니다.

fairing-tfjob-pvc.yaml

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

kubectl을 사용해서 admin 네임스페이스에 pvc를 생성합니다.

kubectl -n admin apply -f fairing-tfjob-pvc.yaml

노트북 생성하기

주피터 노트북 사용자 인터페이스의 메뉴에서 File > New > Notebook 을 클릭하여 노트북 환경에서 새 노트북을 시작하십시오.

노트북 파일 이름을 mnist-dist.ipynb 으로 설정합니다.

노트북 컬럼에 모델 코드와 faring 코드를 입력하겠습니다.

다음은 mnist 숫자를 분류하는 모델을 텐서플로우 케라스로 작성한 코드입니다.

from __future__ import absolute_import, division, print_function, unicode_literals

import os
import tensorflow as tf
import tensorflow_datasets as tfds

def build_and_compile_model():
    model = tf.keras.models.Sequential([
        tf.keras.layers.Flatten(input_shape=(28, 28, 1)),
        tf.keras.layers.Dense(128, activation='relu'),
        tf.keras.layers.Dropout(0.2),
        tf.keras.layers.Dense(10)
    ])
    model.compile(loss=tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True),
                  optimizer=tf.keras.optimizers.Adam(),
                  metrics=['accuracy'])
    return model


@tfds.decode.make_decoder(output_dtype=tf.float32)
def decode_image(example, feature):
    return tf.cast(feature.decode_example(example), dtype=tf.float32) / 255


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

    BATCH_SIZE = 64

    tb_dir = '/app/data/logs'
    model_dir = '/app/data/export'
    version = 2
    export_dir = os.path.join(model_dir, str(version))

    strategy = tf.distribute.experimental.MultiWorkerMirroredStrategy()
    mnist = tfds.builder('mnist', data_dir='/app/mnist')
    mnist.download_and_prepare()

    mnist_train, mnist_test = mnist.as_dataset(
        split=['train', 'test'],
        decoders={'image': decode_image()},
        as_supervised=True)
    train_input_dataset = mnist_train.cache().repeat().shuffle(
        buffer_size=50000).batch(BATCH_SIZE)
    # eval_input_dataset = mnist_test.cache().repeat().batch(BATCH_SIZE)

    options = tf.data.Options()
    options.experimental_distribute.auto_shard_policy = tf.data.experimental.AutoShardPolicy.OFF
    train_input_dataset = train_input_dataset.with_options(options)

    print("Training...")

    with strategy.scope():
        multi_worker_model = build_and_compile_model()

    num_train_examples = mnist.info.splits['train'].num_examples
    train_steps = num_train_examples // BATCH_SIZE
    train_epochs = 10

    callbacks = [
        tf.keras.callbacks.TensorBoard(log_dir=tb_dir),
    ]

    history = multi_worker_model.fit(train_input_dataset, epochs=train_epochs, steps_per_epoch=train_steps,
                                     callbacks=callbacks)

    print("\\ntraining_history:", history.history)

    multi_worker_model.save(export_dir)

다음은 fairing 코드 입니다.

def fairing_run():
    import uuid
    from kubeflow import fairing
    from kubeflow.fairing.kubernetes import utils as k8s_utils
    from kubeflow.fairing.builders.cluster.minio_context import MinioContextSource

        
    CONTAINER_REGISTRY = 'kangwoo'

    namespace = 'admin'
    job_name = f'mnist-tfjob-{uuid.uuid4().hex[:4]}'

    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"

    minio_context_source = MinioContextSource(endpoint_url=minio_endpoint, minio_secret=minio_username, minio_secret_key=minio_key, region_name=minio_region)
    fairing.config.set_builder('cluster', registry=CONTAINER_REGISTRY, image_name="mnist-dist", dockerfile_path="Dockerfile",
                           context_source=minio_context_source)

    fairing.config.set_deployer('tfjob', namespace=namespace, job_name=job_name, cleanup=False, stream_log=True, 
                            worker_count=2,
                            pod_spec_mutators=[k8s_utils.mounting_pvc(pvc_name='fairing-tfjob-data-pvc', pvc_mount_path='/mnt/pv')])

    fairing.config.run()

전처리기를 설정하는 set_preprocessor() 구문이 없습니다. 노트북에서는 별도의 전처리기를 설정하지 않으면 “notebook” 전처리기가 사용됩니다. “notebook” 전처리기는 노트북의 파이썬 코드 셀을 파이썬 파일로 변경해 줍니다.

tfjob 배포자를 사용하였습니다. 워커를 2개 실행시키 위해서 worker_count=2를 설정하였습니다. 그리고, 모델을 저장하기 위한 퍼시스턴스 볼륨을 마운트하기 위해서 pod_spec_mutators=[k8s_utils.mounting_pvc(pvc_name=’fairing-tfjob-data-pvc’, pvc_mount_path=’/mnt/pv’)]) 을 설정하였습니다.

    fairing.config.set_deployer('tfjob', namespace=namespace, job_name=job_name, cleanup=False, stream_log=True, 
                            worker_count=2,
                            pod_spec_mutators=[k8s_utils.mounting_pvc(pvc_name='fairing-tfjob-data-pvc', pvc_mount_path='/mnt/pv')])

컨테이너 이미지 만들기

모델을 학습할 때 데이터를 가져오기 위해서 tensorflow-dataset을 사용합니다. 그래서 컨테이너 이미지에 tensorflow-dataset 패키지가 포함되어야 합니다. 노트북 환경에서 별도의 전처리기를 설정하지 않으면, 노트북_이름.ipynb 파일이 /app/노트북_이름.py 파일로 생성됩니다. 예제에서 사용한 노트북 파일 이름이 mnist-dist.ipynb 이기 때문에 “ADD /app/mnist-dist.py /app/”를 사용해서, 모델 파일을 컨테이너 이미지에 추가해줍니다.

Dockerfile

FROM tensorflow/tensorflow:2.1.0-py3

RUN pip install tensorflow-datasets==2.0.0

RUN mkdir -p /app
RUN ls -al /kaniko/buildcontext/app
ADD /app/mnist-dist.py /app/

다음은 환경 변수에 따라 모델 훈련을 실행하거나, fairing을 실행하는 부분입니다.

if __name__ == '__main__':
    if os.getenv('FAIRING_RUNTIME', None) is None:
        fairing_run()
    else:
        train()

shift + enter 키를 눌러 셀 들을 실행시키면, 다음과 같은 응답 결과를 얻을 수 있습니다.

[I 200308 03:06:38 config:125] Using preprocessor: <kubeflow.fairing.preprocessors.converted_notebook.ConvertNotebookPreprocessor object at 0x7f274c3a56a0>
[I 200308 03:06:38 config:127] Using builder: <kubeflow.fairing.builders.cluster.cluster.ClusterBuilder object at 0x7f26d167dc88>
[I 200308 03:06:38 config:129] Using deployer: <kubeflow.fairing.deployers.tfjob.tfjob.TfJob object at 0x7f274d7284a8>
[I 200308 03:06:38 cluster:46] Building image using cluster builder.
[I 200308 03:06:38 base:107] Creating docker context: /tmp/fairing_context_z45ivvfn
[I 200308 03:06:39 converted_notebook:127] Converting mnist-dist.ipynb to mnist-dist.py
[W 200308 03:06:39 manager:296] Waiting for fairing-builder-zp7xj-m6ctb to start...
[I 200308 03:06:41 manager:302] Pod started running True
[W 200308 03:08:07 job:101] The tfjob mnist-tfjob-d20e launched.
작성일자

Kubeflow – TensorFlow 학습하기

텐서플로우 알아 보기

텐서플로우(TensorFlow)는 구글에서 만든 오픈소스 딥러닝 프레임워크입니다. 딥러닝 프로그램을 쉽게 구현할 수 있도록 다양한 기능을 제공해주고 있습니다. 텐서플로우 자체는 기본적으로 C++로 구현 되어 있으며, Python, Java, Go 등 다양한 언어를 지원합니다.

TFJob을 이용해서 텐서플로우로 모델을 학습 할 수 있습니다. TFJob은 쿠버네티스에서 분산 또는 비 분산 텐서플로우 작업을 쉽게 실행할 수 있는 쿠버네티스 사용자 리소스(Custom Resource) 입니다.

TFJob

TFJob은 쿠버네티스에서 텐서플로우를 이용한 학습 작업을 할 수 있게 해주는 쿠버네티스 사용자 리소스 입니다. TFJob의 구현은 tf-operator에 있습니다. tf-operator는 TFJob을 관리합니다. 쿠버네티스에 TFJob이 등록되면, 필요한 포드들을 생성하여 작업을 실행할 수 있도록 도와줍니다.

TFJob은 다음과 같이 YAML 형식으로 표현할 수 있는 쿠버네티스 사용자 리소스입니다.

apiVersion: kubeflow.org/v1
kind: TFJob
metadata:
  name: mnist-tfjob-dist
  namespace: admin
spec:
  cleanPodPolicy: Running
  tfReplicaSpecs:
    Chief:
      replicas: 1
      template:
        metadata:
          annotations:
            sidecar.istio.io/inject: "false"
          name: tensorflow
        spec:
          containers:
          - command:
            - python
            - mnist-dist.py
            image: kangwoo/tfjob-dist:0.0.1
            name: tensorflow
            volumeMounts:
            - mountPath: /app/data
              name: tfjob-data-volume
            workingDir: /app
          restartPolicy: Never
          volumes:
          - name: tfjob-data-volume
            persistentVolumeClaim:
              claimName: tfjob-data-volume
    Worker:
      replicas: 2
      template:
        metadata:
          annotations:
            sidecar.istio.io/inject: "false"
          name: tensorflow
        spec:
          containers:
          - command:
            - python
            - mnist-dist.py
            image: kangwoo/tfjob-dist:0.0.1
            name: tensorflow
            volumeMounts:
            - mountPath: /app/data
              name: tfjob-data-volume
            workingDir: /app
          restartPolicy: Never
          volumes:
          - name: tfjob-data-volume
            persistentVolumeClaim:
              claimName: tfjob-data-volume

분산 처리 기술을 사용한 모델 학습

쿠버네티스에서 TFJob을 관리하고 있는 tf-operator 는 텐서플로우를 사용한 모델 학습시 분산 처리가 가능하도록 도와주고 있습니다.

텐서플로우에는 분산 처리 기술을 사용하기 위해서 다음과 같은 역할을 가진 프로세스가 존재합니다.

  • Chief: Chief는 학습을 조정하는 역할을 합니다. 그리고 모델의 체크포인트 같은 작업을 수행할 책임이 있습니다.
  • Ps (Paramter Server) : Ps는 파라미터 서버로서, 모델 파라미터에 대한 분산 데이터 저장소를 제공하는 역할을 합니다.
  • Worker : Worker는 실제 모델 학습 작업을 수행하는 역할을 합니다. 경우에 따라서 0번 Worker가 Chief 역할을 수행 할 수 있습니다.
  • Evaluator : Evaluator는 평가 지표를 계산하는 역할을 합니다. 모델을 학습 할 때 Evaluator를 사용하여 평가 지표를 계산할 수 있습니다.

TFJob은 TF_CONFIG라는 환경 변수를 사용하여 프로세스의 역할 정의합니다.

모델 학습을 시작하려면 TF_CONFIG라는 환경 변수에 다음과 같은 변수가 json 형태로 설정되어 있어야 합니다.

{
	"cluster": {
		"worker": ["host1:port", "host2:port", "host3:port"],
	  "ps": ["host4:port", "host5:port"]
   },
	"task": {"type": "worker", "index": 0}
}
  • cluster : 클러스터 내에 있는 서버들의 정보를 지정합니다.
  • task : 클러스터 내에서 현재 작업이 담당한 역할을 지정합니다. 클러스터 내에서 가질 수 있는 역할은 “chief”, “worker”, “ps”, “evaluator” 중 하나입니다. 단, “ps” 역할은 tf.distribute.experimental.ParameterServerStrategy 전략을 사용할 때만 쓸 수 있습니다

TFJob 이라는 쿠터네티스트 사용자 리소스를 생성하여 모델 학습을 진행 경우에는, tf-operator 가 프로세스의 역할에 맞게 자동으로 환경 변수를 설정해 줍니다. 그래서 분산 작업을 위한 별도의 환경 변수 설정 작업을 하지 않아도 됩니다.

다음은 TFJob에서 생성한 포드의 환경 변수를 일부 출력해 본 것입니다.

spec:
  containers:
  - env:
    - name: TF_CONFIG
      value: '{"cluster":{"chief":["mnist-tfjob-dist-chief-0.admin.svc:2222"],"worker":["mnist-tfjob-dist-worker-0.admin.svc:2222","mnist-tfjob-dist-worker-1.admin.svc:2222"]},"task":{"type":"chief","index":0},"environment":"cloud"}'

텐서플로우 학습 작업 실행하기

TFJob을 정의한 후 학습 작업을 생성해 보겠습니다.

모델 코드 작성하기

텐서플로우로 학습할 모델을 작성해 보겠습니다.

다음은 텐서플로우 케라스를 사용하여 mnist 숫자 이미지를 분류하는 파이썬 코드인 mnist-dist.py입니다. Gtf.distribute.experimental.MultiWorkerMirroredStrategy 를 사용해서 분산 학습을 지원하고 있습니다.

mnist-dist.py

from __future__ import absolute_import, division, print_function, unicode_literals

import os
import json
import tensorflow as tf
import tensorflow_datasets as tfds


def build_and_compile_model():
    model = tf.keras.models.Sequential([
        tf.keras.layers.Flatten(input_shape=(28, 28, 1)),
        tf.keras.layers.Dense(128, activation='relu'),
        tf.keras.layers.Dropout(0.2),
        tf.keras.layers.Dense(10)
    ])
    model.compile(loss=tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True),
                  optimizer=tf.keras.optimizers.Adam(),
                  metrics=['accuracy'])
    return model


@tfds.decode.make_decoder(output_dtype=tf.float32)
def decode_image(example, feature):
    return tf.cast(feature.decode_example(example), dtype=tf.float32) / 255


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

    tf_config = os.environ.get('TF_CONFIG', '{}')
    print("TF_CONFIG %s", tf_config)
    tf_config_json = json.loads(tf_config)
    cluster = tf_config_json.get('cluster')
    job_name = tf_config_json.get('task', {}).get('type')
    task_index = tf_config_json.get('task', {}).get('index')
    print("cluster={} job_name={} task_index={}}", cluster, job_name, task_index)

    BATCH_SIZE = 64

    tb_dir = '/app/data/logs'
    model_dir = '/app/data/export'
    version = 1
    export_dir = os.path.join(model_dir, str(version))

    strategy = tf.distribute.experimental.MultiWorkerMirroredStrategy()
    mnist = tfds.builder('mnist', data_dir='/app/mnist')
    mnist.download_and_prepare()

    mnist_train, mnist_test = mnist.as_dataset(
        split=['train', 'test'],
        decoders={'image': decode_image()},
        as_supervised=True)
    train_input_dataset = mnist_train.cache().repeat().shuffle(
        buffer_size=50000).batch(BATCH_SIZE)
    # eval_input_dataset = mnist_test.cache().repeat().batch(BATCH_SIZE)

    options = tf.data.Options()
    options.experimental_distribute.auto_shard_policy = tf.data.experimental.AutoShardPolicy.OFF
    train_input_dataset = train_input_dataset.with_options(options)

    print("Training...")

    with strategy.scope():
        multi_worker_model = build_and_compile_model()

    num_train_examples = mnist.info.splits['train'].num_examples
    train_steps = num_train_examples // BATCH_SIZE
    train_epochs = 10

    callbacks = [
        tf.keras.callbacks.TensorBoard(log_dir=tb_dir),
    ]

    history = multi_worker_model.fit(train_input_dataset, epochs=train_epochs, steps_per_epoch=train_steps,
                                     callbacks=callbacks)

    print("\\ntraining_history:", history.history)

    multi_worker_model.save(export_dir)


if __name__ == '__main__':
    train()

모델 컨테이너 이미지 만들기

모델 학습용 컨테이너 이미지를 만들기 위해서 Dockerfile을 생성하겠습니다. 텐서플로우 데이터셋을 사용하기 때문에 pip를 이용해서 추가해주었습니다.

다음은 텐서플로우2.1을 기반 이미지로 해서, 모델 파일을 추가하는 Dockerfile 입니다.

Dockerfile

FROM tensorflow/tensorflow:2.1.0-py3

RUN pip install tensorflow-datasets==2.0.0

RUN mkdir -p /app
ADD mnist-dist.py /app/

다음 명령어를 실행하면 kkangwoo/tfjob-dist:0.0.1 라는 이름의 컨테이너 이미지를 빌드 할 수 있습니다.

docker build -t kangwoo/tfjob-dist:0.0.1 .

빌드한 컨테이너 이미지를 컨테이너 레지스트리로 업로드 하겠습니다.

docker push kangwoo/tfjob-dist:0.0.1

PVC 생성하기

텐서플로우 학습 작업 중 생성되는 매트릭과, 학습이 완료된 모델을 저장하기 위해서 별도의 볼륨을 만들어서 사용하겠습니다.

다음은 100메가의 저장 용량을 가진 볼륨 생성을 요청하는 PVC 매니페스트입니다.

tfjob-pvc.yaml

kind: PersistentVolumeClaim
apiVersion: v1
metadata:
  name: tfjob-data-volume
spec:
  accessModes:
    - ReadWriteOnce
  resources:
    requests:
      storage: 100Mi

kubectl을 사용해서 admin 네임스페이스에 pvc를 생성합니다.

kubectl -n admin apply -f tfjob-pvc.yaml

TFJob 생성하기

tf-operator를 사용해서 텐서플로우로 작성한 모델을 학습 하려면 TFJob을 정의해야합니다.

  1. tfReplicaSpecs를 설정합니다. “tfReplicaSpecs”은 텐서플로우의 분산 학습시 사용하는 프로세스들을 정의하는데 사용합니다. Chief를 1개로 설정하고, Worker를 2개로 설정하였습니다.. Kubeflow 클러스터에 istio가 설치되어 있기 때문에, 자동으로 istio-proxy가 포드에 주입됩니다. 이것을 방지하기 위해서 어노테이션에 sidecar.istio.io/inject: “false” 을 추가해 주었습니다. 그리고 앞에 생성한 볼륨을 마운트 하였습니다. 
tfReplicaSpecs:
    Chief:
      replicas: 1
      template:
        metadata:
          annotations:
            sidecar.istio.io/inject: "false"
          name: tensorflow
        spec:
          containers:
          - command:
            - python
            - mnist-dist.py
            image: kangwoo/tfjob-dist:0.0.1
            name: tensorflow
            volumeMounts:
            - mountPath: /app/data
              name: tfjob-data-volume
            workingDir: /app
          restartPolicy: Never
          volumes:
          - name: tfjob-data-volume
            persistentVolumeClaim:
              claimName: tfjob-data-volume
    Worker:
      replicas: 2
      template:
        metadata:
          annotations:
            sidecar.istio.io/inject: "false"
          name: tensorflow
        spec:
          containers:
          - command:
            - python
            - mnist-dist.py
            image: kangwoo/tfjob-dist:0.0.1
            name: tensorflow
            volumeMounts:
            - mountPath: /app/data
              name: tfjob-data-volume
            workingDir: /app
          restartPolicy: Never
          volumes:
          - name: tfjob-data-volume
            persistentVolumeClaim:
              claimName: tfjob-data-volume

다음은 TFJob 을 생성한 위한 메니페스트입니다.

tfjob-dist.yaml

apiVersion: kubeflow.org/v1
kind: TFJob
metadata:
  name: mnist-tfjob-dist
spec:
  tfReplicaSpecs:
    Chief:
      replicas: 1
      template:
        metadata:
          annotations:
            sidecar.istio.io/inject: "false"
          name: tensorflow
        spec:
          containers:
          - command:
            - python
            - mnist-dist.py
            image: kangwoo/tfjob-dist:0.0.1
            name: tensorflow
            volumeMounts:
            - mountPath: /app/data
              name: tfjob-data-volume
            workingDir: /app
          restartPolicy: Never
          volumes:
          - name: tfjob-data-volume
            persistentVolumeClaim:
              claimName: tfjob-data-volume
    Worker:
      replicas: 2
      template:
        metadata:
          annotations:
            sidecar.istio.io/inject: "false"
          name: tensorflow
        spec:
          containers:
          - command:
            - python
            - mnist-dist.py
            image: kangwoo/tfjob-dist:0.0.1
            name: tensorflow
            volumeMounts:
            - mountPath: /app/data
              name: tfjob-data-volume
            workingDir: /app
          restartPolicy: Never
          volumes:
          - name: tfjob-data-volume
            persistentVolumeClaim:
              claimName: tfjob-data-volume

다음 명령어를 실행하면 admin 네임스페이스에 tfjob-dist 이라는 이름의 TFJob을 생성할 수 있습니다.

kubectl -n admin apply -f tfjob-dist.yaml

텐서플로우 학습 작업 확인하기

생성한 TFJob은 다음 명령어를 실행해서 확인 해 볼 수 있습니다.

kubectl -n admin get tfjob

생성된 TFJob이 있다면, 다음과 같은 응답 결과를 얻을 수 있습니다.

NAME               STATE       AGE
mnist-tfjob-dist   Succeeded   19m

TFJob이 생성되면, tf-operator 에 의해서 포드들이 생성됩니다. TFJob 매니페스트에 정의한 개수대로 chief, worker 포드가 생성되게 됩니다.

생성된 포드들은 다음 명령어를 실행해서 확인 해 볼 수 있습니다.

kubectl -n admin get pod -l tf-job-name=mnist-tfjob-dist

생성된 포드들이 남아 있다면, 다음과 같은 응답 결과를 얻을 수 있습니다.

NAME                    READY   STATUS    RESTARTS   AGE
mnist-tfjob-dist-chief-0    0/1     Completed   0          20m
mnist-tfjob-dist-worker-0   0/1     Completed   0          20m
mnist-tfjob-dist-worker-1   0/1     Completed   0          20m

TFJob은 작업이 끝난 후, 관련 포드들을 삭제해버립니다. 그래서 작업이 완료되면 포드가 조회되지 않을 수 있습니다. 작업이 완료되어도 포드들을 남겨 두고 싶다면, TFJob 매니페스트의 spec 부분에 “cleanPodPolicy: None” 를 추가하시면 됩니다.

TFJob spec의 CleanPodPolicy는 작업이 종료 될 때 포드 삭제를 제어할 때 사용합니다. 다음 값들 중 하나를 사용할 수 있습니다.

  • Running : 작업이 완료되었을 때, 실행이 끝난(Completed) 포드들은 삭제하지 않고, 실행중인(Running) 포드들만 삭제합니다.
  • All : 작업이 완료되었을 때, 실행이 끝난 포드들을 즉시 삭제합니다.
  • None : 작업이 완료되어도 포드들을 삭제하지 않습니다.

다음은 cleanPodPolicy를 추가한 메니페스트 예제입니다.

apiVersion: kubeflow.org/v1
kind: TFJob
metadata:
  name: mnist-tfjob-dist
spec:
spec:
  cleanPodPolicy: None
...

TFJob의 작업 상태를 알고 싶으면 describe 명령어를 사용할 수 있습니다.

다음 명령어를 실행하면 admin 네임스페이스에 mnist-tfjob-dist 이라는 이름의 TFJob의 상태를 조회할 수 있습니다.

kubectl -n admin describe tfjob mnist-tfjob-dist

다음은 예제 작업에 대한 샘플 출력입니다.

Name:         mnist-tfjob-dist
Namespace:    admin
Labels:       <none>
Annotations:  ...
API Version:  kubeflow.org/v1
Kind:         TFJob
Metadata:
...
Spec:
...
Status:
  Completion Time:  2020-03-07T02:17:17Z
  Conditions:
    Last Transition Time:  2020-03-07T02:16:10Z
    Last Update Time:      2020-03-07T02:16:10Z
    Message:               TFJob mnist-tfjob-dist is created.
    Reason:                TFJobCreated
    Status:                True
    Type:                  Created
    Last Transition Time:  2020-03-07T02:16:12Z
    Last Update Time:      2020-03-07T02:16:12Z
    Message:               TFJob mnist-tfjob-dist is running.
    Reason:                TFJobRunning
    Status:                False
    Type:                  Running
    Last Transition Time:  2020-03-07T02:17:17Z
    Last Update Time:      2020-03-07T02:17:17Z
    Message:               TFJob mnist-tfjob-dist successfully completed.
    Reason:                TFJobSucceeded
    Status:                True
    Type:                  Succeeded
  Replica Statuses:
    Chief:
      Succeeded:  1
    Worker:
      Succeeded:  2
  Start Time:     2020-03-07T02:16:11Z
Events:        <none>

텐서플로우 학습 작업 삭제하기

작업이 완료되어도 TFJob은 삭제되지 않습니다.

다음 명령어를 실행하면 admin 네임스페이스에 mnist-tfjob-dist이라는 이름의 TFJob을 삭제할 수 있습니다.

kubectl -n admin delete tfjob mnist-tfjob-dist

텐서보드

TensorBoard는 텐서플로우의 실행 및 그래프를 검사하고, 이해하기 위한 웹 응용 도구로서, 텐서플로우에서 기록한 로그를 그래프로 시각화하여 보여줍니다. 텐서보드는 텐서플로우에 포함되어 있습니다.

텐서보드를 실행하기 위해서 쿠버네티스 디플로이먼트 리소스를 정의해 보겠습니다.

“tensorflow/tensorflow:2.1.0-py3” 라는 컨테이너 이미지를 사용하여 텐서보드를 실행하겠습니다. 텐서보드는는 실행 될 때, 주어진 파리미터 값에 의해 로그를 읽어옵니다.

다음은 텐서보드에서 사용하는 몇 가지 파라미터입니다.

  • port : 텐서보드에 접근하기 위한 포트입니다.
  • logdir : 로그가 저장된 디렉토리 경로입니다.
  • path_prefix : 접속 URL에서 사용하는 경로 접두사입니다. 예를 들어 /foo/bar의 경로 접두사를 사용하면, http://localhost/이 아닌 http://localhost:6006/foo/bar/ 를 통해서 TensorBoard에 접속할 수 있습니다.
- command:
        - /usr/local/bin/tensorboard
        - --logdir=/app/data/logs
        - --path_prefix=/namespace/admin/tensorboard/mnist-dist/
        - --port=80

앞서 실행한 TFJob에서 텐서플로우 로그를 쿠버네티스의 퍼시스턴스 볼륨에 저장하였습니다. tfjob-data-volume 라는 퍼시스턴스 볼륨 클레임 이름을 가지고 있습니다. 해당 볼륨을 텐서보드에 마운트 해줍니다.

        volumeMounts:
        - mountPath: /app/data
          name: tfjob-data-volume
      volumes:
      - name: tfjob-data-volume
        persistentVolumeClaim:
          claimName: tfjob-data-volume

다음은 디플로이먼트 매니페스트입니다.

mnist-dist-tensorboard-deployment.yaml

apiVersion: apps/v1
kind: Deployment
metadata:
  labels:
    app: mnist-dist-tensorboard
  name: mnist-dist-tensorboard
  namespace: admin
spec:
  selector:
    matchLabels:
      app: mnist-dist-tensorboard
  template:
    metadata:
      labels:
        app: mnist-dist-tensorboard
    spec:
      serviceAccount: default-editor
      containers:
      - command:
        - /usr/local/bin/tensorboard
        - --logdir=/app/data/logs
        - --path_prefix=/namespace/admin/tensorboard/mnist-dist/
        - --port=80
        image: tensorflow/tensorflow:2.1.0-py3
        name: tensorboard
        ports:
        - containerPort: 80
        volumeMounts:
        - mountPath: /app/data
          name: tfjob-data-volume
      volumes:
      - name: tfjob-data-volume
        persistentVolumeClaim:
          claimName: tfjob-data-volume

kubectl을 사용해서 admin 네임스페이스에 디플로이먼트 리소스를 생성합니다.

kubectl -n admin apply -f mnist-dist-tensorboard-deployment.yaml

텐서보드에 접속하기 위해서 쿠버네티스 서비스를 리소스를 정의합니다.

mnist-dist-tensorboard-service.yaml

apiVersion: v1
kind: Service
metadata:
  labels:
    app: mnist-dist-tensorboard
  name: mnist-dist-tensorboard
  namespace: admin
spec:
  ports:
  - name: http
    port: 80
    targetPort: 80
  selector:
    app: mnist-dist-tensorboard
  type: ClusterIP

kubectl을 사용해서 admin 네임스페이스에 서비스 리소스를 생성합니다.

kubectl -n admin apply -f mnist-dist-tensorboard-service.yaml

Kubeflow 클러스터에는 istio가 설치되어 있었습니다. 텐서보드를 istio-ingressgateway를 통해서 접근할 수 있도록 하기위해서 VirtualService 라는 사용자 리소스를 정의하겠습니다.

mnist-dist-tensorboard-virtualservice.yaml

apiVersion: networking.istio.io/v1alpha3
kind: VirtualService
metadata:
  name: mnist-dist-tensorboard
  namespace: admin
spec:
  gateways:
  - kubeflow/kubeflow-gateway
  hosts:
  - '*'
  http:
  - match:
    - uri:
        prefix: /namespace/admin/tensorboard/mnist-dist/
    route:
    - destination:
        host: mnist-dist-tensorboard.admin.svc.cluster.local
    timeout: 300s

kubectl을 사용해서 admin 네임스페이스에 서비스 리소스를 생성합니다.

kubectl -n admin apply -f mnist-dist-tensorboard-virtualservice.yaml

다음 주소로 텐서보드에 접근할 수 있습니다.

http://kubeflow-address/namespace/admin/tensorboard/mnist-dist/

텐서플로우 서빙

Serving은 학습이 완료된 모델을 실제 서비스에 적용하는 것입니다. 텐서플로우 서빙을 이용하면 텐서플로우에서 생성한 모델을 서비스 환경에 쉽고 빠르게 적용할 수 있습니다.

텐서플로우 서빙을 이용해서 모델 서버 실행하기

TensorFlow 모델 서버를 실행하기 위해서 쿠버네티스 디플로이먼트 리소스를 정의해 보겠습니다.

“tensorflow/serving:2.1.0” 라는 컨테이너 이미지를 사용하여 모델 서버를 실행하겠습니다. 모델 서버는 실행 될 때, 주어진 파리미터 값에 의해 모델을 로드하게 됩니다. 모델 서버에 모델이 로드 되면, REST나 GRPC를 사용하여 추론 요청을 시작할 수 있습니다.

다음은 모델 서버에서 사용하는 몇 가지 파라미터입니다.

  • port : GRPC 요청에 사용할 포트입니다.
  • rest_api_port : REST 요청에 사용할 포트입니다.
  • modeL_name : REST 요청의 URL에서 이를 사용합니다.
  • mode_base_path : 모델을 저장 한 디렉토리 경로입니다. containers: – args: – –port=9000 – –rest_api_port=8500 – –model_name=mnist – –model_base_path=/app/data/export command: – /usr/bin/tensorflow_model_server image: tensorflow/serving:2.1.0

앞서 TFJob을 사용하여, 학습된 모델을 쿠버네티스의 퍼시스턴스 볼륨에 저장하였습니다. tfjob-data-volume 라는 퍼시스턴스 볼륨 클레임 이름을 가지고 있습니다. 해당 볼륨을 모델 서버에 마운트 해줍니다.

        volumeMounts:
        - mountPath: /app/data
          name: tfjob-data-volume
      volumes:
      - name: tfjob-data-volume
        persistentVolumeClaim:
          claimName: tfjob-data-volume

다음은 디플로이먼트 매니페스트입니다.

mnist-model-server-deployment.yaml

apiVersion: apps/v1
kind: Deployment
metadata:
  labels:
    app: mnist-model-server
  name: mnist-model-server
spec:
  selector:
    matchLabels:
      app: mnist-model-server
  template:
    metadata:
      annotations:
        sidecar.istio.io/inject: "false"
      labels:
        app: mnist-model-server
        version: v1
    spec:
      serviceAccount: default-editor
      containers:
      - args:
        - --port=9000
        - --rest_api_port=8500
        - --model_name=mnist
        - --model_base_path=/app/data/export
        command:
        - /usr/bin/tensorflow_model_server
        image: tensorflow/serving:2.1.0
        imagePullPolicy: IfNotPresent
        livenessProbe:
          initialDelaySeconds: 30
          periodSeconds: 30
          tcpSocket:
            port: 9000
        name: mnist
        ports:
        - containerPort: 9000
        - containerPort: 8500
        volumeMounts:
        - mountPath: /app/data
          name: tfjob-data-volume
      volumes:
      - name: tfjob-data-volume
        persistentVolumeClaim:
          claimName: tfjob-data-volume

kubectl을 사용해서 admin 네임스페이스에 디플로이먼트 리소스를 생성합니다.

kubectl -n admin apply -f mnist-model-server-deployment.yaml

모델 서버에서 정상적으로 모델을 로드 했는지 여부를 확인하려면, 포드의 로그를 조회하면 됩니다.

kubectl -n admin logs -l app=mnist-model-server

정상적으로 모델이 로드 되었다면, 다음과 같은 응답 결과를 학인 수 있습니다.

2020-03-07 00:58:07.813234: I tensorflow_serving/model_servers/server.cc:86] Building single TensorFlow model file config:  model_name: mnist model_base_path: /app/data/export
2020-03-07 00:58:07.814324: I tensorflow_serving/model_servers/server_core.cc:462] Adding/updating models.
2020-03-07 00:58:07.814334: I tensorflow_serving/model_servers/server_core.cc:573]  (Re-)adding model: mnist
2020-03-07 00:58:07.918328: I tensorflow_serving/core/basic_manager.cc:739] Successfully reserved resources to load servable {name: mnist version: 1}
2020-03-07 00:58:07.918351: I tensorflow_serving/core/loader_harness.cc:66] Approving load for servable version {name: mnist version: 1}
2020-03-07 00:58:07.918357: I tensorflow_serving/core/loader_harness.cc:74] Loading servable version {name: mnist version: 1}
2020-03-07 00:58:07.918373: I external/org_tensorflow/tensorflow/cc/saved_model/reader.cc:31] Reading SavedModel from: /app/data/export/1
2020-03-07 00:58:07.920422: I external/org_tensorflow/tensorflow/cc/saved_model/reader.cc:54] Reading meta graph with tags { serve }
2020-03-07 00:58:07.920439: I external/org_tensorflow/tensorflow/cc/saved_model/loader.cc:264] Reading SavedModel debug info (if present) from: /app/data/export/1
2020-03-07 00:58:07.945844: I external/org_tensorflow/tensorflow/cc/saved_model/loader.cc:203] Restoring SavedModel bundle.
2020-03-07 00:58:07.977914: I external/org_tensorflow/tensorflow/cc/saved_model/loader.cc:152] Running initialization op on SavedModel bundle at path: /app/data/export/1
2020-03-07 00:58:07.983624: I external/org_tensorflow/tensorflow/cc/saved_model/loader.cc:333] SavedModel load for tags { serve }; Status: success: OK. Took 65247 microseconds.
2020-03-07 00:58:07.984126: I tensorflow_serving/servables/tensorflow/saved_model_warmup.cc:105] No warmup data file found at /app/data/export/1/assets.extra/tf_serving_warmup_requests
2020-03-07 00:58:07.984397: I tensorflow_serving/core/loader_harness.cc:87] Successfully loaded servable version {name: mnist version: 1}
2020-03-07 00:58:07.995156: I tensorflow_serving/model_servers/server.cc:358] Running gRPC ModelServer at 0.0.0.0:9000 ...
2020-03-07 00:58:07.997145: I tensorflow_serving/model_servers/server.cc:378] Exporting HTTP/REST API at:localhost:8500 ...

모델 서버에 접속하기 위해서 쿠버네티스 서비스를 리소스를 정의합니다.

mnist-model-server-service.yaml

apiVersion: v1
kind: Service
metadata:
  labels:
    app: mnist-model-server
  name: mnist-model-server
spec:
  ports:
  - name: grpc-tf-serving
    port: 9000
    targetPort: 9000
  - name: http-tf-serving
    port: 8500
    targetPort: 8500
  selector:
    app: mnist-model-server
  type: ClusterIP

kubectl을 사용해서 admin 네임스페이스에 서비스 리소스를 생성합니다.

kubectl -n admin apply -f mnist-model-server-service.yaml

텐서플로우 서빙에서 로드된 모델에게 요청 하기

아래 예제 코드를 실행하기 위해서는 request와 tensorflow-datasets라는 파이썬 패키지가 필요합니다. 설치가 되어 있지 않다면 다음 명령어로 설치할 수 있습니다.

pip install request
pip install tensorflow-datasets==2.0.0

주피터 노트북 사용자 인터페이스의 메뉴에서 File > New > Notebook 을 클릭하여 노트북 환경에서 새 노트북을 생성하겠습니다.

먼저 요청에 사용할 테스트 데이터를 생성하는 코드를 작성합니다.

import os
import json
import numpy as np
import matplotlib.pyplot as plt
import subprocess
import tensorflow as tf
import tensorflow_datasets as tfds

dataset = tfds.load("mnist", shuffle_files=True, as_supervised=False)
train_dataset, test_dataset = dataset["train"], dataset["test"]
test_data = []
iterator = iter(test_dataset.batch(5))
test_data.append(next(iterator))

이미지와 내용을 화면에 출력하기 위해서 show() 함수를 작성합니다.

def show(idx, title):
  plt.figure()
  plt.imshow(test_images[idx].reshape(28,28))
  plt.axis('off')
  plt.title('\\n\\n{}'.format(title), fontdict={'size': 16})

test_data의 0번째에 있는 이미지와 라벨을 테스트 삼아 출력해 보겠습니다.

show(0, 'An Example Image: {}'.format(test_labels[0]))

모델 서버에 요청할 데이터를 json을 사용해서 작성합니다.

import json
data = json.dumps({"signature_name": "serving_default", "instances": test_images.tolist()})
print('Data: {} ... {}'.format(data[:50], data[len(data)-52:]))

모델 서버에 POST 형식으로 요청 데이터를 전송하고, 예측 결과를 받습니다.

import requests
headers = {"content-type": "application/json"}
json_response = requests.post('<http://mnist-model-server:8500/v1/models/mnist:predict>', data=data, headers=headers)
predictions = json.loads(json_response.text)['predictions']

예측 결과를 show() 함수를 사용하여 화면이 출력합니다.

for i in range(0, len(predictions)):
    show(i, 'The model thought this was number {} , and it was actually number {}'.format(
  np.argmax(predictions[i]), test_labels[i]))

정상적으로 처리 되었다면, 다음과 비슷한 응답 결과를 얻을 수 있습니다.

작성일자

Kubeflow – MXNet 학습하기

MXNet 알아 보기

MXNet은 카네기 멜론 대학과 워싱톤 대학이 시작한 오픈소스 딥러닝 프레임워크입니다. MXNet은 빠르고 확장 가능한 학습 및 추론 프레임워크로서 다양한 언어와 디바이스를 지원하고 있습니다.

MXJob을 이용해서 MXNet을 학습 할 수 있습니다. MXJob은 쿠버네티스에서 분산 또는 비 분산 MXNet 작업을 쉽게 실행할 수 있는 쿠버네티스 CR(Custom Resource) 입니다.

MXJob

MXJob은 Kubernetes에서 MXNet 학습 작업을 실행하는 데 사용할 수 있는, 쿠버네티스 사용자 리소스 입니다. MXJob의 구현은 mxnet-operator에 있습니다. mxnet-operator는 MXJob을 관리합니다. 쿠버네티스에 MXJob이 등록되면, 필요한 포드들을 생성하여 작업을 실행할 수 있도록 도와줍니다.

MXJob은 다음과 같이 YAML 형식으로 표현할 수 있는 쿠버네티스 사용자 리소스입니다.

apiVersion: "kubeflow.org/v1beta1"
kind: "MXJob"
metadata:
  name: "mxnet-job"
spec:
  jobMode: MXTrain
  mxReplicaSpecs:
    Scheduler:
      replicas: 1
      restartPolicy: Never
      template:
        metadata:
          annotations:
            sidecar.istio.io/inject: "false"
        spec:
          containers:
            - name: mxnet
              image: kangwoo/mxnet:cpu
    Server:
      replicas: 1
      restartPolicy: Never
      template:
        metadata:
          annotations:
            sidecar.istio.io/inject: "false"
        spec:
          containers:
            - name: mxnet
              image: kangwoo/mxnet:cpu
    Worker:
      replicas: 1
      restartPolicy: Never
      template:
        metadata:
          annotations:
            sidecar.istio.io/inject: "false"
        spec:
          containers:
            - name: mxnet
              image: kangwoo/mxnet:cpu

분산 처리 기술을 사용한 모델 학습

쿠버네티스에서 MXJob을 관리하고 있는 mxnet-operator 는 MXNet을 사용한 모델 학습시 분산 처리가 가능하도록 도와주고 있습니다.

MXNet에는 분산 처리 기술을 사용하기 위해서 세 가지 역할의 프로세스가 존재합니다.

  • Scheduler (스케줄러) : 스케줄러는 1대만 존재해야만 합니다. 스케줄러의 역할은 클러스터를 설정하는 것입니다. 모든 프로세스에 클러스터의 다른 노드들 알려서 서로 통신 할 수 있도록 도와주고 있습니다.
  • Server (서버) : 모델의 매개 변수를 저장하고 작업자와 통신하는 역할을 합니다. 서버는 1대 이상 있을 수 있습니다.
  • Worker (작업자) : 작업자 노드는 실제 모델 학습 작업을 수행하는 역할을 합니다. 각 배치를 처리하기 전에 서버에서 가중치를 가지고 오고, 배치 작업이 끝난 후 서버에 가중치를 그라디언트를 보냅니다. 작업자는 1대 이상 있을 수 있습니다.

MXNet은 환경 변수를 사용하여 프로세스의 역할 정의하고, 다른 프로세스가 스케줄러를 찾도록 합니다.

모델 학습을 시작하려면 다음과 같은 환경 변수가 올바르게 설정되어 있어야 합니다.

  • DMLC_ROLE : 프로세스의 역할을 설정합니다. 서버, 작업자 또는 스케줄러 일 수 있습니다. 스케줄러는 하나만 있어야 합니다.
  • DMLC_PS_ROOT_URI : 스케줄러의 IP를 설정합니다.
  • DMLC_PS_ROOT_PORT : 스케줄러의 포트를 설정합니다.
  • DMLC_NUM_SERVER : 클러스터에있는 서버 노드 수를 설정합니다.
  • DMLC_NUM_WORKER : 클러스터에있는 작업자 노드 수를 설정합니다.

MXJob 이라는 쿠터네티스트 CR을 생성하여 모델 학습을 진행 경우에는, mxnet-operator 가 프로세스의 역할에 맞게 자동으로 환경 변수를 설정해 줍니다. 그래서 분산 작업을 위한 별도의 환경 변수 설정 작업을 하지 않아도 됩니다.

다음은 MXJob에서 생성한 포드의 환경 변수를 일부 출력해 본 것입니다.

spec:
  containers:
  - env:
    - name: MX_CONFIG
      value: '{"cluster":{"scheduler":[{"url":"mxnet-job-scheduler-0","port":9091}],"server":[{"url":"mxnet-job-server-0","port":9091}],"worker":[{"url":"mxnet-job-worker-0","port":9091}]},"labels":{"scheduler":"","server":"","worker":""},"task":{"type":"server","index":0}}'
    - name: DMLC_PS_ROOT_PORT
      value: "9091"
    - name: DMLC_PS_ROOT_URI
      value: mxnet-job-scheduler-0
    - name: DMLC_NUM_SERVER
      value: "1"
    - name: DMLC_NUM_WORKER
      value: "1"
    - name: DMLC_ROLE
      value: server
    - name: DMLC_USE_KUBERNETES
      value: "1"

MXJob CRD와 Operator 설치하기

MXJob을 사용하려면 쿠버네티스에 CRD와 Operator가 설치되어 있어야 합니다.

쿠버네티스가 MXJob을 지원하고 있는지 확인하기

다음 명령어를 실행하면 쿠버네티스 클러스터에 MXNet CRD가 설치되어 있는지 확인할 수 있습니다.

kubectl get crd | grep mxjobs

CRD가 설치되어 있다면, 다음과 같은 응답 결과를 얻을 수 있습니다.

mxjobs.kubeflow.org                                  2020-03-01T08:41:16Z

만약 설치되어 있지 않은 경우에는, 다음과 같은 명령어로 설치할 수 있습니다.

git clone <https://github.com/kubeflow/manifests>
cd manifests/mxnet-job/mxnet-operator
kubectl apply -k base

참고 : 다음은 MXJob CRD와 mxnet-operator 생성이 필요한 전체 매니페스트 내용입니다.

apiVersion: apiextensions.k8s.io/v1beta1
kind: CustomResourceDefinition
metadata:
  labels:
    kustomize.component: mxnet-operator
  name: mxjobs.kubeflow.org
spec:
  group: kubeflow.org
  names:
    kind: MXJob
    plural: mxjobs
    singular: mxjob
  scope: Namespaced
  version: v1beta1
---
apiVersion: v1
kind: ServiceAccount
metadata:
  labels:
    app: mxnet-operator
    kustomize.component: mxnet-operator
  name: mxnet-operator
  namespace: kubeflow
---
apiVersion: rbac.authorization.k8s.io/v1beta1
kind: ClusterRole
metadata:
  labels:
    app: mxnet-operator
    kustomize.component: mxnet-operator
  name: mxnet-operator
rules:
- apiGroups:
  - kubeflow.org
  resources:
  - mxjobs
  verbs:
  - '*'
- apiGroups:
  - apiextensions.k8s.io
  resources:
  - customresourcedefinitions
  verbs:
  - '*'
- apiGroups:
  - storage.k8s.io
  resources:
  - storageclasses
  verbs:
  - '*'
- apiGroups:
  - batch
  resources:
  - jobs
  verbs:
  - '*'
- apiGroups:
  - ""
  resources:
  - configmaps
  - pods
  - services
  - endpoints
  - persistentvolumeclaims
  - events
  verbs:
  - '*'
- apiGroups:
  - apps
  - extensions
  resources:
  - deployments
  verbs:
  - '*'
---
apiVersion: rbac.authorization.k8s.io/v1beta1
kind: ClusterRoleBinding
metadata:
  labels:
    app: mxnet-operator
    kustomize.component: mxnet-operator
  name: mxnet-operator
roleRef:
  apiGroup: rbac.authorization.k8s.io
  kind: ClusterRole
  name: mxnet-operator
subjects:
- kind: ServiceAccount
  name: mxnet-operator
  namespace: kubeflow
---
apiVersion: apps/v1
kind: Deployment
metadata:
  labels:
    kustomize.component: mxnet-operator
  name: mxnet-operator
  namespace: kubeflow
spec:
  replicas: 1
  selector:
    matchLabels:
      kustomize.component: mxnet-operator
  template:
    metadata:
      annotations:
        sidecar.istio.io/inject: "false"
      labels:
        kustomize.component: mxnet-operator
        name: mxnet-operator
    spec:
      containers:
      - command:
        - /opt/kubeflow/mxnet-operator.v1beta1
        - --alsologtostderr
        - -v=1
        env:
        - name: MY_POD_NAMESPACE
          valueFrom:
            fieldRef:
              fieldPath: metadata.namespace
        - name: MY_POD_NAME
          valueFrom:
            fieldRef:
              fieldPath: metadata.name
        image: mxjob/mxnet-operator:v1beta1
        imagePullPolicy: Always
        name: mxnet-operator
      serviceAccountName: mxnet-operator

MXNet 학습 작업 실행하기

MXTrain 모드로 MXJob을 정의한 후 학습 작업을 생성해 보겠습니다.

모델 코드 작성하기

MXNet으로 학습할 모델을 작성해 보겠습니다.

다음은 MXNet에서 제공하고 있는 이미지 분류 파이썬 코드인 image_classification.py 와 data.py 입니다. Gluon (https://github.com/apache/incubator-mxnet/tree/master/example/distributed_training) 을 사용해서 분산 학습을 지원하고 있습니다.

image_classification.py

# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   <http://www.apache.org/licenses/LICENSE-2.0>
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.

from __future__ import division

import argparse, time, os
import logging

import mxnet as mx
from mxnet import gluon
from mxnet import profiler
from mxnet.gluon import nn
from mxnet.gluon.model_zoo import vision as models
from mxnet import autograd as ag
from mxnet.test_utils import get_mnist_iterator
from mxnet.metric import Accuracy, TopKAccuracy, CompositeEvalMetric
import numpy as np

from data import (get_cifar10_iterator, get_imagenet_iterator,
                  get_caltech101_iterator, dummy_iterator)

# logging
logging.basicConfig(level=logging.INFO)
fh = logging.FileHandler('image-classification.log')
logger = logging.getLogger()
logger.addHandler(fh)
formatter = logging.Formatter('%(message)s')
fh.setFormatter(formatter)
fh.setLevel(logging.DEBUG)
logging.debug('\\n%s', '-' * 100)
formatter = logging.Formatter('%(asctime)s %(levelname)s %(message)s')
fh.setFormatter(formatter)

# CLI
parser = argparse.ArgumentParser(description='Train a model for image classification.')
parser.add_argument('--dataset', type=str, default='cifar10',
                    help='dataset to use. options are mnist, cifar10, caltech101, imagenet and dummy.')
parser.add_argument('--data-dir', type=str, default='',
                  help='training directory of imagenet images, contains train/val subdirs.')
parser.add_argument('--num-worker', '-j', dest='num_workers', default=4, type=int,
                    help='number of workers for dataloader')
parser.add_argument('--batch-size', type=int, default=32,
                    help='training batch size per device (CPU/GPU).')
parser.add_argument('--gpus', type=str, default='',
                    help='ordinates of gpus to use, can be "0,1,2" or empty for cpu only.')
parser.add_argument('--epochs', type=int, default=120,
                    help='number of training epochs.')
parser.add_argument('--lr', type=float, default=0.1,
                    help='learning rate. default is 0.1.')
parser.add_argument('--momentum', type=float, default=0.9,
                    help='momentum value for optimizer, default is 0.9.')
parser.add_argument('--wd', type=float, default=0.0001,
                    help='weight decay rate. default is 0.0001.')
parser.add_argument('--seed', type=int, default=123,
                    help='random seed to use. Default=123.')
parser.add_argument('--mode', type=str,
                    help='mode in which to train the model. options are symbolic, imperative, hybrid')
parser.add_argument('--model', type=str, required=True,
                    help='type of model to use. see vision_model for options.')
parser.add_argument('--use_thumbnail', action='store_true',
                    help='use thumbnail or not in resnet. default is false.')
parser.add_argument('--batch-norm', action='store_true',
                    help='enable batch normalization or not in vgg. default is false.')
parser.add_argument('--use-pretrained', action='store_true',
                    help='enable using pretrained model from gluon.')
parser.add_argument('--prefix', default='', type=str,
                    help='path to checkpoint prefix, default is current working dir')
parser.add_argument('--start-epoch', default=0, type=int,
                    help='starting epoch, 0 for fresh training, > 0 to resume')
parser.add_argument('--resume', type=str, default='',
                    help='path to saved weight where you want resume')
parser.add_argument('--lr-factor', default=0.1, type=float,
                    help='learning rate decay ratio')
parser.add_argument('--lr-steps', default='30,60,90', type=str,
                    help='list of learning rate decay epochs as in str')
parser.add_argument('--dtype', default='float32', type=str,
                    help='data type, float32 or float16 if applicable')
parser.add_argument('--save-frequency', default=10, type=int,
                    help='epoch frequence to save model, best model will always be saved')
parser.add_argument('--kvstore', type=str, default='device',
                    help='kvstore to use for trainer/module.')
parser.add_argument('--log-interval', type=int, default=50,
                    help='Number of batches to wait before logging.')
parser.add_argument('--profile', action='store_true',
                    help='Option to turn on memory profiling for front-end, '\\
                         'and prints out the memory usage by python function at the end.')
parser.add_argument('--builtin-profiler', type=int, default=0, help='Enable built-in profiler (0=off, 1=on)')
opt = parser.parse_args()

# global variables
logger.info('Starting new image-classification task:, %s',opt)
mx.random.seed(opt.seed)
model_name = opt.model
dataset_classes = {'mnist': 10, 'cifar10': 10, 'caltech101':101, 'imagenet': 1000, 'dummy': 1000}
batch_size, dataset, classes = opt.batch_size, opt.dataset, dataset_classes[opt.dataset]
context = [mx.gpu(int(i)) for i in opt.gpus.split(',')] if opt.gpus.strip() else [mx.cpu()]
num_gpus = len(context)
batch_size *= max(1, num_gpus)
lr_steps = [int(x) for x in opt.lr_steps.split(',') if x.strip()]
metric = CompositeEvalMetric([Accuracy(), TopKAccuracy(5)])
kv = mx.kv.create(opt.kvstore)

def get_model(model, ctx, opt):
    """Model initialization."""
    kwargs = {'ctx': ctx, 'pretrained': opt.use_pretrained, 'classes': classes}
    if model.startswith('resnet'):
        kwargs['thumbnail'] = opt.use_thumbnail
    elif model.startswith('vgg'):
        kwargs['batch_norm'] = opt.batch_norm

    net = models.get_model(model, **kwargs)
    if opt.resume:
        net.load_parameters(opt.resume)
    elif not opt.use_pretrained:
        if model in ['alexnet']:
            net.initialize(mx.init.Normal())
        else:
            net.initialize(mx.init.Xavier(magnitude=2))
    net.cast(opt.dtype)
    return net

net = get_model(opt.model, context, opt)

def get_data_iters(dataset, batch_size, opt):
    """get dataset iterators"""
    if dataset == 'mnist':
        train_data, val_data = get_mnist_iterator(batch_size, (1, 28, 28),
                                                  num_parts=kv.num_workers, part_index=kv.rank)
    elif dataset == 'cifar10':
        train_data, val_data = get_cifar10_iterator(batch_size, (3, 32, 32),
                                                    num_parts=kv.num_workers, part_index=kv.rank)
    elif dataset == 'imagenet':
        shape_dim = 299 if model_name == 'inceptionv3' else 224

        if not opt.data_dir:
            raise ValueError('Dir containing raw images in train/val is required for imagenet.'
                             'Please specify "--data-dir"')

        train_data, val_data = get_imagenet_iterator(opt.data_dir, batch_size,
                                                                opt.num_workers, shape_dim, opt.dtype)
    elif dataset == 'caltech101':
        train_data, val_data = get_caltech101_iterator(batch_size, opt.num_workers, opt.dtype)
    elif dataset == 'dummy':
        shape_dim = 299 if model_name == 'inceptionv3' else 224
        train_data, val_data = dummy_iterator(batch_size, (3, shape_dim, shape_dim))
    return train_data, val_data

def test(ctx, val_data):
    metric.reset()
    val_data.reset()
    for batch in val_data:
        data = gluon.utils.split_and_load(batch.data[0].astype(opt.dtype, copy=False),
                                          ctx_list=ctx, batch_axis=0)
        label = gluon.utils.split_and_load(batch.label[0].astype(opt.dtype, copy=False),
                                           ctx_list=ctx, batch_axis=0)
        outputs = [net(X) for X in data]
        metric.update(label, outputs)
    return metric.get()

def update_learning_rate(lr, trainer, epoch, ratio, steps):
    """Set the learning rate to the initial value decayed by ratio every N epochs."""
    new_lr = lr * (ratio ** int(np.sum(np.array(steps) < epoch)))
    trainer.set_learning_rate(new_lr)
    return trainer

def save_checkpoint(epoch, top1, best_acc):
    if opt.save_frequency and (epoch + 1) % opt.save_frequency == 0:
        fname = os.path.join(opt.prefix, '%s_%d_acc_%.4f.params' % (opt.model, epoch, top1))
        net.save_parameters(fname)
        logger.info('[Epoch %d] Saving checkpoint to %s with Accuracy: %.4f', epoch, fname, top1)
    if top1 > best_acc[0]:
        best_acc[0] = top1
        fname = os.path.join(opt.prefix, '%s_best.params' % (opt.model))
        net.save_parameters(fname)
        logger.info('[Epoch %d] Saving checkpoint to %s with Accuracy: %.4f', epoch, fname, top1)

def train(opt, ctx):
    if isinstance(ctx, mx.Context):
        ctx = [ctx]

    train_data, val_data = get_data_iters(dataset, batch_size, opt)
    net.collect_params().reset_ctx(ctx)
    trainer = gluon.Trainer(net.collect_params(), 'sgd',
                            optimizer_params={'learning_rate': opt.lr,
                                              'wd': opt.wd,
                                              'momentum': opt.momentum,
                                              'multi_precision': True},
                            kvstore=kv)
    loss = gluon.loss.SoftmaxCrossEntropyLoss()

    total_time = 0
    num_epochs = 0
    best_acc = [0]
    for epoch in range(opt.start_epoch, opt.epochs):
        trainer = update_learning_rate(opt.lr, trainer, epoch, opt.lr_factor, lr_steps)
        tic = time.time()
        train_data.reset()
        metric.reset()
        btic = time.time()
        for i, batch in enumerate(train_data):
            data = gluon.utils.split_and_load(batch.data[0].astype(opt.dtype), ctx_list=ctx, batch_axis=0)
            label = gluon.utils.split_and_load(batch.label[0].astype(opt.dtype), ctx_list=ctx, batch_axis=0)
            outputs = []
            Ls = []
            with ag.record():
                for x, y in zip(data, label):
                    z = net(x)
                    L = loss(z, y)
                    # store the loss and do backward after we have done forward
                    # on all GPUs for better speed on multiple GPUs.
                    Ls.append(L)
                    outputs.append(z)
                ag.backward(Ls)
            trainer.step(batch.data[0].shape[0])
            metric.update(label, outputs)
            if opt.log_interval and not (i+1)%opt.log_interval:
                name, acc = metric.get()
                logger.info('Epoch[%d] Batch [%d]\\tSpeed: %f samples/sec\\t%s=%f, %s=%f'%(
                               epoch, i, batch_size/(time.time()-btic), name[0], acc[0], name[1], acc[1]))
            btic = time.time()

        epoch_time = time.time()-tic

        # First epoch will usually be much slower than the subsequent epics,
        # so don't factor into the average
        if num_epochs > 0:
          total_time = total_time + epoch_time
        num_epochs = num_epochs + 1

        name, acc = metric.get()
        logger.info('[Epoch %d] training: %s=%f, %s=%f'%(epoch, name[0], acc[0], name[1], acc[1]))
        logger.info('[Epoch %d] time cost: %f'%(epoch, epoch_time))
        name, val_acc = test(ctx, val_data)
        logger.info('[Epoch %d] validation: %s=%f, %s=%f'%(epoch, name[0], val_acc[0], name[1], val_acc[1]))

        # save model if meet requirements
        save_checkpoint(epoch, val_acc[0], best_acc)
    if num_epochs > 1:
        print('Average epoch time: {}'.format(float(total_time)/(num_epochs - 1)))

def main():
    if opt.builtin_profiler > 0:
        profiler.set_config(profile_all=True, aggregate_stats=True)
        profiler.set_state('run')
    if opt.mode == 'symbolic':
        data = mx.sym.var('data')
        if opt.dtype == 'float16':
            data = mx.sym.Cast(data=data, dtype=np.float16)
        out = net(data)
        if opt.dtype == 'float16':
            out = mx.sym.Cast(data=out, dtype=np.float32)
        softmax = mx.sym.SoftmaxOutput(out, name='softmax')
        mod = mx.mod.Module(softmax, context=context)
        train_data, val_data = get_data_iters(dataset, batch_size, opt)
        mod.fit(train_data,
                eval_data=val_data,
                num_epoch=opt.epochs,
                kvstore=kv,
                batch_end_callback = mx.callback.Speedometer(batch_size, max(1, opt.log_interval)),
                epoch_end_callback = mx.callback.do_checkpoint('image-classifier-%s'% opt.model),
                optimizer = 'sgd',
                optimizer_params = {'learning_rate': opt.lr, 'wd': opt.wd, 'momentum': opt.momentum, 'multi_precision': True},
                initializer = mx.init.Xavier(magnitude=2))
        mod.save_parameters('image-classifier-%s-%d-final.params'%(opt.model, opt.epochs))
    else:
        if opt.mode == 'hybrid':
            net.hybridize()
        train(opt, context)
    if opt.builtin_profiler > 0:
        profiler.set_state('stop')
        print(profiler.dumps())

if __name__ == '__main__':
    if opt.profile:
        import hotshot, hotshot.stats
        prof = hotshot.Profile('image-classifier-%s-%s.prof'%(opt.model, opt.mode))
        prof.runcall(main)
        prof.close()
        stats = hotshot.stats.load('image-classifier-%s-%s.prof'%(opt.model, opt.mode))
        stats.strip_dirs()
        stats.sort_stats('cumtime', 'calls')
        stats.print_stats()
    else:
        main()

data.py

# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   <http://www.apache.org/licenses/LICENSE-2.0>
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.

# pylint: skip-file
""" data iterator for mnist """
import os
import random
import tarfile
import logging
import tarfile
logging.basicConfig(level=logging.INFO)

import mxnet as mx
from mxnet.test_utils import get_cifar10
from mxnet.gluon.data.vision import ImageFolderDataset
from mxnet.gluon.data import DataLoader
from mxnet.contrib.io import DataLoaderIter

def get_cifar10_iterator(batch_size, data_shape, resize=-1, num_parts=1, part_index=0):
    get_cifar10()

    train = mx.io.ImageRecordIter(
        path_imgrec = "data/cifar/train.rec",
        # mean_img    = "data/cifar/mean.bin",
        resize      = resize,
        data_shape  = data_shape,
        batch_size  = batch_size,
        rand_crop   = True,
        rand_mirror = True,
        num_parts=num_parts,
        part_index=part_index)

    val = mx.io.ImageRecordIter(
        path_imgrec = "data/cifar/test.rec",
        # mean_img    = "data/cifar/mean.bin",
        resize      = resize,
        rand_crop   = False,
        rand_mirror = False,
        data_shape  = data_shape,
        batch_size  = batch_size,
        num_parts=num_parts,
        part_index=part_index)

    return train, val

def get_imagenet_transforms(data_shape=224, dtype='float32'):
    def train_transform(image, label):
        image, _ = mx.image.random_size_crop(image, (data_shape, data_shape), 0.08, (3/4., 4/3.))
        image = mx.nd.image.random_flip_left_right(image)
        image = mx.nd.image.to_tensor(image)
        image = mx.nd.image.normalize(image, mean=(0.485, 0.456, 0.406), std=(0.229, 0.224, 0.225))
        return mx.nd.cast(image, dtype), label

    def val_transform(image, label):
        image = mx.image.resize_short(image, data_shape + 32)
        image, _ = mx.image.center_crop(image, (data_shape, data_shape))
        image = mx.nd.image.to_tensor(image)
        image = mx.nd.image.normalize(image, mean=(0.485, 0.456, 0.406), std=(0.229, 0.224, 0.225))
        return mx.nd.cast(image, dtype), label
    return train_transform, val_transform

def get_imagenet_iterator(root, batch_size, num_workers, data_shape=224, dtype='float32'):
    """Dataset loader with preprocessing."""
    train_dir = os.path.join(root, 'train')
    train_transform, val_transform = get_imagenet_transforms(data_shape, dtype)
    logging.info("Loading image folder %s, this may take a bit long...", train_dir)
    train_dataset = ImageFolderDataset(train_dir, transform=train_transform)
    train_data = DataLoader(train_dataset, batch_size, shuffle=True,
                            last_batch='discard', num_workers=num_workers)
    val_dir = os.path.join(root, 'val')
    if not os.path.isdir(os.path.expanduser(os.path.join(root, 'val', 'n01440764'))):
        user_warning = 'Make sure validation images are stored in one subdir per category, a helper script is available at <https://git.io/vNQv1>'
        raise ValueError(user_warning)
    logging.info("Loading image folder %s, this may take a bit long...", val_dir)
    val_dataset = ImageFolderDataset(val_dir, transform=val_transform)
    val_data = DataLoader(val_dataset, batch_size, last_batch='keep', num_workers=num_workers)
    return DataLoaderIter(train_data, dtype), DataLoaderIter(val_data, dtype)

def get_caltech101_data():
    url = "<https://s3.us-east-2.amazonaws.com/mxnet-public/101_ObjectCategories.tar.gz>"
    dataset_name = "101_ObjectCategories"
    data_folder = "data"
    if not os.path.isdir(data_folder):
        os.makedirs(data_folder)
    tar_path = mx.gluon.utils.download(url, path=data_folder)
    if (not os.path.isdir(os.path.join(data_folder, "101_ObjectCategories")) or
        not os.path.isdir(os.path.join(data_folder, "101_ObjectCategories_test"))):
        tar = tarfile.open(tar_path, "r:gz")
        tar.extractall(data_folder)
        tar.close()
        print('Data extracted')
    training_path = os.path.join(data_folder, dataset_name)
    testing_path = os.path.join(data_folder, "{}_test".format(dataset_name))
    return training_path, testing_path

def get_caltech101_iterator(batch_size, num_workers, dtype):
    def transform(image, label):
        # resize the shorter edge to 224, the longer edge will be greater or equal to 224
        resized = mx.image.resize_short(image, 224)
        # center and crop an area of size (224,224)
        cropped, crop_info = mx.image.center_crop(resized, (224, 224))
        # transpose the channels to be (3,224,224)
        transposed = mx.nd.transpose(cropped, (2, 0, 1))
        return transposed, label

    training_path, testing_path = get_caltech101_data()
    dataset_train = ImageFolderDataset(root=training_path, transform=transform)
    dataset_test = ImageFolderDataset(root=testing_path, transform=transform)

    train_data = DataLoader(dataset_train, batch_size, shuffle=True, num_workers=num_workers)
    test_data = DataLoader(dataset_test, batch_size, shuffle=False, num_workers=num_workers)
    return DataLoaderIter(train_data), DataLoaderIter(test_data)

class DummyIter(mx.io.DataIter):
    def __init__(self, batch_size, data_shape, batches = 100):
        super(DummyIter, self).__init__(batch_size)
        self.data_shape = (batch_size,) + data_shape
        self.label_shape = (batch_size,)
        self.provide_data = [('data', self.data_shape)]
        self.provide_label = [('softmax_label', self.label_shape)]
        self.batch = mx.io.DataBatch(data=[mx.nd.zeros(self.data_shape)],
                                     label=[mx.nd.zeros(self.label_shape)])
        self._batches = 0
        self.batches = batches

    def next(self):
        if self._batches < self.batches:
            self._batches += 1
            return self.batch
        else:
            self._batches = 0
            raise StopIteration

def dummy_iterator(batch_size, data_shape):
    return DummyIter(batch_size, data_shape), DummyIter(batch_size, data_shape)

class ImagePairIter(mx.io.DataIter):
    def __init__(self, path, data_shape, label_shape, batch_size=64, flag=0, input_aug=None, target_aug=None):
        super(ImagePairIter, self).__init__(batch_size)
        self.data_shape = (batch_size,) + data_shape
        self.label_shape = (batch_size,) + label_shape
        self.input_aug = input_aug
        self.target_aug = target_aug
        self.provide_data = [('data', self.data_shape)]
        self.provide_label = [('label', self.label_shape)]
        is_image_file = lambda fn: any(fn.endswith(ext) for ext in [".png", ".jpg", ".jpeg"])
        self.filenames = [os.path.join(path, x) for x in os.listdir(path) if is_image_file(x)]
        self.count = 0
        self.flag = flag
        random.shuffle(self.filenames)

    def next(self):
        from PIL import Image
        if self.count + self.batch_size <= len(self.filenames):
            data = []
            label = []
            for i in range(self.batch_size):
                fn = self.filenames[self.count]
                self.count += 1
                image = Image.open(fn).convert('YCbCr').split()[0]
                if image.size[0] > image.size[1]:
                    image = image.transpose(Image.TRANSPOSE)
                image = mx.nd.expand_dims(mx.nd.array(image), axis=2)
                target = image.copy()
                for aug in self.input_aug:
                    image = aug(image)
                for aug in self.target_aug:
                    target = aug(target)
                data.append(image)
                label.append(target)

            data = mx.nd.concat(*[mx.nd.expand_dims(d, axis=0) for d in data], dim=0)
            label = mx.nd.concat(*[mx.nd.expand_dims(d, axis=0) for d in label], dim=0)
            data = [mx.nd.transpose(data, axes=(0, 3, 1, 2)).astype('float32')/255]
            label = [mx.nd.transpose(label, axes=(0, 3, 1, 2)).astype('float32')/255]

            return mx.io.DataBatch(data=data, label=label)
        else:
            raise StopIteration

    def reset(self):
        self.count = 0
        random.shuffle(self.filenames)

모델 컨테이너 이미지 만들기

모델 학습용 컨테이너 이미지를 만들기 위해서 Dockerfile을 생성하겠습니다.

다음은 MXNet 1.5를 기반 이미지로 해서, 모델 파일을 추가하는 Dockerfile 입니다.

Dockerfile

FROM mxnet/python:1.5.0_cpu_py3

RUN mkdir -p /app
COPY image_classification.py /app/
COPY data.py /app/

WORKDIR /app

CMD ["python3", "image_classification.py", "--dataset", "cifar10", "--model", "vgg11", "--epochs", "1", "--kvstore", "dist_sync"]

다음 명령어를 실행하면 kangwoo/mxnet:cpu 라는 이름의 컨테이너 이미지를 빌드 할 수 있습니다.

docker build -t kangwoo/mxnet:cpu .

빌드한 컨테이너 이미지를 컨테이너 레지스트리로 업로드 하겠습니다.

docker push kangwoo/mxnet:cpu

MXJob 생성하기

mxnet-operator를 사용해서 MXNet 모델 학습을 하라면 MXJob을 정의해야합니다.

  1. jobMode를 MXTrain로 설정합니다. jobMode: MXTrain
  2. mxReplicaSpecs를 설정합니다. “mxReplicaSpecs”은 MXNet의 분산 학습시 사용하는 프로세스들을 정의하는데 사용합니다. Scheduler와 Server를 1개로 설정하고, Worker도 1개로 설정합니다. Kubeflow 클러스터에 istio가 설치되어 있기 때문에, 자동으로 istio-proxy가 포드에 주입됩니다. 이것을 방지하기 위해서 어노테이션에 sidecar.istio.io/inject: “false” 을 추가해 주었습니다. mxReplicaSpecs: Scheduler: replicas: 1 restartPolicy: Never template: metadata: annotations: sidecar.istio.io/inject: “false” spec: containers: – name: mxnet image: kangwoo/mxnet:cpu Server: replicas: 1 restartPolicy: Never template: metadata: annotations: sidecar.istio.io/inject: “false” spec: containers: – name: mxnet image: kangwoo/mxnet:cpu Worker: replicas: 1 restartPolicy: Never template: metadata: annotations: sidecar.istio.io/inject: “false” spec: containers: – name: mxnet image: kangwoo/mxnet:cpu

다음은 MXJob 을 생성한 위한 메니페스트입니다.

mxnet-job.yaml

apiVersion: "kubeflow.org/v1beta1"
kind: "MXJob"
metadata:
  name: "mxnet-job"
spec:
  jobMode: MXTrain
  mxReplicaSpecs:
    Scheduler:
      replicas: 1
      restartPolicy: Never
      template:
        metadata:
          annotations:
            sidecar.istio.io/inject: "false"
        spec:
          containers:
            - name: mxnet
              image: kangwoo/mxnet:cpu
    Server:
      replicas: 1
      restartPolicy: Never
      template:
        metadata:
          annotations:
            sidecar.istio.io/inject: "false"
        spec:
          containers:
            - name: mxnet
              image: kangwoo/mxnet:cpu
    Worker:
      replicas: 1
      restartPolicy: Never
      template:
        metadata:
          annotations:
            sidecar.istio.io/inject: "false"
        spec:
          containers:
            - name: mxnet
              image: kangwoo/mxnet:cpu

다음 명령어를 실행하면 admin 네임스페이스에 mxnet-job 이라는 이름의 MXJob을 생성할 수 있습니다.

kubectl -n admin apply -f mxnet-job.yaml

MXNet 학습 작업 확인하기

생성한 MXJob은 다음 명령어를 실행해서 확인 해 볼 수 있습니다.

kubectl -n admin get mxjob

생성된 MXJob이 있다면, 다음과 같은 응답 결과를 얻을 수 있습니다.

NAME        AGE
mxnet-job   5s

MXJob이 생성되면, mxnet-operator 에 의해서 포드들이 생성됩니다. MXjob 매니페스트에 정의한 갯수대로 scheduler, server, worker 포드가 생성되게 됩니다.

생성된 포드들은 다음 명령어를 실행해서 확인 해 볼 수 있습니다.

kubectl -n admin get pod -l mxnet_job_name=mxnet-job

생성된 포드들이 남아 있다면, 다음과 같은 응답 결과를 얻을 수 있습니다.

NAME                    READY   STATUS    RESTARTS   AGE
mxnet-job-scheduler-0   1/1     Running   0          9s
mxnet-job-server-0      1/1     Running   0          9s
mxnet-job-worker-0      1/1     Running   0          9

MXJob은 작업이 끝난 후, 관련 포드들을 삭제해버립니다. 그래서 작업이 완료되면 포드가 조회되지 않을 수 있습니다. 작업이 완료되어도 포드들을 남겨 두고 싶다면, MXJob 매니페스트의 spec 부분에 “cleanPodPolicy: None” 를 추가하시면 됩니다.

MXJob spec의 CleanPodPolicy는 작업이 종료 될 때 포드 삭제를 제어할 때 사용합니다. 다음 값들 중 하나를 사용할 수 있습니다.

  • Running : 작업이 완료되었을 때, 실행이 끝난(Completed) 포드들은 삭제하지 않고, 실행중인(Running) 포드들만 삭제합니다.
  • All : 작업이 완료되었을 때, 실행이 끝난 포드들을 즉시 삭제합니다.
  • None : 작업이 완료되어도 포드들을 삭제하지 않습니다.

다음은 cleanPodPolicy를 추가한 메니페스트 예제입니다.

apiVersion: "kubeflow.org/v1beta1"
kind: "MXJob"
metadata:
  name: "mxnet-job"
spec:
  cleanPodPolicy: None
...

MXJob의 작업 상태를 알고 싶으면 describe 명령어를 사용할 수 있습니다.

다음 명령어를 실행하면 admin 네임스페이스에 mxnet-job 이라는 이름의 MXJob의 상태를 조회할 수 있습니다.

kubectl -n admin describe mxjob mxnet-job

다음은 예제 작업에 대한 샘플 출력입니다.

Name:         mxnet-job
Namespace:    admin
Labels:       <none>
Annotations:  ...
API Version:  kubeflow.org/v1beta1
Kind:         MXJob
Metadata:
...
Spec:
...
Status:
  Completion Time:  2020-03-08T00:44:23Z
  Conditions:
    Last Transition Time:  2020-03-07T23:45:12Z
    Last Update Time:      2020-03-07T23:45:12Z
    Message:               MXJob mxnet-job is created.
    Reason:                MXJobCreated
    Status:                True
    Type:                  Created
    Last Transition Time:  2020-03-07T23:45:12Z
    Last Update Time:      2020-03-07T23:45:14Z
    Message:               MXJob mxnet-job is running.
    Reason:                MXJobRunning
    Status:                False
    Type:                  Running
    Last Transition Time:  2020-03-07T23:45:12Z
    Last Update Time:      2020-03-08T00:44:23Z
    Message:               MXJob mxnet-job is successfully completed.
    Reason:                MXJobSucceeded
    Status:                True
    Type:                  Succeeded
  Mx Replica Statuses:
    Scheduler:
    Server:
    Worker:
  Start Time:  2020-03-07T23:45:14Z
Events:        <none>

MXNet 학습 작업 삭제하기

작업이 완료되어도 MXJob은 삭제되지 않습니다.

다음 명령어를 실행하면 admin 네임스페이스에 mxnet-job 이라는 이름의 MXJob을 삭제할 수 있습니다.

kubectl -n admin delete mxjob mxnet-job

참고

작성일자

Kubeflow – Jupyter Notebooks 커스텀 이미지

주피터 노트북 커스텀 이미지

주피터 노트북 커스텀 이미지 만드는 방법

주피터 노트북에서 사용할 사용자 커스텀 이미지를 만드는 방법에 대해서 알아보겠습니다.

Kubeflow에서 사용자가 만든 커스텀 이미지를 사용하려면 몇 가지 요구 사항을 충족해야합니다. Kubeflow는 컨테이너 이미지가 실행되면, 주피터가 자동적으로 시작되는 것으로 알고 있습니다. 그래서 컨테이너 이미지에 주피터를 시작하는 기본 명령을 설정해야합니다.

먼저 주피터를 시작하는 명령어가 필요합니다. 다음은 주피터를 실행하는 명령어 입니다.

jupyter notebook

그리고 주피터에게 설정 정보를 넘겨줘야합니다. 다음은 주피터 실행에 필요한 설정 정보들입니다.

  • 작업 디렉토리 : /home/jovyan 디렉토리는 쿠버네티스 PV와 마운트 됩니다. –notebook-dir=/home/jovyan
  • 접근 허용 IP : 주피터 노트북 서버에 모든 IP 에서 접근가능하도록 합니다. –ip=0.0.0.0
  • 노트북 루트 권한 : 사용자가 노트북을 루트로 실행하는것을 허용합니다. –allow-root
  • 포트 설정 : 주피터 포드의 포트를 설정합니다. –port=8888
  • 인증 비활성화 : 주피터의 인증 기능을 비활성화 합니다. Kubeflow에서 사용하는 istio가 인증을 담당하기 때문에, 주피터에서 제공하는 기능을 비활성화 시키는것입니다. –NotebookApp.token=” –NotebookApp.password=”
  • 모든 오리진(origin) 허용 : 주피터 노트북 서버에 모든 오리진이 접근할 수 있도록 허용합니다. –NotebookApp.allow_origin=’*’
  • 기본 URL 설정 : Kubeflow에서 노트북 서버를 관리하는 노트북 컨트롤러는 NB_PREFIX 라는 환경 변수를 사용하여 노트북 서버의 기본 URL을 넘겨줍니다. –NotebookApp.base_url=${NB_PREFIX}

다음은 Dockerfile에 포함해야할 CMD 예제입니다.

ENV NB_PREFIX /

CMD ["sh","-c", "jupyter notebook --notebook-dir=/home/jovyan --ip=0.0.0.0 --no-browser --allow-root --port=8888 --NotebookApp.token='' --NotebookApp.password='' --NotebookApp.allow_origin='*' --NotebookApp.base_url=${NB_PREFIX}"]

주의 하실 점은 ${NB_PREFIX} 라는 환경 변수를 사용하기 때문에 sh 이나 bash 등을 이용해서 노트북을 실행해야합니다.

주피터 노트북 커스텀 이미지 만들기

기존 주피터 노트북 이미지로 만들기

Kubeflow에서 기본으로 제공하는 주피터 노트북 이미지를 가지고 커스텀 이미지를 만들어 보겠습니다.

다음은 https://github.com/kubeflow/kubeflow/blob/master/components/tensorflow-notebook-image/Dockerfile 을 약간 수정한 Dockerfile 입니다.

# Copyright (c) Jupyter Development Team.
# Distributed under the terms of the Modified BSD License.

ARG BASE_IMAGE=tensorflow/tensorflow:2.1.0-py3-jupyter

FROM $BASE_IMAGE

ARG TF_SERVING_VERSION=0.0.0
ARG NB_USER=jovyan

# TODO: User should be refactored instead of hard coded jovyan

USER root

ENV DEBIAN_FRONTEND noninteractive

ENV NB_USER $NB_USER

ENV NB_UID 1000
ENV HOME /home/$NB_USER
ENV NB_PREFIX /


# Use bash instead of sh
SHELL ["/bin/bash", "-c"]

RUN apt-get update && apt-get install -yq --no-install-recommends \\
  apt-transport-https \\
  build-essential \\
  bzip2 \\
  ca-certificates \\
  curl \\
  g++ \\
  git \\
  gnupg \\
  graphviz \\
  locales \\
  lsb-release \\
  openssh-client \\
  sudo \\
  unzip \\
  vim \\
  wget \\
  zip \\
  emacs \\
  python3-pip \\
  python3-dev \\
  python3-setuptools \\
  && apt-get clean && \\
  rm -rf /var/lib/apt/lists/*

# Install Nodejs for jupyterlab-manager
RUN curl -sL <https://deb.nodesource.com/setup_12.x> | sudo -E bash -
RUN apt-get update && apt-get install -yq --no-install-recommends \\
  nodejs \\
  && apt-get clean && \\
  rm -rf /var/lib/apt/lists/*

ENV DOCKER_CREDENTIAL_GCR_VERSION=1.4.3
RUN curl -LO <https://github.com/GoogleCloudPlatform/docker-credential-gcr/releases/download/v${DOCKER_CREDENTIAL_GCR_VERSION}/docker-credential-gcr_linux_amd64-${DOCKER_CREDENTIAL_GCR_VERSION}.tar.gz> && \\
    tar -zxvf docker-credential-gcr_linux_amd64-${DOCKER_CREDENTIAL_GCR_VERSION}.tar.gz && \\
    mv docker-credential-gcr /usr/local/bin/docker-credential-gcr && \\
    rm docker-credential-gcr_linux_amd64-${DOCKER_CREDENTIAL_GCR_VERSION}.tar.gz && \\
    chmod +x /usr/local/bin/docker-credential-gcr

# Install AWS CLI
RUN curl "<https://s3.amazonaws.com/aws-cli/awscli-bundle.zip>" -o "/tmp/awscli-bundle.zip" && \\
    unzip /tmp/awscli-bundle.zip && ./awscli-bundle/install -i /usr/local/aws -b /usr/local/bin/aws && \\
    rm -rf ./awscli-bundle


RUN echo "en_US.UTF-8 UTF-8" > /etc/locale.gen && \\
    locale-gen

ENV LC_ALL en_US.UTF-8
ENV LANG en_US.UTF-8
ENV LANGUAGE en_US.UTF-8

# Create NB_USER user with UID=1000 and in the 'users' group
# but allow for non-initial launches of the notebook to have
# $HOME provided by the contents of a PV
RUN useradd -M -s /bin/bash -N -u $NB_UID $NB_USER && \\
    chown -R ${NB_USER}:users /usr/local/bin && \\
    mkdir -p $HOME

RUN export CLOUD_SDK_REPO="cloud-sdk-$(lsb_release -c -s)" && \\
    echo "deb <https://packages.cloud.google.com/apt> $CLOUD_SDK_REPO main" > /etc/apt/sources.list.d/google-cloud-sdk.list && \\
    curl <https://packages.cloud.google.com/apt/doc/apt-key.gpg> | apt-key add - && \\
    apt-get update && \\
    apt-get install -y google-cloud-sdk kubectl

# Install Tini - used as entrypoint for container
RUN cd /tmp && \\
    wget --quiet <https://github.com/krallin/tini/releases/download/v0.18.0/tini> && \\
    echo "12d20136605531b09a2c2dac02ccee85e1b874eb322ef6baf7561cd93f93c855 *tini" | sha256sum -c - && \\
    mv tini /usr/local/bin/tini && \\
    chmod +x /usr/local/bin/tini

# NOTE: Beyond this point be careful of breaking out
# or otherwise adding new layers with RUN, chown, etc.
# The image size can grow significantly.

# Install base python3 packages
RUN pip3 --no-cache-dir install \\
    jupyter-console==6.0.0 \\
    jupyterlab \\
    xgboost \\
    kubeflow-fairing==0.7.1.1


RUN docker-credential-gcr configure-docker && chown ${NB_USER}:users $HOME/.docker/config.json

# Configure container startup
EXPOSE 8888
USER jovyan
ENTRYPOINT ["tini", "--"]
CMD ["sh","-c", "jupyter lab --notebook-dir=/home/${NB_USER} --ip=0.0.0.0 --no-browser --allow-root --port=8888 --NotebookApp.token='' --NotebookApp.password='' --NotebookApp.allow_origin='*' --NotebookApp.base_url=${NB_PREFIX}"]

베이스 이미지를 tensorflow/tensorflow:2.1.0-py3-jupyter 로 사용하였고, 파이썬 패키지인 kubeflow-fairing을 0.7.1 버전으로 설치하였습니다. 그리고 CMD를 수정해서 주피터 노트북이 아니라, 주피터 랩이 실행되게 하였습니다.

docker build -t kangwoo/tensorflow-2.1.0-notebook-cpu:1.0.0 .

생성한 컨테이너 이미지를 컨테이너 이미지 레지스트리에 푸시 하려면 접근 권한이 필요합니다. “컨테이너 이미지 레지스트리에 접근할 수 있도록 도커 설정하기”를 참고하기 바랍니다.

docker push kangwoo/tensorflow-2.1.0-notebook-cpu:1.0.0

노트북 목록 화면에서, “NEW SERVER” 버튼을 클릭하여, 새로운 노트북 서버를 생성하는 페이지로 이동합니다.

“Custom Image”를 체크하고, 새로 만든 커스텀 이미지 주소를 입력합니다.

나머지 필드에는 적당한 값을 입력하고, “LAUNCH” 버튼을 클릭하여, 새로운 노트북 서버를 생성합니다.

노트북 서버 목록에서 CONNECT 버튼을 누르고, 노트북 랩에 접속할 수 있습니다.