training_ensemble.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Ensemble methods"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In this notebook, we are going to build the following estimators from the training set:\n",
    "- Random forest regressor (<i>Sklearn</i>)\n",
    "- Gradient-boosted trees (<i>Apache Spark</i>)\n",
    "\n",
    "After training the ensemble models will be assessed on an evaluation set."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's start with importing all indispensable libraries."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "import numpy as np\n",
    "import pickle\n",
    "import h5py\n",
    "import joblib\n",
    "from sklearn.model_selection import train_test_split"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we will load the normalization parameters, training set, and evaluation and test sets."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "with open(r\"norm_params.pickle\", \"rb\") as output_file:\n",
    "    norm_params = pickle.load(output_file)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "h5f = h5py.File('training.hdf5', 'r')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "eval_test_df = pd.read_csv('round2_competition_data/eval_test/eval_test.csv', nrows=2e+6)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To limit the size of the evaluation and test sets we will take a random sample of <i>eval_test</i> dataframe."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Take a random sample of eval_test dataframe\n",
    "eval_test_df = eval_test_df.sample(n=500000)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Next, we will split this sample into the evaluation and test sets."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "eval_df, test_df = train_test_split(eval_test_df, test_size=0.4)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We have to separate the independent variables from the dependent variables and perform the normalization."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "eval_x = np.array(eval_df[norm_params['input_features']])\n",
    "eval_y = np.array(eval_df[norm_params['target']])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "x_min = np.array([norm_params[col]['min'] for col in norm_params['input_features']])\n",
    "x_max = np.array([norm_params[col]['max'] for col in norm_params['input_features']])\n",
    "\n",
    "y_min = np.array([norm_params[col]['min'] for col in norm_params['target']])\n",
    "y_max = np.array([norm_params[col]['max'] for col in norm_params['target']])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "normalize = True\n",
    "\n",
    "# Normalize the data\n",
    "if normalize:\n",
    "    eval_x = (eval_x - x_min) / (x_max - x_min)\n",
    "    eval_y = (eval_y - y_min) / (y_max - y_min)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Below we will define a function that converts WSG84 coordinates to cartesian ones."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "def lla_to_ecef(df):\n",
    "    \"\"\"Converts WSG84 coordinates to cartesian ones.\n",
    "    \n",
    "    \"\"\"\n",
    "    \n",
    "    # Inverse the normalization\n",
    "    if normalize:\n",
    "        df = df * (y_max - y_min) + y_min\n",
    "    \n",
    "    latitude = np.radians(df[:, 0])\n",
    "    longitude = np.radians(df[:, 1])\n",
    "    altitude = df[:, 2]\n",
    "\n",
    "    # WSG84 ellipsoid constants\n",
    "    a = 6378137\n",
    "    e = 8.1819190842622e-2\n",
    "\n",
    "    # Prime vertical radius of curvature\n",
    "    N = a / np.sqrt(1 - e**2 * np.sin(latitude)**2)\n",
    "    \n",
    "    x = (N + altitude) * np.cos(latitude) * np.cos(longitude)\n",
    "    y = (N + altitude) * np.cos(latitude) * np.sin(longitude)\n",
    "    z = ((1 - e**2) * N + altitude) * np.sin(latitude)\n",
    "\n",
    "    df = np.hstack([np.expand_dims(x, axis=1), np.expand_dims(y, axis=1), np.expand_dims(z, axis=1)])\n",
    "    \n",
    "    return df"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### RandomForestRegressor"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In this section, we are going to build the RandomForestRegressor from the training set. \n",
    "\n",
    "Because we are dealing with huge amounts of data, we will have to train the ensemble on sequential small batches. This can be achieved by setting the <i>warm_start</i> parameter to True so that we will be able to fit and add more estimators to the ensemble, where each tree is trained on a separate batch of data.\n",
    "\n",
    "More information about this ensemble method can be found [here](https://scikit-learn.org/stable/modules/generated/sklearn.ensemble.RandomForestRegressor.html#sklearn.ensemble.RandomForestRegressor)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sklearn.ensemble import RandomForestRegressor"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we have to specify the batch size or the number of the estimators. In the first case, the dataset will be split based on the batch size into a number of batches, where each batch corresponds to a single estimator. When we specify the <i>n_estimators</i> parameter then a number of batches will be deducted from the number of trees. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Specify batch_size or n_estimators\n",
    "batch_size = None\n",
    "n_estimators = 100\n",
    "\n",
    "h5f_length = h5f['input']['table'].len()\n",
    "\n",
    "if not batch_size:\n",
    "    indices = np.linspace(0, h5f_length, n_estimators)\n",
    "    indices = [int(idx) for idx in indices]\n",
    "    indices = list(zip(indices[:-1], indices[1:]))\n",
    "else:\n",
    "    n_batches = int(h5f_length/batch_size)\n",
    "    indices = [[i*batch_size,(i+1)*batch_size] for i in range(n_batches+1)]\n",
    "    indices[-1][-1] = h5f_length\n",
    "\n",
    "# Instantiate the ensemble model\n",
    "est = RandomForestRegressor(warm_start=True, n_estimators=1)\n",
    "\n",
    "for idxs in indices:\n",
    "    start_idx, end_idx = idxs\n",
    "\n",
    "    # fetch the input and target data\n",
    "    x = h5f['input']['table'][start_idx:end_idx]['values_block_0']\n",
    "    y = h5f['target']['table'][start_idx:end_idx]['values_block_0']\n",
    "    \n",
    "    if normalize:    \n",
    "        x = (x - x_min) / (x_max - x_min)\n",
    "        y = (y - y_min) / (y_max - y_min)\n",
    "    \n",
    "    # Fit the batch of the data into the model\n",
    "    est.fit(x, y)\n",
    "    est.n_estimators += 1 # add 1 tree\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Coefficient of determination (R^2): 0.98791\n"
     ]
    }
   ],
   "source": [
    "print('Coefficient of determination (R^2): {:.5f}'.format(est.score(eval_x, eval_y)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The coefficient of determination is close to 1, which denotes that the output values are explained very well by the determining (independent) variables."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Next, we will predict the regression targets for the evaluation set and calculate the mean distance error."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "pred = est.predict(eval_x)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "pred_ecef = lla_to_ecef(pred)\n",
    "target_ecef = lla_to_ecef(eval_y)\n",
    "\n",
    "# Calculate the average prediciton - target distance error in kilometers\n",
    "dist_error = np.sqrt((pred_ecef[:, 0] - target_ecef[:, 0])**2 + (pred_ecef[:, 1] - target_ecef[:, 1])**2 + \\\n",
    "        (pred_ecef[:, 2] - target_ecef[:, 2])**2) / 1000\n",
    "\n",
    "dist_error = np.mean(np.abs(dist_error))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Mean distance error: 71.783 km\n"
     ]
    }
   ],
   "source": [
    "print('Mean distance error: {:.3f} km'.format(dist_error))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we can save the model for inference."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['sklearn_random_forest']"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "joblib.dump(est, 'sklearn_random_forest')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "###  Apache Spark GradientBoostedTrees"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The second model that will be used is the GradientBoostedTrees ensemble from the <i>Apache Spark</i> framework. Regarding <i>Apache Spark 3.0.0</i>, the aforementioned ensemble of decision trees doesn't support multi-target regression, thus to circumvent that restriction we will create a separate GradientBoostedTrees for each target variable."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We will load all the data using <i>Spark</i>, therefore we can restart the kernel and import all libraries."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.regression import GBTRegressor, GBTRegressionModel\n",
    "from pyspark.ml.evaluation import RegressionEvaluator\n",
    "from pyspark.ml.feature import VectorAssembler\n",
    "from pyspark.sql import functions as F\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.types import *\n",
    "from pyspark.sql.types import Row \n",
    "from sklearn.model_selection import train_test_split\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "import pickle\n",
    "import os"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "For preprocessing purposes, we have to separate the independent variables from dependant variables. The simplest way to achieve this is to export dataframes containing those variables to CSV files so that Spark will be able to read them later."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "store = pd.HDFStore('training.hdf5')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "if not os.path.exists('training_feat_csv') and not os.path.exists('training_target_csv'):\n",
    "    pd.DataFrame(store['input']).to_csv('training_feat_csv')\n",
    "    pd.DataFrame(store['target']).to_csv('training_target_csv')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we can instantiate the <i>SparkSession</i>. For testing, the application will be running locally with 2 cores, and 6 GB of memory for the driver process. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark = SparkSession.builder \\\n",
    "    .master(\"local[2]\") \\\n",
    "    .appName(\"ads-b machine learning\") \\\n",
    "    .config(\"spark.driver.memory\", \"6g\") \\\n",
    "    .getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Set number of output partitions\n",
    "spark.conf.set(\"spark.sql.shuffle.partitions\", 100)\n",
    "\n",
    "# Set log level\n",
    "spark.sparkContext.setLogLevel(\"ERROR\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Subsequently, we will load the CSV files containing input features and target variables."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_x = spark.read.format(\"csv\") \\\n",
    "    .options(header='True', inferSchema='True') \\\n",
    "    .load(\"training_feat_csv\") "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_y = spark.read.format(\"csv\") \\\n",
    "    .options(header='True', inferSchema='True') \\\n",
    "    .load(\"training_target_csv\") "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We will utilize the <i>VectorAssembler</i> to merge all features into a single vector column."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+--------------------+\n",
      "|_c0|            features|\n",
      "+---+--------------------+\n",
      "|  0|[1325.88,1.005387...|\n",
      "|  1|[12192.0,2.078536...|\n",
      "|  2|(41,[0,1,2,3,4,13...|\n",
      "|  3|[10988.04,1.43566...|\n",
      "|  4|(41,[0,1,2,3,4,13...|\n",
      "+---+--------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "vectorAssembler_x = VectorAssembler(inputCols=df_x.columns[1:], outputCol = 'features')\n",
    "df_x = vectorAssembler_x.transform(df_x)\n",
    "df_x = df_x.select('_c0', 'features')\n",
    "df_x.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Next, we will concatenate dataframes of input features and targets."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df_x.join(df_y, on='_c0')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+--------------------+------------------+------------------+-----------+\n",
      "|_c0|            features|          latitude|         longitude|geoAltitude|\n",
      "+---+--------------------+------------------+------------------+-----------+\n",
      "|148|[1280.16,7.764831...|         51.091732| 6.521911599999999|     1219.2|\n",
      "|229|[11582.4,1.347169...|          49.92778|7.4755460000000005|   11376.66|\n",
      "|307|[9083.04,4.398826...|          53.68795|       -0.39098403|    8983.98|\n",
      "|326|[11277.6,2.266288...|         50.756012|          8.111547|   11109.96|\n",
      "|463|[10363.2,3.180464...|53.063126000000004|         -2.331142|   10256.52|\n",
      "+---+--------------------+------------------+------------------+-----------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we can load the evaluation and test sets, and merge all input features into a single vector column using the <i>VectorAssembler</i> object created earlier."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "eval_df = spark.read.format(\"csv\") \\\n",
    "    .options(header='True', inferSchema='True') \\\n",
    "    .load(\"round2_competition_data/eval_test/eval_test.csv\") "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [],
   "source": [
    "x_cols = [col for col in eval_df.columns if col not in ['latitude', 'longitude', 'geoAltitude']]\n",
    "target_cols = ['latitude', 'longitude', 'geoAltitude']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+---------+-----------+-----------+\n",
      "|            features| latitude|  longitude|geoAltitude|\n",
      "+--------------------+---------+-----------+-----------+\n",
      "|[6355.08,5.770127...|52.378876|-0.65986633|     6286.5|\n",
      "|[9144.0,9.6834109...|  42.2872|  1.7981373|     8915.4|\n",
      "|(41,[0,1,2,3,4,13...|  47.0522|  5.9036407|    10820.4|\n",
      "|[10058.4,6.109551...|51.624344|  5.0306993|    9959.34|\n",
      "|[2125.98,7.921691...|50.820427|  3.6992645|    2080.26|\n",
      "+--------------------+---------+-----------+-----------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "eval_df = vectorAssembler_x.transform(eval_df)\n",
    "eval_df = eval_df.select('features', 'latitude', 'longitude', 'geoAltitude')\n",
    "eval_df.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In order to be able to join the evaluation dataframe with predictions, we have to create an index."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "eval_df = eval_df.withColumn('id', F.monotonically_increasing_id())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+---------+-----------+-----------+---+\n",
      "|            features| latitude|  longitude|geoAltitude| id|\n",
      "+--------------------+---------+-----------+-----------+---+\n",
      "|[6355.08,5.770127...|52.378876|-0.65986633|     6286.5|  0|\n",
      "|[9144.0,9.6834109...|  42.2872|  1.7981373|     8915.4|  1|\n",
      "|(41,[0,1,2,3,4,13...|  47.0522|  5.9036407|    10820.4|  2|\n",
      "|[10058.4,6.109551...|51.624344|  5.0306993|    9959.34|  3|\n",
      "|[2125.98,7.921691...|50.820427|  3.6992645|    2080.26|  4|\n",
      "+--------------------+---------+-----------+-----------+---+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "eval_df.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we will train and save the ensemble model for each of the target labels separately."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [],
   "source": [
    "models = {}\n",
    "\n",
    "for target in target_cols:\n",
    "    gbt = GBTRegressor(labelCol=target, maxDepth=5, maxBins=128)\n",
    "    model = gbt.fit(df)\n",
    "    model.write().overwrite().save('model_{}'.format(target))\n",
    "    \n",
    "    models[target] = model"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "After models are trained we can perform the evaluation."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [],
   "source": [
    "eval_metrics = {}\n",
    "\n",
    "for target in target_cols:\n",
    "    predictions = models[target].transform(eval_df)\n",
    "    predictions = predictions.withColumnRenamed('prediction', 'pred_{}'.format(target))\n",
    "    predictions = predictions.select('id', target, 'pred_{}'.format(target))\n",
    "    \n",
    "    evaluator = RegressionEvaluator(predictionCol='pred_{}'.format(target), labelCol=target)\n",
    "    r2 = evaluator.evaluate(predictions, {evaluator.metricName: \"r2\"})\n",
    "    rmse = evaluator.evaluate(predictions, {evaluator.metricName: \"rmse\"})\n",
    "    \n",
    "    eval_metrics.setdefault(target, {})\n",
    "    eval_metrics[target]['rmse'] = rmse\n",
    "    eval_metrics[target]['r2'] = r2\n",
    "        \n",
    "    predictions = predictions.select('id', 'pred_{}'.format(target))\n",
    "    eval_df = eval_df.join(predictions, on='id')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we can print the evaluation metrics."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>latitude</th>\n",
       "      <th>longitude</th>\n",
       "      <th>geoAltitude</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>rmse</th>\n",
       "      <td>0.584128</td>\n",
       "      <td>0.947858</td>\n",
       "      <td>101.624622</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>r2</th>\n",
       "      <td>0.981343</td>\n",
       "      <td>0.982412</td>\n",
       "      <td>0.999095</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "      latitude  longitude  geoAltitude\n",
       "rmse  0.584128   0.947858   101.624622\n",
       "r2    0.981343   0.982412     0.999095"
      ]
     },
     "execution_count": 28,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pd.DataFrame(eval_metrics)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Below is the dataframe containing input features, target labels and corresponding predictions."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+--------------------+---------+---------+-----------+------------------+-------------------+------------------+\n",
      "| id|            features| latitude|longitude|geoAltitude|     pred_latitude|     pred_longitude|  pred_geoAltitude|\n",
      "+---+--------------------+---------+---------+-----------+------------------+-------------------+------------------+\n",
      "| 26|[11574.78,7.18157...|54.548813| 9.581694|   11399.52| 54.12998804944908|  9.797703511734095|11443.640799377914|\n",
      "| 29|[7482.84,6.108366...|51.647003|2.5104501|    7452.36| 51.99551398301082|  2.975453452634862| 7328.026292611733|\n",
      "| 61|(41,[0,1,2,3,4,13...|51.944897|0.9212036|    7078.98| 51.29541225342674| 1.0365954278826348| 6994.998493908969|\n",
      "|303|[11574.78,1.09689...|43.607895|1.3043071|   11452.86|43.817816392782696| 2.0580020458729993|11432.630151365685|\n",
      "|474|[4922.52,7.318009...|51.751877|1.0184623|    4899.66| 50.31336482339997|0.46009835927473464| 4879.642274044956|\n",
      "+---+--------------------+---------+---------+-----------+------------------+-------------------+------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "eval_df.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We will also calculate the mean distance error."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [],
   "source": [
    "eval_df = eval_df \\\n",
    "    .withColumn('N1', 6378137 / (F.sqrt(1 - 8.1819190842622e-2**2 * F.sin(F.radians(F.col('latitude'))**2)))) \\\n",
    "    .withColumn('N2', 6378137 / (F.sqrt(1 - 8.1819190842622e-2**2 * F.sin(F.radians(F.col('pred_latitude'))**2))))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [],
   "source": [
    "eval_df = eval_df \\\n",
    "    .withColumn('x1', (F.col('N1') + F.col('geoAltitude')) * F.cos(F.radians(F.col('latitude'))) * F.cos(F.radians(F.col('longitude')))) \\\n",
    "    .withColumn('y1', (F.col('N1') + F.col('geoAltitude')) * F.cos(F.radians(F.col('latitude'))) * F.sin(F.radians(F.col('longitude')))) \\\n",
    "    .withColumn('z1', ((1 - 8.1819190842622e-2**2) * F.col('N1') + F.col('geoAltitude')) * F.sin(F.radians(F.col('latitude')))) \\\n",
    "    .withColumn('x2', (F.col('N2') + F.col('pred_geoAltitude')) * F.cos(F.radians(F.col('pred_latitude'))) * F.cos(F.radians(F.col('pred_longitude')))) \\\n",
    "    .withColumn('y2', (F.col('N2') + F.col('pred_geoAltitude')) * F.cos(F.radians(F.col('pred_latitude'))) * F.sin(F.radians(F.col('pred_longitude')))) \\\n",
    "    .withColumn('z2', ((1 - 8.1819190842622e-2**2) * F.col('N2') + F.col('pred_geoAltitude')) * F.sin(F.radians(F.col('pred_latitude'))))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [],
   "source": [
    "eval_df = eval_df.withColumn('dist', F.sqrt((F.col('x1') - F.col('x2'))**2 + \\\n",
    "                                           (F.col('y1') - F.col('y2'))**2 + \\\n",
    "                                           (F.col('z1') - F.col('z2'))**2) / 1000)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Mean distance error: 78.506 km\n"
     ]
    }
   ],
   "source": [
    "print('Mean distance error: {:.3f} km' \\\n",
    "      .format(eval_df.select((F.mean(F.abs(F.col('dist'))))).collect()[0][0]))"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "aircraft",
   "language": "python",
   "name": "aircraft"
  },
  "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.8.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}