SDK로 파이프라인 만들기

파이프라인은 컨테이너 이미지 기반으로 실행되기 때문에 개발 환경에 도커 클라이언트가 설치되어 있어야 합니다.
또한 파이프라인 SDK도 설치되어 있어야합니다.

파이프라인은 DSL(도메인 특화 언어)을 통해서 작성되며, 컴파일 과정을 거쳐 k8s 파이프라인 리소스로 변환되어 사용됩니다.

파이프라인을 작성하는 방법 2가지

kfp.dsl 패키지 내의 ContainerOp를 이용하는 방법

먼저 ML 워크플로의 특정 스텝에서 필요한 테스크를 수행하는 애플리케이션이 작성되어 있어야 하며, 그 애플리케이션이 도커 이미지로 패키징되어 단독으로도 실행할 수 있어야 합니다.
ContainerOp는 그 도커 이미지를 매개변수로 작성됩니다.
ContainerOp의 생성자는 아래와 같습니다.

ContainerOp(
    name: str,
    image: str,
    command: String Or StringList = None,
    arguments: String Or StringList = None,
    init_containers: List[UserContainer] = None,
    sidecars: List[Sidecar] = None,
    container_kwargs: Dict = None,
    artifact_argument_path: List[InputArgumentPath] = None,
    file_outputs: Dict[str, str] = None,
    output_artifact_path: Dict[str, str] = None,
    pvolumes: Dict[str, V1Volume] = None
)

1. name : str 필수값, 컴포넌트의 이름, 중복이름일 경우에는 새로운 이름으로 생성합니다.
2. image: str 필수값, 컨테이너 이미지 이름
3. command : 컨테이너 실행 명령어
4. arguments : 컨테이너 실행 인자값
5. init_containers : 메인 컨테이너가 실행되기 전에 실행되는 컨테이너
   • dsl.InitContainer('print', 'busybox:latest', command='echo hello')
6. sidecar : 메인 컨테이너와 같이 실행되는 컨테이너
7. container_kwargs : 컨테이너 환경변수 값
   • {'env': [V1EnvVar('foo','bar')]}
8. artifact_argument_paths: 인풋파일경로가 애플리케이션에서 하드코딩되어 있을 때 사용
9. file_outputs : Dict[str(아웃풋 라벨), str(로컬 파일)], 파이프라인 실행 시점에서 PipelineParam의 값이 설정한 로컬파일에 저장됩니다. 컨테이너의 결과를 외부로 노출시키는 유일한 방법입니다.
   • {'merged': '/tmp/message.txt'}
10. output_artifact_paths : Dict[str(아티팩트 라벨), str(로컬 아티팩트 경로)]
    • {'mlpipeline-ui-metadata': '/mlpipeline-ui-metadata.json',
      'mlpipeline-metrics': '.mlpipeline-metrics.json'}
11. pvolumes : Dict[컨테이너 경로, V1Volume], 다른 파이프라인의 볼륨이나 VolumeOp로 만들어진 볼륨을, 정의한 컨테이너 경로에 마운트
    • {"/my/path": vol, "/mnt": other_op.pvolumes["/output]}

containerOp을 활용하여 파이프라인을 생성할 떄의 예제는 아래와 같습니다.
bash 컨테이너 이미지를 이용하여 콘솔창에 Hellow World라는 메세지를 찍는 작업을 수행하는 파이프라인입니다.

import kfp
from kfp import dsl

def echo_op() :
	return dsl.ContainerOp(
    	name='echo',
        image='library/bash:4.4.23',
        command=['sh', '-c'],
        arguments=['echo "Hello World"']
    )
    
@dsl.pipeline(
	name='ContainerOp pipeline',
    description='ContainerOp'
)

def hellow_world_pipeline():
    echo_task = echo_op()
    
## 쥬피터 노트북에서 사용할 경우
if __name__ == '__main__' :
    kfp.compiler.Compiler().compile(hellow_world_pipeline, 'containerop.pipeline.tar.gz')
    
## dsl-compile 툴을 이용할 경우
$ dsl-compile --py containerop.py --output containerop.pipeline.tar.gz
)

1. name='echo' : 컴포넌트의 이름
2. @dsl.pipeline : 쿠버네티스의 리소스 메타정보 데코레이션, 필수입니다.
3. compiler.Compiler().compile : 함수 명을 매개변수로 받고 파이프라인 리소스를 포함하는 containerop.pipeline.tar.gz 파일 생성

두 번째로 파이썬 함수를 파이프라인으로 변환하는 방법이 있습니다. 여기선 별도의 컨테이너 이미지가 필요없습니다.

파이썬 함수를 파이프라인으로 생성할 때의 예제는 아래와 같습니다.

import kfp.dsl as dsl

@dsl.pipeline(
    name='exampe_1',
    description='description'
)

def my_pipeline(a: int = 1, b: str = "default value"):
    print(a)
    print(b)
    
if __name__ == "__main__" :
    import kfp.compiler as compiler
    compiler.Compiler().compile(my_pipeline, 'my_pipeline.pipeline.tar.gz')

실행하면 위와 같은 파일을 생성하면서 메세지를 노출합니다.
함수의 매개변수는 파이프라인의 PipelineParam 형태로 변환되어집니다.

이제 만들어진 파일을 파이프라인 UI를 통해서 등록해보겠습니다.
대시보드의 업로드 파이프라인 버튼을 눌러 등록차응로 이동합니다.
Kubeflow 1.0 이상의 버전부터는 파이프라인을 버전별로 관리가 가능합니다.

여기에 아까 생성한 containerop.pipeline.tar.gz를 선택합니다.(1번 예제에서 dsl-compile --py containerop.py --output containerop.pipeline.tar.gz 명령을 통해 생성합니다.)
파이프라인의 이름은 압축파일명의 제일 앞부분인 containerop로 자동입력됩니다.
물론 파이프라인의 이름은 중복 불가이며 수정은 가능합니다.
Create버튼을 누르면 파이프라인이 등록됩니다.

등록이 완료되면 파이프라인 리스트에서 containerop를 확인할 수 있습니다.

containerop를 클릭하면 파이프라인 그래프를 확인할 수 있으며,
그 그래프를 클릭하면 해당 컴포넌트의 상세정보를 확인할 수 있습니다.

만들어진 파이프라인으로 런(Run)을 하나 만들어 봅시다.
여기서 테스트로 containerop라는 Experiment도 같이 만들어서 그 안에 런을 실행시켜봅시다.

Experiment 메뉴로 이동하여 Create Experiment 버튼을 눌러 생성합니다.

Experiment를 생성하면 바로 런 생성 메뉴로 넘어가게 됩니다.
이 과정을 그냥 패스할 수도 있습니다. 실행할 파이프라인을 선택하고, 런 이름을 정한 후 Start를 누릅니다.
(run-type은 그냥 One-off, Experiment는 자동으로 선택됩니다.)

이슈 해결

Run creation failed

• Dismiss 후 생성 재시도 하면 해결됩니다.

(도커이슈)This step is in Pending state with this message: ContainerCreating

• log에 'hello world'가 떠야하는데 안떠서 상태를 확인해보니 Pending excution 문제가 있습니다.
  
  

• 파드 로그를 살펴봅시다! 이벤트에 자세한 내용이 나와있군요.(Reason: FailedMount)
  - kubectl describe pod containerop-pipeline-nhm5j-3333120606 -n moey920
  

  • 이 문제를 해결하면 되겠군요. MountVolume.SetUp failed for volume "docker-sock" : hostPath type check failed: /var/run/docker.sock is not a socket file

1. 도커가 인터넷에 연결할 수 있는지 확인합니다.
   docker pull tutum/hello-world
   
   - 도커에 문제가 있네요.
   - Cannot connect to the Docker daemon at unix:///var/run/docker.sock. Is the docker daemon running?
   - 현재 백그라운드에 Docker(도커)가 실행되고있지 않기 때문입니다. 도커 데스크탑을 실행한 다음에 명령어를 입력해 정상 작동하는지 확인해주세요!
   
   
   
2. 그래도 Pod가 ContainerCreating 상태에 빠져있다면 다시 FailedMount를 힌트로 해결 방법을 찾아봅시다.
   • kubectl get pods containerop-pipeline-nhm5j-3333120606 -n moey920 -o yaml
     
   • spec: serviceAccount가 default-editor로 자동 할당되었습니다.(문제 없는 것으로 보입니다.)
   • 아무래도 쿠버네티스 1.2부터 도커를 지원하지 않는게 문제인 것 같습니다. 아래에 해결 방법을 정리하겠습니다.

(해결) Argo Workflow Executor 선택

0. Kubeflow Pipelines를 설치한 네임스페스로 전환합니다.

   • kubectl config set-context --current --namespace moey920
     

1. 현재 워크플로 실행자를 확인합니다.

   • kubectl describe configmap workflow-controller-configmap -n kubeflow | grep -A 2 containerRuntimeExecutor
     

2. 워크플로 실행자를 PNS로 구성합니다.

   • kubectl patch configmap workflow-controller-configmap -n kubeflow --patch '{"data":{"containerRuntimeExecutor":"pns"}}'
     

3. 워크플로 실행자가 정상적으로 바뀌었는지 확인합니다.

   • kubectl describe configmap workflow-controller-configmap -n kubeflow | grep -A 2 containerRuntimeExecutor
     

4. 수동으로도 확인하고 deployment를 다시 로드해봅시다.

   • kubectl edit configmap/workflow-controller-configmap -n kubeflow
     
   • kubectl rollout restart deploy workflow-controller -n kubeflow