💻Develop/🛝Kubeflow

[Kubeflow] Component 사용해보기

Guraeng 2023. 5. 31. 15:21

Component?

이전에는 Function을 만들어서 Pipeline에 사용하는 방법과 Volume을 연결하는 방법에 대해 간단하게 적어봤다.

직접 함수를 만들어 사용하는 방법은 간단한 동작이나 잠깐 사용하고 다시 사용할일이 없으면 상관 없으나, 보통 그렇지 않아 불편을 유발한다.

이러한 점을 해결하기 위한 Component 방식으로 Pipeline을 구축할 경우 코드의 재사용성과 모듈성이 향상된다.
독립적인 작업 단위로 설계되어 다른 파이프라인에서 모듈처럼 불러와 사용할 수 있다.

# 함수 단위 파이프라인
@func_to_container_op
def func_sum(val1: int, val2: in2):
    return val1 + val2

@dsl.pipeline(name="test pipeline")
def pipeline(val1: int, val2: int):
    sum_result = func_sum(val1, val2)

# 컴포넌트 단위 파이프라인
import kfp.components as comp
compo_sum = comp.load_component_from_file(file_path)

@dsl.pipeline(nmae="test pipeline")
def pipeline(val1: int, val2: int):
    step_sum = compo_sum(val1, val2)

 

Component 생성 과정

1. 컴포넌트의 인터페이스 정의

컴포넌트의 입력과 출력을 정의한다.

컴포넌트가 어떤 데이터나 매개변수를 입력으로 필요로 하는지, 어떤 데이터를 출력으로 생성하는지를 결정한다.

# interface.yaml
name: float_sum_interface
description: integer sum float

inputs:
    - { name: intValue, type: Integer, default: '1', description: Integer value}
    - { name: floatValue, type: Float, default: '3.14159', description: Float value}
outputs:
    - { name: floatOutput, type: String, description: float output}

implementation:
  container:
        image: docker.io/sgyeong97/kubelofw:v0.1.4
        command: [
        python3,
        /pipelines/component/src/component.py,
        --input1-path,
        {inputValue: intValue},
        --input2-path,
        {inputValue: floatValue},
        --output1-path,
        {outputPath: floatOutput}
    ]
  • name : 인터페이스의 이름을 정의한다.
  • description : 선택사항이며, 사람이 읽을 수 있는 설명을 명시한다.
  • inputs, outputs의 데이터 type의 경우 문서를 참고하여 사용하면 된다.
  • implementation : 컴포넌트의 구현 세부 정보 (컴포넌트가 실행되는 환경과 실행되는 명령어)를 정의 한다.
    해당 문서의 경우 container의 이미지와 관련된 정보를 명시하고있다.
    • image : 저장소에 올라간 컨테이너 이미지를 지정한다. 이 부분이 이해가 되지 않을 경우 Docker에 대해 알아볼 필요가 있다.
    • command : 컨테이너에서 프로그램을 실행하는 데 사용되는 명령줄 인수를 지정한다.
    • python3, /pipelines/component/src/float_sum.py
      컨테이너 내의 경로에 위치한 Python 스크립트를 실행하는 명령어 명시
    • 나머지는 입/출력 매개변수에 관하여 스크립트에 전달하는걸 정의하는 것 이다.
      이에 관하여 자세한 정보는 공식 문서에 나와있으며 자주 사용하는 3개는 아래와 같다.
      1. inputValue 지정된 입력 값으로 대체된다. 숫자나 작은 문자열과 같은 작은 입력 데이터에 쓰인다.
      2. inputPath 입력에 대한 경로를 파일로 대체한다. 구성 요소는 파이프라인이 실행되는 중에 해당 경로에서 입력 내용을 읽을 수 있다.
      3. outputPath 프로그램이 출력 데이터를 쓰는 경로로 대체된다. Kubeflow Pipeline 시스템이 파일의 내용을 읽고 지정된 출력 값으로 저장할 수 있게 해준다.

2. 컴포넌트 구현

# component.py
import argparse
from pathlib import Path

def floatSum(intV, floV, flOut):
    result = intV + floV
    try:
        _ = flOut.write(str(result))
    except Exception as e:
      print(f"flOut error\n{e}")

if __name__ == '__main__':
    parser = argparse.ArgumentParser(description="Float sum component")
    parser.add_argument('--input1-path', type=int, help="Integer value")
    parser.add_argument('--input2-path', type=float, help="Float value")
    parser.add_argument('--output1-path', type=str, help="Output file path")

    args = parser.parse_args()

    Path(args.output1_path).parent.mkdir(parents=True, exist_ok=True)
    with open(args.output1_path, 'w') as output1_file:
        floatSum(args.input1_path, args.input2_path, output1_file)

3. 컴포넌트 컨테이너화 및 이미지 게시

컴포넌트 코드와 종속성을 컨테이너 이미지로 패키징한 이후, 컴포넌트 이미지를 파이프라인에서 접근할 수 있는 레지스트리에 게시하여 사용할 수 있게 해준다.

4. 파이프라인 정의

# pipeline.py
import kfp
from kfp import dsl
import kfp.components as comp


create_step_float_sum = comp.load_component_from_file("interface.yaml")


@dsl.pipeline(name="float_sum_pipeline")
def my_pipeline(intV: int = 1, floV: float = 3.14159):
    float_sum_step = create_step_float_sum(intV, floV)

if __name__ == "__main__":
    kfp.compiler.Compiler().compile(
        pipeline_func=my_pipeline, package_path="pipeline.yaml"
    )

파이프라인 정의에 대해서는 크게 다른게 없이, 함수 대신 인터페이스를 사용해주면 된다.

6. 파이프라인 실행

생성된 pipeline.yaml 파일을 Dashboard에 게시하여 실행한다.
그림1 과 같이 잘 실행되는걸 확인할 수 있다.

그림1

저작자표시 (새창열림)