[Kubeflow] 모델 학습부터 서빙까지 파이프라인 구축하기

from kfp.components import create_component_from_func,func_to_container_op
@create_component_from_func
def test_component():
	pass
 
@func_to_container_op
def test_container_op():
	pass
 
print(type(test_component()))
print(type(test_container_op()))
 
>>> <class 'kfp.components._structures.TaskSpec'>
>>> <class 'kfp.components._structures.TaskSpec'>
 

from functools import partial
from kfp.components import create_component_from_func
 
 
@partial(
    create_component_from_func,
    packages_to_install=["pandas"],
)
def load_data():
 
    import pandas as pd
    import os
 
    # 경로 확인 용
    print("list_dir : \n ", os.listdir())
 
    # load data from github
    df_train = pd.read_csv(
        "https://raw.github.com/yangoos57/Learning_kubeflow/main/mini_project/data/train.csv"
    )
    df_evaluation = pd.read_csv(
        "https://raw.github.com/yangoos57/Learning_kubeflow/main/mini_project/data/validation.csv"
    )
 
    df_train.to_csv("pvc/train.csv", index=False)
    df_evaluation.to_csv("pvc/evaluation.csv", index=False)
 
    print("complete Loading Data")
 

@partial(create_component_from_func, base_image="679oose/python_huggingface")
def train_model():
 
    from transformers import (
        DistilBertForSequenceClassification,
        DistilBertTokenizer,
        Trainer,
        TrainingArguments,
        TrainerCallback,
    )
 
    from datasets import Dataset
 
    import os
 
    # 경로 확인 용
    print("list_dir : \n ", os.listdir())
    print("list_dir : \n ", os.getcwd())
    os.chdir("/")
 
    train_dataset = Dataset.from_csv("pvc/train.csv").select(range(32))
    evaluation_dataset = Dataset.from_csv("pvc/evaluation.csv")
 
    # tokenizing
    tokenizer = DistilBertTokenizer.from_pretrained("distilbert-base-uncased")
 
    def tokenize_function(item):
        return tokenizer(item["text"], padding="max_length", max_length=128, truncation=True)
 
    train = train_dataset.map(tokenize_function)
    evaluation = evaluation_dataset.map(tokenize_function)
 
    print("complete Tokenizing")
 
    model = DistilBertForSequenceClassification.from_pretrained(
        "distilbert-base-uncased", num_labels=len(set(train_dataset["label"]))
    )
 
    tra_arg = TrainingArguments(
        per_device_train_batch_size=8,
        output_dir="test",
        num_train_epochs=1,
        logging_steps=2,
        # evaluation_strategy="epoch",
        disable_tqdm=True,
        save_strategy="no",
    )
 
    class myCallback(TrainerCallback):
        def on_log(self, args, state, control, logs=None, **kwargs):
            print(f"{state.global_step} Steps ")
 
    trainer = Trainer(
        model=model,
        args=tra_arg,
        train_dataset=train,
        eval_dataset=evaluation,
        callbacks=[myCallback],
    )
 
    trainer.train()
 
    # Saving Tokenizer, Model
    trainer.save_model("pvc/torch_model")
    tokenizer.save_pretrained("pvc/torch_model")
 
    print("Saving Model & Tokenizer Complete !!")
 
    # config for torchserve
    import json
    config = dict(
        inference_address="http://0.0.0.0:8085",
        management_address="http://0.0.0.0:8085",
        metrics_address="http://0.0.0.0:8082",
        grpc_inference_port=7070,
        grpc_management_port=7071,
        enable_envvars_config="true",
        install_py_dep_per_model="true",
        model_store="model-store",
        model_snapshot=json.dumps({
            "name": "startup.cfg",
            "modelCount": 1,
            "models": {
                "torch-model": {  # Model Name
                    "1.0": {
                        "defaultVersion": "true",
                        "marName": "torch-model.mar",
                        "minWorkers": 1,
                        "maxWorkers": 5,
                        "batchSize": 1,
                        "maxBatchDelay": 10,
                        "responseTimeout": 60,
                    }
                }
            },
        }),
    )
 
    # creating config & config folder
    if not os.path.exists("pvc/torch_model/config"):
        os.mkdir("pvc/torch_model/config")
 
    with open("pvc/torch_model/config/config.properties", "w") as f:
        for i, j in config.items():
            f.write(f"{i}={j}\n")
        f.close()
 
    print("Saving config.properties !!")
 
    # handler for torchserve
    x = '''
from abc import ABC
import json
import logging
import os
 
import torch
from transformers import AutoModelForSequenceClassification, AutoTokenizer
from ts.torch_handler.base_handler import BaseHandler
 
logger = logging.getLogger(__name__)
 
 
class TransformersClassifierHandler(BaseHandler, ABC):
    def __init__(self):
        super(TransformersClassifierHandler, self).__init__()
        self.initialized = False
 
    def initialize(self, ctx):
        self.manifest = ctx.manifest
 
        properties = ctx.system_properties
        model_dir = properties.get("model_dir")
        self.device = torch.device(
            "cuda:" + str(properties.get("gpu_id")) if torch.cuda.is_available() else "cpu"
        )
 
        # Read model serialize/pt file
        self.model = AutoModelForSequenceClassification.from_pretrained(model_dir)
        self.tokenizer = AutoTokenizer.from_pretrained(model_dir)
 
        self.model.to(self.device)
        # BetterTransformer
        self.model.eval()
 
        logger.debug(f"Transformer model from path {model_dir} loaded successfully")
 
        # Read the mapping file, index to object name
        mapping_file_path = os.path.join(model_dir, "index_to_name.json")
 
        if os.path.isfile(mapping_file_path):
            with open(mapping_file_path) as f:
                self.mapping = json.load(f)
        else:
            logger.warning(
                "Missing the index_to_name.json file. Inference output will not include class name."
            )
 
        self.initialized = True
 
    def preprocess(self, data):
        """Very basic preprocessing code - only tokenizes.
        Extend with your own preprocessing steps as needed.
        """
        print("------- input data --------")
        print(data)
        text = data[0].get("data")
        if text is None:
            text = data[0].get("body")
 
        logger.info(f"Received text: {text}")
 
        inputs = self.tokenizer.encode_plus(text, add_special_tokens=True, return_tensors="pt")
        return inputs
 
    def inference(self, inputs):
        """
        Predict the class of a text using a trained transformer model.
        """
        # NOTE: This makes the assumption that your model expects text to be tokenized
        # with "input_ids" and "token_type_ids" - which is true for some popular transformer models, e.g. bert.
        # If your transformer model expects different tokenization, adapt this code to suit
        # its expected input format.
        inputs = inputs.to(self.device)
 
        prediction = self.model(**inputs)[0].argmax().item()
        logger.info(f"Model predicted: {prediction}")
 
        if self.mapping:
            prediction = self.mapping[str(prediction)]
        return [prediction]
 
    def postprocess(self, inference_output):
        # TODO: Add any needed post-processing of the model predictions here
        logger.info(f"Model Name: {self.model.config._name_or_path}")
        logger.info(f"Model predicted: {inference_output}")
        return inference_output
 
 
_service = TransformersClassifierHandler()
 
 
def handle(data, context):
    try:
        if not _service.initialized:
            _service.initialize(context)
 
        if data is None:
            return None
 
        data = _service.preprocess(data)
        data = _service.inference(data)
        data = _service.postprocess(data)
 
        return data
    except Exception as e:
        raise e
 
 
    '''
    with open("pvc/torch_model/handler.py", "w") as f:
        f.write(x)
    f.close()
 
    print("Saving handler.py complete !!")
 

 
from kfp.dsl import ContainerOp
 
 
def create_marfile():
    return ContainerOp(
        name="Creating Marfile",
        command=["/bin/sh"],
        image="python:3.9",
        arguments=[
            "-c",
            "cd pvc/torch_model; pip install torchserve torch-model-archiver torch-workflow-archiver; torch-model-archiver --model-name torch-model --version 1.0 --serialized-file pytorch_model.bin --handler handler.py --extra-files config.json,vocab.txt --force; mkdir model-store; mv -f torch-model.mar model-store"
        ],  # pip install => create mar file => make model_store folder => mv marfile to model_store
    )
 

from kfp.components import load_component_from_url
 
def create_inference_model():
    kserve_op = load_component_from_url('https://raw.githubusercontent.com/kubeflow/pipelines/'
                                               'master/components/kserve/component.yaml')
 
    model_name = "torchserve"
    namespace = "kubeflow-user-example-com"
    model_uri = "pvc://leeway/torch_model"
    framework="pytorch"
 
    return kserve_op(action="apply",
              model_name=model_name,
              model_uri=model_uri,
              namespace=namespace,
              framework=framework)

from kfp.dsl import ContainerOp, pipeline
from kfp import onprem
 
 
@pipeline(name="NLP_Pipeline")
def NLP_Pipeline():
    ### 데이터 로드
    data = load_data()
    data.apply(onprem.mount_pvc(pvc_name="leeway", volume_name="test-lee", volume_mount_path="pvc"))
 
 
    ### 모델 학습
    model = train_model()
    model.apply(onprem.mount_pvc(pvc_name="leeway", volume_name="test-lee", volume_mount_path="pvc"))
    model.set_cpu_limit(cpu="1").set_memory_limit(memory="2G")
    model.set_display_name("Finetuning Text Classification Model")
    model.after(data)
 
 
    ### Mar file 생성
    marfile = create_marfile()
    marfile.apply(onprem.mount_pvc(pvc_name="leeway", volume_name="test-lee", volume_mount_path="pvc"))
    marfile.set_display_name("Creating Marfile")
    marfile.execution_options.caching_strategy.max_cache_staleness = "P0D" # cache 사용않는 옶션
    marfile.after(model)
 
 
    ### 모델 서빙
    inference_model = create_inference_model()
    inference_model.apply(onprem.mount_pvc(pvc_name="leeway", volume_name="test-lee", volume_mount_path="pvc"))
    inference_model.after(marfile)

 
import re
import requests
from urllib.parse import urlsplit
import os
 
# https 사용 시 인증서 등록
os.environ["cert_for_kubeflow"] = "/Users/git_repo/Learning_kubeflow/manifests/deployment/cert/leeway.crt"
 
 
 
def get_istio_auth_session(url: str, username: str, password: str) -> dict:
    """
    Determine if the specified URL is secured by Dex and try to obtain a session cookie.
    WARNING: only Dex `staticPasswords` and `LDAP` authentication are currently supported
             (we default default to using `staticPasswords` if both are enabled)
 
    :param url: Kubeflow server URL, including protocol
    :param username: Dex `staticPasswords` or `LDAP` username
    :param password: Dex `staticPasswords` or `LDAP` password
    :return: auth session information
    """
    # define the default return object
    auth_session = {
        "endpoint_url": url,  # KF endpoint URL
        "redirect_url": None,  # KF redirect URL, if applicable
        "dex_login_url": None,  # Dex login URL (for POST of credentials)
        "is_secured": None,  # True if KF endpoint is secured
        "session_cookie": None,  # Resulting session cookies in the form "key1=value1; key2=value2"
    }
 
    # use a persistent session (for cookies)
    with requests.Session() as s:
 
        ################
        # Determine if Endpoint is Secured
        ################
        resp = s.get(url, allow_redirects=True, verify=os.getenv("cert_for_kubeflow"))
        if resp.status_code != 200:
            raise RuntimeError(f"HTTP status code '{resp.status_code}' for GET against: {url}")
 
        auth_session["redirect_url"] = resp.url
 
        # if we were NOT redirected, then the endpoint is UNSECURED
        if len(resp.history) == 0:
            auth_session["is_secured"] = False
            return auth_session
        else:
            auth_session["is_secured"] = True
 
        ################
        # Get Dex Login URL
        ################
        redirect_url_obj = urlsplit(auth_session["redirect_url"])
 
        # if we are at `/auth?=xxxx` path, we need to select an auth type
        if re.search(r"/auth$", redirect_url_obj.path):
 
            #######
            # TIP: choose the default auth type by including ONE of the following
            #######
 
            # OPTION 1: set "staticPasswords" as default auth type
            redirect_url_obj = redirect_url_obj._replace(
                path=re.sub(r"/auth$", "/auth/local", redirect_url_obj.path)
            )
            # OPTION 2: set "ldap" as default auth type
            # redirect_url_obj = redirect_url_obj._replace(
            #     path=re.sub(r"/auth$", "/auth/ldap", redirect_url_obj.path)
            # )
 
        # if we are at `/auth/xxxx/login` path, then no further action is needed (we can use it for login POST)
        if re.search(r"/auth/.*/login$", redirect_url_obj.path):
            auth_session["dex_login_url"] = redirect_url_obj.geturl()
 
        # else, we need to be redirected to the actual login page
        else:
            # this GET should redirect us to the `/auth/xxxx/login` path
            resp = s.get(redirect_url_obj.geturl(), allow_redirects=True)
            if resp.status_code != 200:
                raise RuntimeError(
                    f"HTTP status code '{resp.status_code}' for GET against: {redirect_url_obj.geturl()}"
                )
 
            # set the login url
            auth_session["dex_login_url"] = resp.url
 
        ################
        # Attempt Dex Login
        ################
        resp = s.post(
            auth_session["dex_login_url"],
            data={"login": username, "password": password},
            allow_redirects=True,
        )
        if len(resp.history) == 0:
            raise RuntimeError(
                f"Login credentials were probably invalid - "
                f"No redirect after POST to: {auth_session['dex_login_url']}"
            )
 
        # store the session cookies in a "key1=value1; key2=value2" string
        auth_session["session_cookie"] = "; ".join([f"{c.name}={c.value}" for c in s.cookies])
    return auth_session
 
 
import kfp
import os
 
KUBEFLOW_ENDPOINT = "https://localhost:8080"
KUBEFLOW_USERNAME = "user@example.com"
KUBEFLOW_PASSWORD = "12341234"
 
 
auth_session = get_istio_auth_session(
    url=KUBEFLOW_ENDPOINT, username=KUBEFLOW_USERNAME, password=KUBEFLOW_PASSWORD
)
 
client = kfp.Client(
    host=f"{KUBEFLOW_ENDPOINT}/pipeline",
    cookies=auth_session["session_cookie"],
    ssl_ca_cert=os.getenv("cert_for_kubeflow"),
)
 
if __name__ == "__main__":
    client.create_run_from_pipeline_func(
        NLP_Pipeline, arguments={}, namespace="kubeflow-user-example-com"
    )
 

torchserve_name = "torch-model"
model_name = "torchserve"
url = f"https://localhost:8080/v1/models/{torchserve_name}:predict"
host = f"{model_name}.kubeflow-user-example-com.example.com"
session={'authservice_session':auth_session["session_cookie"].replace('authservice_session=','')}
data = {"instances": [{"data": "Hello World!"}]}  # data
 
x = requests.post(
    url=url, verify=False, cookies=session, headers={"Host": host}, json=data
)
x.text