{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "\n# Wrapping PySpark Models\n\nThis example requires full licence, and the program will break if you use the trial licence.\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "Installation\n\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "collapsed": false
      },
      "outputs": [],
      "source": [
        "# To install the required package, use the following command:\n# !pip install modeva"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "Authentication\n\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "collapsed": false
      },
      "outputs": [],
      "source": [
        "# To get authentication, use the following command: (To get full access please replace the token to your own token)\n# from modeva.utils.authenticate import authenticate\n# authenticate(auth_code='eaaa4301-b140-484c-8e93-f9f633c8bacb')"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "Import required modules\n\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "collapsed": false
      },
      "outputs": [],
      "source": [
        "import numpy as np\nimport pandas as pd\nfrom pyspark.sql import SparkSession\nfrom pyspark.ml.classification import LogisticRegression\nfrom pyspark.ml.feature import VectorAssembler\nfrom pyspark.sql.functions import monotonically_increasing_id\nfrom sklearn.datasets import load_breast_cancer\nfrom modeva import DataSet\nfrom modeva import TestSuite\nfrom modeva.models.wrappers.api import modeva_arbitrary_classifier"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Scripts for building a pyspark model\n\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "collapsed": false
      },
      "outputs": [],
      "source": [
        "# Initialize Spark session\nspark = SparkSession.builder.appName(\"PySpark-Wrapper-Example\").getOrCreate()\n\n# Load and prepare dataset\ndata = load_breast_cancer()\ndf = pd.DataFrame(data.data, columns=data.feature_names)\ndf['label'] = data.target\n\n# Convert Pandas DataFrame to Spark DataFrame and add an index column\nspark_df = spark.createDataFrame(df)\nspark_df = spark_df.withColumn(\"index\", monotonically_increasing_id())\n\n# Assemble features into a vector\nassembler = VectorAssembler(inputCols=data.feature_names, outputCol=\"features\")\nspark_df = assembler.transform(spark_df).select(\"features\", \"label\", \"index\")\n\n# Split the data into train and test sets\ntrain, test = spark_df.randomSplit([0.8, 0.2], seed=1234)\n\n# Extract the index for each split\ntrain_indices = np.array([row[\"index\"] for row in train.select(\"index\").collect()])\ntest_indices = np.array([row[\"index\"] for row in test.select(\"index\").collect()])"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "Train model\n\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "collapsed": false
      },
      "outputs": [],
      "source": [
        "lr = LogisticRegression(featuresCol=\"features\", labelCol=\"label\")\nlr_model = lr.fit(train)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Wrap the data\n\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "collapsed": false
      },
      "outputs": [],
      "source": [
        "ds = DataSet()\nds.load_dataframe(data=df)\nds.set_train_idx(train_indices)\nds.set_test_idx(test_indices)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Wrap the PySpark model into Modeva\n\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "collapsed": false
      },
      "outputs": [],
      "source": [
        "def predict_func(X):\n    X_spark = spark.createDataFrame(X, schema=data.feature_names.tolist())\n    X_spark = assembler.transform(X_spark)\n    predictions = lr_model.transform(X_spark).select(\"prediction\")\n    return np.array([int(row.prediction) for row in predictions.collect()])\n\ndef predict_proba_func(X):\n    X_spark = spark.createDataFrame(X, schema=data.feature_names.tolist())\n    X_spark = assembler.transform(X_spark)\n    probabilities = lr_model.transform(X_spark).select(\"probability\")\n    return np.array([row.probability.toArray() for row in probabilities.collect()])\n\nmodel = modeva_arbitrary_classifier(\n    name=\"PySpark-LogisticRegression\",\n    predict_function=predict_func,\n    predict_proba_function=predict_proba_func\n)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Create test suite for diagnostics\n\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "collapsed": false
      },
      "outputs": [],
      "source": [
        "ts = TestSuite(ds, model)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "Accuracy table\n\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "collapsed": false
      },
      "outputs": [],
      "source": [
        "results = ts.diagnose_accuracy_table()\nresults.table"
      ]
    }
  ],
  "metadata": {
    "kernelspec": {
      "display_name": "Python 3",
      "language": "python",
      "name": "python3"
    },
    "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.11.11"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 0
}