MLPE 스터디 3주차 - Ch7.1 ~ 7.2 Kubeflow 노트북과 KFP 데이터 전달

apiVersion: kubeflow.org/v1alpha1
kind: PodDefault
metadata:
  name: add-ml-env
  namespace: kubeflow
spec:
  desc: add MLflow endpoint
  env:
    - name: MLFLOW_TRACKING_URI
      value: http://mlflow.kubeflow.svc.cluster.local:5000
  selector:
    matchLabels:
      add-ml-env: "true"

kubectl get crd -o=name | grep kubeflow.org
# customresourcedefinition.apiextensions.k8s.io/notebooks.kubeflow.org

kubectl get cm -n kubeflow | grep jupyter-web-app-config

spawner_ui_config.yaml: |
  spawnerFormDefaults:
    image:                      # Jupyter
      value: kubeflownotebookswg/jupyter-scipy:v1.6.1
      options:
        - asia.gcr.io/kubeflow-notebook-servers/jupyter-tensorflow:2.6.0
        - asia.gcr.io/kubeflow-notebook-servers/jupyter-pytorch:1.9.0
    imageGroupOne:              # code-server
      - kubeflownotebookswg/codeserver-python:v1.6.1
    imageGroupTwo:              # RStudio
      - kubeflownotebookswg/rstudio-tidyverse:v1.6.1

resources:
  - https://github.com/kubeflow/manifests/applications/notebooks-v1/upstream/notebook-controller/overlays/kubeflow?ref=26.03.1

configMapGenerator:
  - name: config
    behavior: merge
    literals:
      - USE_ISTIO=false
      - ISTIO_GATEWAY=
      - ENABLE_CULLING=true
      - CULL_IDLE_TIME=120         # 분
      - IDLENESS_CHECK_PERIOD=5

patches:
  - target: {kind: Deployment, name: notebook-controller-deployment}
    patch: |-
      - op: add
        path: /spec/template/spec/nodeSelector
        value: {kubernetes.io/arch: amd64}

resources:
  - https://github.com/kubeflow/manifests/applications/dashboard/upstream/poddefaults-webhooks/overlays/cert-manager?ref=26.03.1

patches:
  - target: {kind: Deployment, name: poddefaults-webhook-deployment}
    patch: |-
      - op: add
        path: /spec/template/spec/nodeSelector
        value: {kubernetes.io/arch: amd64}

kubectl apply --server-side --force-conflicts -k 01-notebook-controller/
kubectl apply --server-side --force-conflicts -k 02-poddefaults-webhook/
kubectl -n kubeflow wait --for=condition=Ready certificate/poddefaults-webhook-cert --timeout=2m

apiVersion: kubeflow.org/v1alpha1
kind: PodDefault
metadata:
  name: add-ml-env
  namespace: kubeflow-user
spec:
  desc: MLflow, KFP 주소
  selector:
    matchLabels:
      add-ml-env: "true"
  env:
    - name: MLFLOW_TRACKING_URI
      value: http://mlflow.mlflow.svc.cluster.local:5000
    - name: KF_PIPELINES_ENDPOINT
      value: http://ml-pipeline.kubeflow.svc.cluster.local:8888
---
apiVersion: kubeflow.org/v1alpha1
kind: PodDefault
metadata:
  name: add-rustfs-creds
  namespace: kubeflow-user
spec:
  desc: RustFS(S3) 주소와 키
  selector:
    matchLabels:
      add-rustfs-creds: "true"
  env:
    - name: MLFLOW_S3_ENDPOINT_URL
      value: http://rustfs.mlflow.svc.cluster.local:9000
    - name: AWS_ENDPOINT_URL
      value: http://rustfs.mlflow.svc.cluster.local:9000
    - name: AWS_ACCESS_KEY_ID
      valueFrom:
        secretKeyRef: {name: rustfs-credentials, key: AWS_ACCESS_KEY_ID}
    - name: AWS_SECRET_ACCESS_KEY
      valueFrom:
        secretKeyRef: {name: rustfs-credentials, key: AWS_SECRET_ACCESS_KEY}

apiVersion: v1
kind: PersistentVolumeClaim
metadata:
  name: test-nb-workspace
  namespace: kubeflow-user
spec:
  accessModes: ["ReadWriteOnce"]
  storageClassName: local-path-worker1
  resources:
    requests:
      storage: 5Gi
---
apiVersion: kubeflow.org/v1
kind: Notebook
metadata:
  name: test-nb
  namespace: kubeflow-user
  labels:
    add-ml-env: "true"
    add-rustfs-creds: "true"
spec:
  template:
    spec:
      nodeSelector:
        kubernetes.io/arch: amd64
      containers:
        - name: test-nb
          image: ghcr.io/kubeflow/kubeflow/notebook-servers/jupyter-scipy:v1.11.0
          resources:
            requests: {cpu: "500m", memory: 1Gi}
            limits: {cpu: "2", memory: 4Gi}
          volumeMounts:
            - name: workspace
              mountPath: /home/jovyan     # 워크스페이스 볼륨
      volumes:
        - name: workspace
          persistentVolumeClaim:
            claimName: test-nb-workspace

kubectl -n kubeflow-user get notebook,sts,pod,pvc
# notebook.kubeflow.org/test-nb
# statefulset.apps/test-nb                   1/1
# pod/test-nb-0                              1/1   Running   worker1-gpu
# persistentvolumeclaim/test-nb-workspace    Bound   5Gi   RWO   local-path-worker1

kubectl -n kubeflow-user get pod test-nb-0 -o jsonpath='{.metadata.annotations}'
# poddefault.admission.kubeflow.org/poddefault-add-ml-env: "26998020"
# poddefault.admission.kubeflow.org/poddefault-add-rustfs-creds: "26998021"

kubectl -n kubeflow-user port-forward svc/test-nb 18888:80
# 브라우저: http://localhost:18888/notebook/kubeflow-user/test-nb/lab

import kfp.dsl as dsl

@dsl.component(packages_to_install=["pyjokes"], base_image="python:3.11")
def generate_joke() -> str:
    import pyjokes                    # import 는 함수 안에서
    return pyjokes.get_joke()

@dsl.component
def count_words(input: str) -> int:
    return len(input.split())

@dsl.component
def output_result(count: int) -> str:
    return f"Word count: {count}"

@dsl.pipeline(name="joke-pipeline-values", description="simple pipeline demonstrating value passing")
def pipeline():
    generate_joke_op = generate_joke()
    count_word_op = count_words(input=generate_joke_op.output)     # 반환값을 다음 컴포넌트로
    output_result_op = output_result(count=count_word_op.output)

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

exec-count-words:
  container:
    args: [--executor_input, '{{$}}', --function_to_execute, count_words]
    command:
      - sh
      - -c
      - |                                        # ① pip 확인 후 kfp 설치
        if ! [ -x "$(command -v pip)" ]; then python3 -m ensurepip || ...; fi
        PIP_DISABLE_PIP_VERSION_CHECK=1 python3 -m pip install --quiet \
          --no-warn-script-location 'kfp==2.17.3' '--no-deps' \
          'typing-extensions>=3.7.4,<5; python_version<"3.9"' && "$0" "$@"
      - sh
      - -ec
      - |                                        # ② 임시 파일로 쓰고 실행자로 실행
        program_path=$(mktemp -d)
        printf "%s" "$0" > "$program_path/ephemeral_component.py"
        _KFP_RUNTIME=true python3 -m kfp.dsl.executor_main \
          --component_module_path "$program_path/ephemeral_component.py" "$@"
      - |                                        # ③ 주입된 함수 코드
        import kfp
        from kfp import dsl
        from kfp.dsl import *
        from typing import *

        def count_words(input: str) -> int:
            return len(input.split())
    image: python:3.11

kubectl -n kubeflow exec deploy/mysql -- mysql -uroot -N -e \
  'select t.name, substring_index(a.uri,"/",-1) from metadb.Artifact a
   join metadb.Type t on t.id=a.type_id where a.uri like "%56263271%"'
# system.Artifact               raw
# system.ClassificationMetrics  cls_metrics
# system.Dataset                dataset
# system.HTML                   report_html
# system.Markdown               report_md
# system.Metrics                metrics
# system.Model                  model

export KFP_HOST=http://192.168.0.242
python a1_values_pipeline.py && python submit.py a1_values_pipeline.yaml hl-a1
python submit.py a2_artifact_pipeline.yaml hl-a2-42 num_jokes=42
python submit.py a2b_by_value.yaml hl-val-100k size_kb=100
python submit.py a2b_by_artifact.yaml hl-art-200000k size_kb=200000   # 200MB
python submit.py a4_artifact_types.yaml hl-a4