Untitled1.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "cd17a049",
   "metadata": {},
   "outputs": [],
   "source": [
    "def load_and_clean_data():\n",
    "    \n",
    "    import pandas as pd\n",
    "    import numpy as np\n",
    "    \n",
    "    data = pd.read_csv(\"https://raw.githubusercontent.com/TripathiAshutosh/dataset/main/banking.csv\")\n",
    "    \n",
    "    print(\"Null/missingalues available in the data: \\n\")\n",
    "    print(data.isna().sum())\n",
    "    data = data.dropna()\n",
    "    print(\"The data after dropping the na values are: \\n\")\n",
    "    print(data.isna().sum())\n",
    "    \n",
    "    print(\"--------data imported and cleaned----------\")\n",
    "\n",
    "    return data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "efc7ff8e",
   "metadata": {},
   "outputs": [],
   "source": [
    "def preprocessing(data):\n",
    "    \n",
    "    import pandas as pd\n",
    "    import numpy as np\n",
    "    \n",
    "    data = load_and_clean_data()\n",
    "\n",
    "    data['education'] = np.where(data['education'] == 'basic.9y', 'Basic', data['education'])\n",
    "    data['education'] = np.where(data['education'] == 'basic.6y', 'Basic', data['education'])\n",
    "    data['education'] = np.where(data['education'] == 'basic.4y', 'Basic', data['education'])\n",
    "    \n",
    "    categorical_vars = ['job','marital','education', 'default', 'housing', 'loan', 'contact', 'month', 'day_of_week', 'poutcome']\n",
    "    for var in categorical_vars:\n",
    "        cat_list = 'var' + '_' + var\n",
    "        cat_list = pd.get_dummies(data[var], prefix = var) # one hot encoding\n",
    "        data_new = data.join(cat_list)\n",
    "        data = data_new\n",
    "    \n",
    "    categorical_vars = ['job','marital','education', 'default', 'housing', 'loan', 'contact', 'month', 'day_of_week', 'poutcome']\n",
    "    \n",
    "    data_vars = data.columns.values.tolist()\n",
    "    \n",
    "    keeping = [i for i in data_vars if i not in categorical_vars]\n",
    "    \n",
    "    final_df = data[keeping]\n",
    "    \n",
    "    final_df.columns = final_df.columns.str.replace(\".\", \"_\")\n",
    "    final_df.columns = final_df.columns.str.replace(\" \", \"_\")\n",
    "    \n",
    "    print(final_df.head())\n",
    "    \n",
    "    print(\"Education column pre-processed, categorical variables one-hot encoded. Ready to input data to model\")\n",
    "    \n",
    "    return final_df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "bec53974",
   "metadata": {},
   "outputs": [
    {
     "ename": "NameError",
     "evalue": "name 'pd' is not defined",
     "output_type": "error",
     "traceback": [
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[1;31mNameError\u001b[0m                                 Traceback (most recent call last)",
      "Cell \u001b[1;32mIn[3], line 1\u001b[0m\n\u001b[1;32m----> 1\u001b[0m \u001b[38;5;28;01mdef\u001b[39;00m \u001b[38;5;21mtrain_test_split\u001b[39m(final_df: \u001b[43mpd\u001b[49m\u001b[38;5;241m.\u001b[39mDataFrame):\n\u001b[0;32m      3\u001b[0m     \u001b[38;5;28;01mimport\u001b[39;00m \u001b[38;5;21;01mpandas\u001b[39;00m \u001b[38;5;28;01mas\u001b[39;00m \u001b[38;5;21;01mpd\u001b[39;00m\n\u001b[0;32m      4\u001b[0m     \u001b[38;5;28;01mimport\u001b[39;00m \u001b[38;5;21;01mnumpy\u001b[39;00m \u001b[38;5;28;01mas\u001b[39;00m \u001b[38;5;21;01mnp\u001b[39;00m\n",
      "\u001b[1;31mNameError\u001b[0m: name 'pd' is not defined"
     ]
    }
   ],
   "source": [
    "\n",
    "def train_test_split(final_df: pd.DataFrame):\n",
    "\n",
    "    import pandas as pd\n",
    "    import numpy as np\n",
    "    from sklearn.model_selection import train_test_split\n",
    "\n",
    "    #final_df = preprocessing()\n",
    "    \n",
    "    X = final_df.loc[:, final_df.columns != 'y']\n",
    "    y = final_df.loc[:, final_df.columns == 'y']\n",
    "    \n",
    "    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size = 0.3, stratify = y, random_state = 47)\n",
    "    \n",
    "    # np.save(f'data/X_train.npy', X_train) # saved as a numpy binary file (efficient to save and load)\n",
    "    # np.save(f'data/X_test.npy', X_test)\n",
    "    # np.save(f'data/y_train.npy', y_train)\n",
    "    # np.save(f'data/y_test.npy', y_test)\n",
    "    \n",
    "    print(\"\\n---- X_train ----\")\n",
    "    print(\"\\n\")\n",
    "    print(X_train.head())\n",
    "    \n",
    "    print(\"\\n---- X_test ----\")\n",
    "    print(\"\\n\")\n",
    "    print(X_test.head())\n",
    "    \n",
    "    print(\"\\n---- y_test ----\")\n",
    "    print(\"\\n\")\n",
    "    print(y_test.head())\n",
    "\n",
    "    return X_train, X_test, y_train, y_test"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "86878033",
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "def training_basic_classifier(X_train,y_train):\n",
    "    import mlflow\n",
    "    from sklearn.ensemble import RandomForestClassifier\n",
    "    import pandas as pd\n",
    "    import numpy as np\n",
    "    import pickle\n",
    "    \n",
    "    #X_train, X_test, y_train, y_test = train_test_split()\n",
    "\n",
    "    model = RandomForestClassifier(n_estimators=150)\n",
    "    model.fit(X_train, y_train)\n",
    "\n",
    "    #mlflow.set_tracking_uri(\"http://localhost:8000\")\n",
    "\n",
    "    with mlflow.start_run():\n",
    "        mlflow.log_param(\"n_estimators\", 150)\n",
    "        # Log any other hyperparameters you want to track\n",
    "        \n",
    "        mlflow.sklearn.log_model(model, \"model\")\n",
    "        \n",
    "        with open(f'data/model.pkl', 'wb') as f:\n",
    "            pickle.dump(model, f)\n",
    "        \n",
    "    print(\"\\nRandomForest classifier is trained on banking data and saved to PV location /data/model.pkl ----\")\n",
    "    return model\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "d0c2bb9e",
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "def predict_on_test_data(model,X_test):\n",
    "    import pandas as pd\n",
    "    import numpy as np\n",
    "    import sklearn\n",
    "\n",
    "    #model = training_basic_classifier()\n",
    "\n",
    "    print(\"---- Inside predict_on_test_data component ----\")\n",
    "    \n",
    "    X_test = np.load('data/X_test.npy', allow_pickle=True)\n",
    "    y_pred = model.predict(X_test)\n",
    "    np.save('data/y_pred.npy', y_pred)\n",
    "\n",
    "    print(\"\\n---- Predicted classes ----\")\n",
    "    print(\"\\n\")\n",
    "    print(y_pred)\n",
    "\n",
    "    return y_pred"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "14723f4d",
   "metadata": {},
   "outputs": [],
   "source": [
    "import kfp.dsl as dsl\n",
    "import mlflow\n",
    "\n",
    "def get_mlflow_server_url():\n",
    "    # Replace 'your_mlflow_service_name' with the name of the MLflow service in your Kubernetes cluster\n",
    "    mlflow_service_name = 'localhost:5000'\n",
    "\n",
    "    # Retrieve the MLflow server URL using the Kubernetes API\n",
    "    service_url = mlflow.get_service_url(mlflow_service_name)\n",
    "    return service_url\n",
    "\n",
    "\n",
    "@dsl.component\n",
    "def load_data_component():\n",
    "    return dsl.ContainerOp(\n",
    "        name='load-data',\n",
    "        image='python:3.8',\n",
    "        command=['python', 'load_data_script.py'],\n",
    "        output_artifact_paths={'data': '/data'},\n",
    "    )\n",
    "\n",
    "@dsl.component\n",
    "def preprocess_data_component(data):\n",
    "    return dsl.ContainerOp(\n",
    "        name='preprocess-data',\n",
    "        image='python:3.8',\n",
    "        command=['python', 'preprocessing_script.py'],\n",
    "        arguments=['--data', data],\n",
    "        output_artifact_paths={'preprocessed_data': '/data'},\n",
    "    )\n",
    "\n",
    "@dsl.component\n",
    "def split_data_component(final_df):\n",
    "    return dsl.ContainerOp(\n",
    "        name='split-data',\n",
    "        image='python:3.8',\n",
    "        command=['python', 'data_split_script.py'],\n",
    "        arguments=['--final_df', final_df],\n",
    "        output_artifact_paths={\n",
    "            'X_train': '/data/X_train.npy',\n",
    "            'X_test': '/data/X_test.npy',\n",
    "            'y_train': '/data/y_train.npy',\n",
    "            'y_test': '/data/y_test.npy',\n",
    "        },\n",
    "    )\n",
    "\n",
    "@dsl.component\n",
    "def train_classifier_component(X_train, y_train):\n",
    "    return dsl.ContainerOp(\n",
    "        name='train-classifier',\n",
    "        image='python:3.8',\n",
    "        command=['python', 'model_building_script.py'],\n",
    "        arguments=[\n",
    "            '--X_train', X_train,\n",
    "            '--y_train', y_train\n",
    "        ],\n",
    "        output_artifact_paths={'model': '/data/model.pkl'},\n",
    "    )\n",
    "\n",
    "@dsl.component\n",
    "def predict_test_data_component(model, X_test):\n",
    "    return dsl.ContainerOp(\n",
    "        name='predict-test-data',\n",
    "        image='python:3.8',\n",
    "        command=['python', 'prediction_script.py'],\n",
    "        arguments=[\n",
    "            '--model', model,\n",
    "            '--X_test', X_test\n",
    "        ],\n",
    "        output_artifact_paths={'y_pred': '/data/y_pred.npy'},\n",
    "    )\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "1512fc15",
   "metadata": {},
   "outputs": [],
   "source": [
    "import kfp.dsl as dsl\n",
    "@dsl.pipeline(name='ML Pipeline', description='A pipeline for ML model training and prediction')\n",
    "def ml_pipeline():\n",
    "    # Load data\n",
    "    load_data_task = load_data_component()\n",
    "\n",
    "    # Preprocess data\n",
    "    preprocess_task = preprocess_data_component(data=load_data_task.output)\n",
    "\n",
    "    # Split data\n",
    "    split_data_task = split_data_component(preprocessed_data=preprocess_task.output)\n",
    "\n",
    "    # Train classifier with MLflow tracking\n",
    "    train_classifier_task = train_classifier_component(\n",
    "        X_train=split_data_task.outputs['X_train'], \n",
    "        y_train=split_data_task.outputs['y_train']\n",
    "    )\n",
    "\n",
    "    predictions_task = predict_test_data_component(model = train_classifier_task.output, X_test=split_data_task.outputs['X_test'])\n",
    "\n",
    "    # Set the MLflow server URL\n",
    "    mlflow_server_url = get_mlflow_server_url()\n",
    "\n",
    "    # Log the MLflow server URL to the pipeline output\n",
    "    dsl.ContainerOp(\n",
    "        name='log-mlflow-server-url',\n",
    "        image='python:3.8',\n",
    "        command=['python', '-c', f'print(\"{mlflow_server_url}\")']\n",
    "    )\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "a96b3d7d",
   "metadata": {},
   "outputs": [
    {
     "ename": "ValueError",
     "evalue": "I/O operation on closed file",
     "output_type": "error",
     "traceback": [
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[1;31mValueError\u001b[0m                                Traceback (most recent call last)",
      "Cell \u001b[1;32mIn[8], line 13\u001b[0m\n\u001b[0;32m     11\u001b[0m \u001b[38;5;66;03m# Compile the pipeline\u001b[39;00m\n\u001b[0;32m     12\u001b[0m pipeline_filename \u001b[38;5;241m=\u001b[39m \u001b[38;5;124m'\u001b[39m\u001b[38;5;124mml_pipeline.yaml\u001b[39m\u001b[38;5;124m'\u001b[39m\n\u001b[1;32m---> 13\u001b[0m \u001b[43mkfp\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mcompiler\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mCompiler\u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mcompile\u001b[49m\u001b[43m(\u001b[49m\u001b[43mml_pipeline\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43mpipeline_filename\u001b[49m\u001b[43m)\u001b[49m\n\u001b[0;32m     15\u001b[0m \u001b[38;5;66;03m# Create and run the pipeline\u001b[39;00m\n\u001b[0;32m     16\u001b[0m client\u001b[38;5;241m.\u001b[39mcreate_run_from_pipeline_func(ml_pipeline, experiment_name\u001b[38;5;241m=\u001b[39mexperiment_name, run_name\u001b[38;5;241m=\u001b[39mrun_name)\n",
      "File \u001b[1;32m~\\anaconda3\\envs\\mlops\\lib\\site-packages\\kfp\\compiler\\compiler.py:1175\u001b[0m, in \u001b[0;36mCompiler.compile\u001b[1;34m(self, pipeline_func, package_path, type_check, pipeline_conf)\u001b[0m\n\u001b[0;32m   1173\u001b[0m \u001b[38;5;28;01mtry\u001b[39;00m:\n\u001b[0;32m   1174\u001b[0m     kfp\u001b[38;5;241m.\u001b[39mTYPE_CHECK \u001b[38;5;241m=\u001b[39m type_check\n\u001b[1;32m-> 1175\u001b[0m     \u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43m_create_and_write_workflow\u001b[49m\u001b[43m(\u001b[49m\n\u001b[0;32m   1176\u001b[0m \u001b[43m        \u001b[49m\u001b[43mpipeline_func\u001b[49m\u001b[38;5;241;43m=\u001b[39;49m\u001b[43mpipeline_func\u001b[49m\u001b[43m,\u001b[49m\n\u001b[0;32m   1177\u001b[0m \u001b[43m        \u001b[49m\u001b[43mpipeline_conf\u001b[49m\u001b[38;5;241;43m=\u001b[39;49m\u001b[43mpipeline_conf\u001b[49m\u001b[43m,\u001b[49m\n\u001b[0;32m   1178\u001b[0m \u001b[43m        \u001b[49m\u001b[43mpackage_path\u001b[49m\u001b[38;5;241;43m=\u001b[39;49m\u001b[43mpackage_path\u001b[49m\u001b[43m)\u001b[49m\n\u001b[0;32m   1179\u001b[0m \u001b[38;5;28;01mfinally\u001b[39;00m:\n\u001b[0;32m   1180\u001b[0m     kfp\u001b[38;5;241m.\u001b[39mTYPE_CHECK \u001b[38;5;241m=\u001b[39m type_check_old_value\n",
      "File \u001b[1;32m~\\anaconda3\\envs\\mlops\\lib\\site-packages\\kfp\\compiler\\compiler.py:1227\u001b[0m, in \u001b[0;36mCompiler._create_and_write_workflow\u001b[1;34m(self, pipeline_func, pipeline_name, pipeline_description, params_list, pipeline_conf, package_path)\u001b[0m\n\u001b[0;32m   1218\u001b[0m \u001b[38;5;28;01mdef\u001b[39;00m \u001b[38;5;21m_create_and_write_workflow\u001b[39m(\u001b[38;5;28mself\u001b[39m,\n\u001b[0;32m   1219\u001b[0m                                pipeline_func: Callable,\n\u001b[0;32m   1220\u001b[0m                                pipeline_name: Text \u001b[38;5;241m=\u001b[39m \u001b[38;5;28;01mNone\u001b[39;00m,\n\u001b[1;32m   (...)\u001b[0m\n\u001b[0;32m   1223\u001b[0m                                pipeline_conf: dsl\u001b[38;5;241m.\u001b[39mPipelineConf \u001b[38;5;241m=\u001b[39m \u001b[38;5;28;01mNone\u001b[39;00m,\n\u001b[0;32m   1224\u001b[0m                                package_path: Text \u001b[38;5;241m=\u001b[39m \u001b[38;5;28;01mNone\u001b[39;00m) \u001b[38;5;241m-\u001b[39m\u001b[38;5;241m>\u001b[39m \u001b[38;5;28;01mNone\u001b[39;00m:\n\u001b[0;32m   1225\u001b[0m \u001b[38;5;250m    \u001b[39m\u001b[38;5;124;03m\"\"\"Compile the given pipeline function and dump it to specified file\u001b[39;00m\n\u001b[0;32m   1226\u001b[0m \u001b[38;5;124;03m    format.\"\"\"\u001b[39;00m\n\u001b[1;32m-> 1227\u001b[0m     workflow \u001b[38;5;241m=\u001b[39m \u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43m_create_workflow\u001b[49m\u001b[43m(\u001b[49m\u001b[43mpipeline_func\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43mpipeline_name\u001b[49m\u001b[43m,\u001b[49m\n\u001b[0;32m   1228\u001b[0m \u001b[43m                                     \u001b[49m\u001b[43mpipeline_description\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43mparams_list\u001b[49m\u001b[43m,\u001b[49m\n\u001b[0;32m   1229\u001b[0m \u001b[43m                                     \u001b[49m\u001b[43mpipeline_conf\u001b[49m\u001b[43m)\u001b[49m\n\u001b[0;32m   1230\u001b[0m     \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_write_workflow(workflow, package_path)\n\u001b[0;32m   1231\u001b[0m     _validate_workflow(workflow)\n",
      "File \u001b[1;32m~\\anaconda3\\envs\\mlops\\lib\\site-packages\\kfp\\compiler\\compiler.py:1005\u001b[0m, in \u001b[0;36mCompiler._create_workflow\u001b[1;34m(self, pipeline_func, pipeline_name, pipeline_description, params_list, pipeline_conf)\u001b[0m\n\u001b[0;32m   1002\u001b[0m         args_list\u001b[38;5;241m.\u001b[39mappend(param)\n\u001b[0;32m   1004\u001b[0m \u001b[38;5;28;01mwith\u001b[39;00m dsl\u001b[38;5;241m.\u001b[39mPipeline(pipeline_name) \u001b[38;5;28;01mas\u001b[39;00m dsl_pipeline:\n\u001b[1;32m-> 1005\u001b[0m     pipeline_func(\u001b[38;5;241m*\u001b[39margs_list, \u001b[38;5;241m*\u001b[39m\u001b[38;5;241m*\u001b[39mkwargs_dict)\n\u001b[0;32m   1007\u001b[0m pipeline_conf \u001b[38;5;241m=\u001b[39m pipeline_conf \u001b[38;5;129;01mor\u001b[39;00m dsl_pipeline\u001b[38;5;241m.\u001b[39mconf  \u001b[38;5;66;03m# Configuration passed to the compiler is overriding. Unfortunately, it's not trivial to detect whether the dsl_pipeline.conf was ever modified.\u001b[39;00m\n\u001b[0;32m   1009\u001b[0m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_validate_exit_handler(dsl_pipeline)\n",
      "Cell \u001b[1;32mIn[6], line 5\u001b[0m, in \u001b[0;36mml_pipeline\u001b[1;34m()\u001b[0m\n\u001b[0;32m      2\u001b[0m \u001b[38;5;129m@dsl\u001b[39m\u001b[38;5;241m.\u001b[39mpipeline(name\u001b[38;5;241m=\u001b[39m\u001b[38;5;124m'\u001b[39m\u001b[38;5;124mML Pipeline\u001b[39m\u001b[38;5;124m'\u001b[39m, description\u001b[38;5;241m=\u001b[39m\u001b[38;5;124m'\u001b[39m\u001b[38;5;124mA pipeline for ML model training and prediction\u001b[39m\u001b[38;5;124m'\u001b[39m)\n\u001b[0;32m      3\u001b[0m \u001b[38;5;28;01mdef\u001b[39;00m \u001b[38;5;21mml_pipeline\u001b[39m():\n\u001b[0;32m      4\u001b[0m     \u001b[38;5;66;03m# Load data\u001b[39;00m\n\u001b[1;32m----> 5\u001b[0m     load_data_task \u001b[38;5;241m=\u001b[39m \u001b[43mload_data_component\u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m\n\u001b[0;32m      7\u001b[0m     \u001b[38;5;66;03m# Preprocess data\u001b[39;00m\n\u001b[0;32m      8\u001b[0m     preprocess_task \u001b[38;5;241m=\u001b[39m preprocess_data_component(data\u001b[38;5;241m=\u001b[39mload_data_task\u001b[38;5;241m.\u001b[39moutput)\n",
      "File \u001b[1;32m~\\anaconda3\\envs\\mlops\\lib\\site-packages\\kfp\\dsl\\_component.py:117\u001b[0m, in \u001b[0;36mcomponent.<locals>._component\u001b[1;34m(*args, **kargs)\u001b[0m\n\u001b[0;32m    108\u001b[0m                     \u001b[38;5;28;01mif\u001b[39;00m input_spec\u001b[38;5;241m.\u001b[39mname \u001b[38;5;241m==\u001b[39m key \u001b[38;5;129;01mand\u001b[39;00m \u001b[38;5;129;01mnot\u001b[39;00m check_types(\n\u001b[0;32m    109\u001b[0m                             kargs[key]\u001b[38;5;241m.\u001b[39mparam_type, input_spec\u001b[38;5;241m.\u001b[39mtype):\n\u001b[0;32m    110\u001b[0m                         \u001b[38;5;28;01mraise\u001b[39;00m InconsistentTypeException(\n\u001b[0;32m    111\u001b[0m                             \u001b[38;5;124m'\u001b[39m\u001b[38;5;124mComponent \u001b[39m\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124m'\u001b[39m \u001b[38;5;241m+\u001b[39m component_meta\u001b[38;5;241m.\u001b[39mname \u001b[38;5;241m+\u001b[39m\n\u001b[0;32m    112\u001b[0m                             \u001b[38;5;124m'\u001b[39m\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124m is expecting \u001b[39m\u001b[38;5;124m'\u001b[39m \u001b[38;5;241m+\u001b[39m input_spec\u001b[38;5;241m.\u001b[39mname \u001b[38;5;241m+\u001b[39m\n\u001b[0;32m    113\u001b[0m                             \u001b[38;5;124m'\u001b[39m\u001b[38;5;124m to be type(\u001b[39m\u001b[38;5;124m'\u001b[39m \u001b[38;5;241m+\u001b[39m \u001b[38;5;28mstr\u001b[39m(input_spec\u001b[38;5;241m.\u001b[39mtype) \u001b[38;5;241m+\u001b[39m\n\u001b[0;32m    114\u001b[0m                             \u001b[38;5;124m'\u001b[39m\u001b[38;5;124m), but the passed argument is type(\u001b[39m\u001b[38;5;124m'\u001b[39m \u001b[38;5;241m+\u001b[39m\n\u001b[0;32m    115\u001b[0m                             \u001b[38;5;28mstr\u001b[39m(kargs[key]\u001b[38;5;241m.\u001b[39mparam_type) \u001b[38;5;241m+\u001b[39m \u001b[38;5;124m'\u001b[39m\u001b[38;5;124m)\u001b[39m\u001b[38;5;124m'\u001b[39m)\n\u001b[1;32m--> 117\u001b[0m container_op \u001b[38;5;241m=\u001b[39m func(\u001b[38;5;241m*\u001b[39margs, \u001b[38;5;241m*\u001b[39m\u001b[38;5;241m*\u001b[39mkargs)\n\u001b[0;32m    118\u001b[0m container_op\u001b[38;5;241m.\u001b[39m_set_metadata(component_meta)\n\u001b[0;32m    119\u001b[0m \u001b[38;5;28;01mreturn\u001b[39;00m container_op\n",
      "Cell \u001b[1;32mIn[5], line 15\u001b[0m, in \u001b[0;36mload_data_component\u001b[1;34m()\u001b[0m\n\u001b[0;32m     13\u001b[0m \u001b[38;5;129m@dsl\u001b[39m\u001b[38;5;241m.\u001b[39mcomponent\n\u001b[0;32m     14\u001b[0m \u001b[38;5;28;01mdef\u001b[39;00m \u001b[38;5;21mload_data_component\u001b[39m():\n\u001b[1;32m---> 15\u001b[0m     \u001b[38;5;28;01mreturn\u001b[39;00m \u001b[43mdsl\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mContainerOp\u001b[49m\u001b[43m(\u001b[49m\n\u001b[0;32m     16\u001b[0m \u001b[43m        \u001b[49m\u001b[43mname\u001b[49m\u001b[38;5;241;43m=\u001b[39;49m\u001b[38;5;124;43m'\u001b[39;49m\u001b[38;5;124;43mload-data\u001b[39;49m\u001b[38;5;124;43m'\u001b[39;49m\u001b[43m,\u001b[49m\n\u001b[0;32m     17\u001b[0m \u001b[43m        \u001b[49m\u001b[43mimage\u001b[49m\u001b[38;5;241;43m=\u001b[39;49m\u001b[38;5;124;43m'\u001b[39;49m\u001b[38;5;124;43mpython:3.8\u001b[39;49m\u001b[38;5;124;43m'\u001b[39;49m\u001b[43m,\u001b[49m\n\u001b[0;32m     18\u001b[0m \u001b[43m        \u001b[49m\u001b[43mcommand\u001b[49m\u001b[38;5;241;43m=\u001b[39;49m\u001b[43m[\u001b[49m\u001b[38;5;124;43m'\u001b[39;49m\u001b[38;5;124;43mpython\u001b[39;49m\u001b[38;5;124;43m'\u001b[39;49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[38;5;124;43m'\u001b[39;49m\u001b[38;5;124;43mload_data_script.py\u001b[39;49m\u001b[38;5;124;43m'\u001b[39;49m\u001b[43m]\u001b[49m\u001b[43m,\u001b[49m\n\u001b[0;32m     19\u001b[0m \u001b[43m        \u001b[49m\u001b[43moutput_artifact_paths\u001b[49m\u001b[38;5;241;43m=\u001b[39;49m\u001b[43m{\u001b[49m\u001b[38;5;124;43m'\u001b[39;49m\u001b[38;5;124;43mdata\u001b[39;49m\u001b[38;5;124;43m'\u001b[39;49m\u001b[43m:\u001b[49m\u001b[43m \u001b[49m\u001b[38;5;124;43m'\u001b[39;49m\u001b[38;5;124;43m/data\u001b[39;49m\u001b[38;5;124;43m'\u001b[39;49m\u001b[43m}\u001b[49m\u001b[43m,\u001b[49m\n\u001b[0;32m     20\u001b[0m \u001b[43m    \u001b[49m\u001b[43m)\u001b[49m\n",
      "File \u001b[1;32m~\\anaconda3\\envs\\mlops\\lib\\site-packages\\kfp\\dsl\\_container_op.py:1261\u001b[0m, in \u001b[0;36mContainerOp.__init__\u001b[1;34m(self, name, image, command, arguments, init_containers, sidecars, container_kwargs, artifact_argument_paths, file_outputs, output_artifact_paths, is_exit_handler, pvolumes)\u001b[0m\n\u001b[0;32m   1256\u001b[0m arguments \u001b[38;5;241m=\u001b[39m as_string_list(arguments)\n\u001b[0;32m   1258\u001b[0m \u001b[38;5;28;01mif\u001b[39;00m (\u001b[38;5;129;01mnot\u001b[39;00m ContainerOp\u001b[38;5;241m.\u001b[39m_DISABLE_REUSABLE_COMPONENT_WARNING) \u001b[38;5;129;01mand\u001b[39;00m (\n\u001b[0;32m   1259\u001b[0m         \u001b[38;5;124m'\u001b[39m\u001b[38;5;124m--component_launcher_class_path\u001b[39m\u001b[38;5;124m'\u001b[39m \u001b[38;5;129;01mnot\u001b[39;00m \u001b[38;5;129;01min\u001b[39;00m (arguments \u001b[38;5;129;01mor\u001b[39;00m [])):\n\u001b[0;32m   1260\u001b[0m     \u001b[38;5;66;03m# The warning is suppressed for pipelines created using the TFX SDK.\u001b[39;00m\n\u001b[1;32m-> 1261\u001b[0m     \u001b[43mwarnings\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mwarn\u001b[49m\u001b[43m(\u001b[49m\n\u001b[0;32m   1262\u001b[0m \u001b[43m        \u001b[49m\u001b[38;5;124;43m'\u001b[39;49m\u001b[38;5;124;43mPlease create reusable components instead of constructing ContainerOp instances directly.\u001b[39;49m\u001b[38;5;124;43m'\u001b[39;49m\n\u001b[0;32m   1263\u001b[0m \u001b[43m        \u001b[49m\u001b[38;5;124;43m'\u001b[39;49m\u001b[38;5;124;43m Reusable components are shareable, portable and have compatibility and support guarantees.\u001b[39;49m\u001b[38;5;124;43m'\u001b[39;49m\n\u001b[0;32m   1264\u001b[0m \u001b[43m        \u001b[49m\u001b[38;5;124;43m'\u001b[39;49m\u001b[38;5;124;43m Please see the documentation: https://www.kubeflow.org/docs/pipelines/sdk/component-development/#writing-your-component-definition-file\u001b[39;49m\u001b[38;5;124;43m'\u001b[39;49m\n\u001b[0;32m   1265\u001b[0m \u001b[43m        \u001b[49m\u001b[38;5;124;43m'\u001b[39;49m\u001b[38;5;124;43m The components can be created manually (or, in case of python, using kfp.components.create_component_from_func or func_to_container_op)\u001b[39;49m\u001b[38;5;124;43m'\u001b[39;49m\n\u001b[0;32m   1266\u001b[0m \u001b[43m        \u001b[49m\u001b[38;5;124;43m'\u001b[39;49m\u001b[38;5;124;43m and then loaded using kfp.components.load_component_from_file, load_component_from_uri or load_component_from_text: \u001b[39;49m\u001b[38;5;124;43m'\u001b[39;49m\n\u001b[0;32m   1267\u001b[0m \u001b[43m        \u001b[49m\u001b[38;5;124;43m'\u001b[39;49m\u001b[38;5;124;43mhttps://kubeflow-pipelines.readthedocs.io/en/stable/source/kfp.components.html#kfp.components.load_component_from_file\u001b[39;49m\u001b[38;5;124;43m'\u001b[39;49m\u001b[43m,\u001b[49m\n\u001b[0;32m   1268\u001b[0m \u001b[43m        \u001b[49m\u001b[43mcategory\u001b[49m\u001b[38;5;241;43m=\u001b[39;49m\u001b[38;5;167;43;01mFutureWarning\u001b[39;49;00m\u001b[43m,\u001b[49m\n\u001b[0;32m   1269\u001b[0m \u001b[43m    \u001b[49m\u001b[43m)\u001b[49m\n\u001b[0;32m   1270\u001b[0m     \u001b[38;5;28;01mif\u001b[39;00m kfp\u001b[38;5;241m.\u001b[39mCOMPILING_FOR_V2:\n\u001b[0;32m   1271\u001b[0m         \u001b[38;5;28;01mraise\u001b[39;00m \u001b[38;5;167;01mRuntimeError\u001b[39;00m(\n\u001b[0;32m   1272\u001b[0m             \u001b[38;5;124m'\u001b[39m\u001b[38;5;124mConstructing ContainerOp instances directly is deprecated and not \u001b[39m\u001b[38;5;124m'\u001b[39m\n\u001b[0;32m   1273\u001b[0m             \u001b[38;5;124m'\u001b[39m\u001b[38;5;124msupported when compiling to v2 (using v2 compiler or v1 compiler \u001b[39m\u001b[38;5;124m'\u001b[39m\n\u001b[0;32m   1274\u001b[0m             \u001b[38;5;124m'\u001b[39m\u001b[38;5;124mwith V2_COMPATIBLE or V2_ENGINE mode).\u001b[39m\u001b[38;5;124m'\u001b[39m)\n",
      "File \u001b[1;32m~\\anaconda3\\envs\\mlops\\lib\\warnings.py:112\u001b[0m, in \u001b[0;36m_showwarnmsg\u001b[1;34m(msg)\u001b[0m\n\u001b[0;32m    109\u001b[0m         sw(msg\u001b[38;5;241m.\u001b[39mmessage, msg\u001b[38;5;241m.\u001b[39mcategory, msg\u001b[38;5;241m.\u001b[39mfilename, msg\u001b[38;5;241m.\u001b[39mlineno,\n\u001b[0;32m    110\u001b[0m            msg\u001b[38;5;241m.\u001b[39mfile, msg\u001b[38;5;241m.\u001b[39mline)\n\u001b[0;32m    111\u001b[0m         \u001b[38;5;28;01mreturn\u001b[39;00m\n\u001b[1;32m--> 112\u001b[0m \u001b[43m_showwarnmsg_impl\u001b[49m\u001b[43m(\u001b[49m\u001b[43mmsg\u001b[49m\u001b[43m)\u001b[49m\n",
      "File \u001b[1;32m~\\anaconda3\\envs\\mlops\\lib\\warnings.py:30\u001b[0m, in \u001b[0;36m_showwarnmsg_impl\u001b[1;34m(msg)\u001b[0m\n\u001b[0;32m     28\u001b[0m text \u001b[38;5;241m=\u001b[39m _formatwarnmsg(msg)\n\u001b[0;32m     29\u001b[0m \u001b[38;5;28;01mtry\u001b[39;00m:\n\u001b[1;32m---> 30\u001b[0m     \u001b[43mfile\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mwrite\u001b[49m\u001b[43m(\u001b[49m\u001b[43mtext\u001b[49m\u001b[43m)\u001b[49m\n\u001b[0;32m     31\u001b[0m \u001b[38;5;28;01mexcept\u001b[39;00m \u001b[38;5;167;01mOSError\u001b[39;00m:\n\u001b[0;32m     32\u001b[0m     \u001b[38;5;66;03m# the file (probably stderr) is invalid - this warning gets lost.\u001b[39;00m\n\u001b[0;32m     33\u001b[0m     \u001b[38;5;28;01mpass\u001b[39;00m\n",
      "File \u001b[1;32m~\\anaconda3\\envs\\mlops\\lib\\site-packages\\colorama\\ansitowin32.py:47\u001b[0m, in \u001b[0;36mStreamWrapper.write\u001b[1;34m(self, text)\u001b[0m\n\u001b[0;32m     46\u001b[0m \u001b[38;5;28;01mdef\u001b[39;00m \u001b[38;5;21mwrite\u001b[39m(\u001b[38;5;28mself\u001b[39m, text):\n\u001b[1;32m---> 47\u001b[0m     \u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43m__convertor\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mwrite\u001b[49m\u001b[43m(\u001b[49m\u001b[43mtext\u001b[49m\u001b[43m)\u001b[49m\n",
      "File \u001b[1;32m~\\anaconda3\\envs\\mlops\\lib\\site-packages\\colorama\\ansitowin32.py:177\u001b[0m, in \u001b[0;36mAnsiToWin32.write\u001b[1;34m(self, text)\u001b[0m\n\u001b[0;32m    175\u001b[0m \u001b[38;5;28;01mdef\u001b[39;00m \u001b[38;5;21mwrite\u001b[39m(\u001b[38;5;28mself\u001b[39m, text):\n\u001b[0;32m    176\u001b[0m     \u001b[38;5;28;01mif\u001b[39;00m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mstrip \u001b[38;5;129;01mor\u001b[39;00m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mconvert:\n\u001b[1;32m--> 177\u001b[0m         \u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mwrite_and_convert\u001b[49m\u001b[43m(\u001b[49m\u001b[43mtext\u001b[49m\u001b[43m)\u001b[49m\n\u001b[0;32m    178\u001b[0m     \u001b[38;5;28;01melse\u001b[39;00m:\n\u001b[0;32m    179\u001b[0m         \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mwrapped\u001b[38;5;241m.\u001b[39mwrite(text)\n",
      "File \u001b[1;32m~\\anaconda3\\envs\\mlops\\lib\\site-packages\\colorama\\ansitowin32.py:205\u001b[0m, in \u001b[0;36mAnsiToWin32.write_and_convert\u001b[1;34m(self, text)\u001b[0m\n\u001b[0;32m    203\u001b[0m     \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mconvert_ansi(\u001b[38;5;241m*\u001b[39mmatch\u001b[38;5;241m.\u001b[39mgroups())\n\u001b[0;32m    204\u001b[0m     cursor \u001b[38;5;241m=\u001b[39m end\n\u001b[1;32m--> 205\u001b[0m \u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mwrite_plain_text\u001b[49m\u001b[43m(\u001b[49m\u001b[43mtext\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43mcursor\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[38;5;28;43mlen\u001b[39;49m\u001b[43m(\u001b[49m\u001b[43mtext\u001b[49m\u001b[43m)\u001b[49m\u001b[43m)\u001b[49m\n",
      "File \u001b[1;32m~\\anaconda3\\envs\\mlops\\lib\\site-packages\\colorama\\ansitowin32.py:210\u001b[0m, in \u001b[0;36mAnsiToWin32.write_plain_text\u001b[1;34m(self, text, start, end)\u001b[0m\n\u001b[0;32m    208\u001b[0m \u001b[38;5;28;01mdef\u001b[39;00m \u001b[38;5;21mwrite_plain_text\u001b[39m(\u001b[38;5;28mself\u001b[39m, text, start, end):\n\u001b[0;32m    209\u001b[0m     \u001b[38;5;28;01mif\u001b[39;00m start \u001b[38;5;241m<\u001b[39m end:\n\u001b[1;32m--> 210\u001b[0m         \u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mwrapped\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mwrite\u001b[49m\u001b[43m(\u001b[49m\u001b[43mtext\u001b[49m\u001b[43m[\u001b[49m\u001b[43mstart\u001b[49m\u001b[43m:\u001b[49m\u001b[43mend\u001b[49m\u001b[43m]\u001b[49m\u001b[43m)\u001b[49m\n\u001b[0;32m    211\u001b[0m         \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mwrapped\u001b[38;5;241m.\u001b[39mflush()\n",
      "File \u001b[1;32m~\\anaconda3\\envs\\mlops\\lib\\site-packages\\ipykernel\\iostream.py:547\u001b[0m, in \u001b[0;36mOutStream.write\u001b[1;34m(self, string)\u001b[0m\n\u001b[0;32m    544\u001b[0m             \u001b[38;5;28mprint\u001b[39m(\u001b[38;5;124mf\u001b[39m\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mWrite failed: \u001b[39m\u001b[38;5;132;01m{\u001b[39;00me\u001b[38;5;132;01m}\u001b[39;00m\u001b[38;5;124m\"\u001b[39m, file\u001b[38;5;241m=\u001b[39msys\u001b[38;5;241m.\u001b[39m__stderr__)\n\u001b[0;32m    546\u001b[0m \u001b[38;5;28;01mif\u001b[39;00m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mpub_thread \u001b[38;5;129;01mis\u001b[39;00m \u001b[38;5;28;01mNone\u001b[39;00m:\n\u001b[1;32m--> 547\u001b[0m     \u001b[38;5;28;01mraise\u001b[39;00m \u001b[38;5;167;01mValueError\u001b[39;00m(\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mI/O operation on closed file\u001b[39m\u001b[38;5;124m\"\u001b[39m)\n\u001b[0;32m    548\u001b[0m \u001b[38;5;28;01melse\u001b[39;00m:\n\u001b[0;32m    550\u001b[0m     is_child \u001b[38;5;241m=\u001b[39m \u001b[38;5;129;01mnot\u001b[39;00m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_is_master_process()\n",
      "\u001b[1;31mValueError\u001b[0m: I/O operation on closed file"
     ]
    }
   ],
   "source": [
    "import kfp\n",
    "\n",
    "\n",
    "# Set your Kubeflow experiment and run names\n",
    "experiment_name = 'my-ml-experiment'\n",
    "run_name = 'my-ml-run'\n",
    "\n",
    "# Initialize the Kubeflow client\n",
    "client = kfp.Client()\n",
    "\n",
    "# Compile the pipeline\n",
    "pipeline_filename = 'ml_pipeline.yaml'\n",
    "kfp.compiler.Compiler().compile(ml_pipeline, pipeline_filename)\n",
    "\n",
    "# Create and run the pipeline\n",
    "client.create_run_from_pipeline_func(ml_pipeline, experiment_name=experiment_name, run_name=run_name)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e41577c0",
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "39bfb37e",
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d600b46b",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "mlops",
   "language": "python",
   "name": "mlops"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.0"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}