examples/Spark-cuML/pca/Spark_PCA_End_to_End.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "90826259",
   "metadata": {},
   "source": [
    "## Introduction\n",
    "\n",
    "In this notebook, we will show the integrated workflow of Spark RAPIDS accelerated ETL and PCA train & transform."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "0f555c95",
   "metadata": {},
   "outputs": [],
   "source": [
    "import org.apache.spark.ml.linalg._\n",
    "import org.apache.spark.sql.functions._"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6d47a7f0",
   "metadata": {},
   "source": [
    "### Generate dummy data for PCA benchmark\n",
    "\n",
    "Generate the sample data of 2048 columns and 50,000 rows"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "18e00096",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Waiting for a Spark session to start..."
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "rows = 50000\n",
       "dim = 2048\n",
       "r = scala.util.Random@18a448de\n",
       "prepareDf = [array_feature[0]: double, array_feature[1]: double ... 2046 more fields]\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "[array_feature[0]: double, array_feature[1]: double ... 2046 more fields]"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rows = 50000\n",
    "val dim = 2048\n",
    "val r = new scala.util.Random(0)\n",
    "var prepareDf = spark.createDataFrame(\n",
    "      (0 until rows).map(_ => Tuple1(Array.fill(dim)(r.nextDouble))))\n",
    "        .withColumnRenamed(\"_1\", \"array_feature\")\n",
    "        .select((0 until dim).map(i => col(\"array_feature\").getItem(i)): _*)\n",
    "prepareDf.write.mode(\"overwrite\").parquet(\"PCA_raw_parquet\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f2ed419b",
   "metadata": {},
   "source": [
    "### Read raw parquet data\n",
    "\n",
    "The parquet file contains the raw data for PCA train and transform.\n",
    "\n",
    "There're 2048 columns in the table naming as \"array_feature[0], array_feature[1] ... array_feature[2047]\"."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "0084a38f",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "df = [array_feature[0]: double, array_feature[1]: double ... 2046 more fields]\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "[array_feature[0]: double, array_feature[1]: double ... 2046 more fields]"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val df = spark.read.parquet(\"PCA_raw_parquet\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "58a33554",
   "metadata": {},
   "source": [
    "### ETL: Calculate mean value for each column\n",
    "\n",
    "PCA algorithm is expecting mean centered data as input, so use a simple ETL process to do mean centering."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "a152bd3d",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "dim = 2048\n",
       "avgValue = [0.5014784341440235,0.5007938298214618,0.4988382739107633,0.5004857021518329,0.4976086737881863,0.501459317390976,0.4998871629299758,0.5003749032337383,0.5004268051953419,0.4992212831312325,0.5002230208274252,0.49916485476370304,0.49928552249125024,0.5001192271170941,0.4974153011145406,0.500340861041902,0.500511698285404,0.5029175790341269,0.5000848064753295,0.49946358217105435,0.4991402970341374,0.4999057035861329,0.4993188619485362,0.49782509547668896,0.5001573241354326,0.4991954590903186,0.4988846878237177,0.5008673384728016,0.4982505290656533,0.5000069827383224,0.49830672380384944,0.49849188876978057,0.502253148518209,0.4995624384114367,0.5006052199700368,0.49922409882583835,0.4996825327694508,0.4983465266402566,0.5001149704952238...\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "[0.5014784341440235,0.5007938298214618,0.4988382739107633,0.5004857021518329,0.4976086737881863,0.501459317390976,0.4998871629299758,0.5003749032337383,0.5004268051953419,0.4992212831312325,0.5002230208274252,0.49916485476370304,0.49928552249125024,0.5001192271170941,0.4974153011145406,0.500340861041902,0.500511698285404,0.5029175790341269,0.5000848064753295,0.49946358217105435,0.4991402970341374,0.4999057035861329,0.4993188619485362,0.49782509547668896,0.5001573241354326,0.4991954590903186,0.4988846878237177,0.5008673384728016,0.4982505290656533,0.5000069827383224,0.49830672380384944,0.49849188876978057,0.502253148518209,0.4995624384114367,0.5006052199700368,0.49922409882583835,0.4996825327694508,0.4983465266402566,0.5001149704952238..."
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val dim = 2048\n",
    "val avgValue = df.select(\n",
    "    (0 until dim).map(\"array_feature[\" + _ + \"]\").map(col).map(avg): _*).first()\n",
    "val inputCols = (0 until dim).map(i =>\n",
    "    (col(\"array_feature[\" + i + \"]\") - avgValue.getDouble(i)).alias(\"feature_\"+i)\n",
    " )\n",
    "val meanCenterDf = df.select(inputCols:_*)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0bc302cd",
   "metadata": {},
   "source": [
    "### Spark RAPIDS accelerated PCA can accept ArrayType column as the input column.\n",
    "\n",
    "Comparing to the original Spark PCA requirement, there's no need to do extra `Vectorize` work for the input column.\n",
    "\n",
    "For example, the following code is required when using standard Spark PCA:\n",
    "\n",
    "```scala\n",
    "val convertToVector = udf((array: Seq[Double]) => {\n",
    "  Vectors.dense(array.map(_.toDouble).toArray)\n",
    "})\n",
    "val vectorDf = dataDf.withColumn(\"feature_vec\", convertToVector(col(\"feature\")))\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "ac57b415",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "dataDf = [feature_0: double, feature_1: double ... 2047 more fields]\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "[feature_0: double, feature_1: double ... 2047 more fields]"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val dataDf = meanCenterDf.withColumn(\"feature\",array(meanCenterDf.columns.map(col):_*))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ec1138d3",
   "metadata": {},
   "source": [
    "### Use Spark RAPIDS accelerated PCA\n",
    "\n",
    "Comparing to the original PCA training API:\n",
    "\n",
    "```scala\n",
    "val pca = new org.apache.spark.ml.feature.PCA()\n",
    "  .setInputCol(\"feature\")\n",
    "  .setOutputCol(\"pca_features\")\n",
    "  .setK(3)\n",
    "  .fit(vectorDf)\n",
    "```\n",
    "\n",
    "We used a customized class and user will need to do `no code change` to enjoy the GPU acceleration:\n",
    "\n",
    "```scala\n",
    "val pca = new com.nvidia.spark.ml.feature.PCA()\n",
    "...\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "8d9fb9bd",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "pcaGpu = pca_6b8d054604e4\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "pca_6b8d054604e4"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val pcaGpu = new com.nvidia.spark.ml.feature.PCA().setInputCol(\"feature\").setOutputCol(\"pca_features\").setK(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "d95ce4f9",
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "pcaModelGpu = PCAModel: uid=pca_6b8d054604e4, k=3\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Time taken: 8280 ms\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "PCAModel: uid=pca_6b8d054604e4, k=3"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val pcaModelGpu = spark.time(pcaGpu.fit(dataDf))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "0263a1b2",
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------------------------------+\n",
      "|pca_features                             |\n",
      "+-----------------------------------------+\n",
      "|[0.568805417, 0.041445481, 0.621107902]  |\n",
      "|[0.378405859, -0.244389411, -0.358809445]|\n",
      "|[0.421817533, -0.309621711, -0.159095405]|\n",
      "|[0.424088954, 0.09907811, 0.252832213]   |\n",
      "|[0.481344556, 0.303004001, 0.06884068]   |\n",
      "|[0.837941281, 0.113648256, -0.319001501] |\n",
      "|[0.093790516, -0.364140016, -0.33318393] |\n",
      "|[0.103996026, -0.174265839, 0.226559042] |\n",
      "|[-0.283206201, -0.487276589, 0.174362571]|\n",
      "|[0.101710379, 0.569866637, 0.118964435]  |\n",
      "+-----------------------------------------+\n",
      "only showing top 10 rows\n",
      "\n",
      "Time taken: 3284 ms\n"
     ]
    }
   ],
   "source": [
    "spark.time(pcaModelGpu.transform(dataDf).select(\"pca_features\").show(10, false))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4e09e7fa",
   "metadata": {},
   "source": [
    "### Use original Spark PCA"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "6db8b704",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "convertToVector = SparkUserDefinedFunction($Lambda$4927/1016137095@128d5536,org.apache.spark.ml.linalg.VectorUDT@3bfc3ba7,List(Some(class[value[0]: array<double>])),None,true,true)\n",
       "vectorDf = [feature_0: double, feature_1: double ... 2048 more fields]\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "[feature_0: double, feature_1: double ... 2048 more fields]"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val convertToVector = udf((array: Seq[Double]) => {\n",
    "  Vectors.dense(array.map(_.toDouble).toArray)\n",
    "})\n",
    "val vectorDf = dataDf.withColumn(\"feature_vec\", convertToVector(col(\"feature\")))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "b870173b",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "pcaCpu = pca_3f8feb827742\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "pca_3f8feb827742"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val pcaCpu = new org.apache.spark.ml.feature.PCA().setInputCol(\"feature_vec\").setOutputCol(\"pca_features\").setK(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "e3e1a9ca",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "pcaModelCpu = PCAModel: uid=pca_3f8feb827742, k=3\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Time taken: 140539 ms\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "PCAModel: uid=pca_3f8feb827742, k=3"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val pcaModelCpu = spark.time(pcaCpu.fit(vectorDf))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "504f0bb7",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------------------------------------------------+\n",
      "|pca_features                                                  |\n",
      "+--------------------------------------------------------------+\n",
      "|[0.5688054172628585,-0.04144548077183109,-0.6211079018457807] |\n",
      "|[0.37840585922945796,0.24438941118757604,0.3588094451238177]  |\n",
      "|[0.4218175332258925,0.3096217108376109,0.15909540520858537]   |\n",
      "|[0.4240889539599815,-0.09907811042793396,-0.2528322129752815] |\n",
      "|[0.4813445560531313,-0.3030040008580291,-0.06884068037876276] |\n",
      "|[0.8379412808563966,-0.11364825624115062,0.3190015014324452]  |\n",
      "|[0.09379051625949268,0.3641400160124998,0.33318393004824964]  |\n",
      "|[0.10399602625088979,0.17426583892592548,-0.2265590421381768] |\n",
      "|[-0.2832062006131796,0.4872765894121887,-0.1743625713365004]  |\n",
      "|[0.10171037937872408,-0.5698666372294762,-0.11896443456600647]|\n",
      "+--------------------------------------------------------------+\n",
      "only showing top 10 rows\n",
      "\n",
      "Time taken: 12738 ms\n"
     ]
    }
   ],
   "source": [
    "spark.time(pcaModelCpu.transform(vectorDf).select(\"pca_features\").show(10, false))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d486eb9a",
   "metadata": {},
   "source": [
    "### Summary\n",
    "\n",
    "With the data of 50,000 rows, we achived:\n",
    "\n",
    "the speedup for training: 140539 / 8280  = `16.97`\n",
    "\n",
    "the speedup for transform: 12738 / 3284   = `3.87`"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "671bc74d",
   "metadata": {},
   "source": [
    "### Note\n",
    "\n",
    "Some columns in GPU output have different signs from that in CPU output, this is due to the calculation nature of SVD algorithm which doesn't impact the effectiveness of the SVD results. More details could be found in the [wiki](https://en.wikipedia.org/wiki/Singular_value_decomposition#Relation_to_eigenvalue_decomposition)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4d53fb76",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "spark-rapids-ml-pca - Scala",
   "language": "scala",
   "name": "spark-rapids-ml-pca_scala"
  },
  "language_info": {
   "codemirror_mode": "text/x-scala",
   "file_extension": ".scala",
   "mimetype": "text/x-scala",
   "name": "scala",
   "pygments_lexer": "scala",
   "version": "2.12.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}