diff --git a/samples/features/sql-big-data-cluster/spark/dataloading/hello_PySpark.ipynb b/samples/features/sql-big-data-cluster/spark/dataloading/hello_PySpark.ipynb new file mode 100644 index 00000000..30412807 --- /dev/null +++ b/samples/features/sql-big-data-cluster/spark/dataloading/hello_PySpark.ipynb @@ -0,0 +1,36 @@ +{ + "metadata": { + "kernelspec": { + "name": "pyspark3kernel", + "display_name": "PySpark3" + }, + "language_info": { + "name": "pyspark3", + "mimetype": "text/x-python", + "codemirror_mode": { + "name": "python", + "version": 3 + }, + "pygments_lexer": "python3" + } + }, + "nbformat_minor": 2, + "nbformat": 4, + "cells": [ + { + "cell_type": "code", + "source": "print(\"Hello World! \")\r\n\r\nimport sys\r\nprint(\"Python version \",sys.version)\r\n\r\n#Run some python in notebook\r\nnum = [i*i for i in range(0,20)]\r\nprint(\"My squared numbers \", num)\r\n", + "metadata": { + "language": "python" + }, + "outputs": [ + { + "name": "stdout", + "text": "Hello World! \nPython version 3.5.2 (default, Nov 12 2018, 13:43:14) \n[GCC 5.4.0 20160609]\nMy squared numbers [0, 1, 4, 9, 16, 25, 36, 49, 64, 81, 100, 121, 144, 169, 196, 225, 256, 289, 324, 361]", + "output_type": "stream" + } + ], + "execution_count": 1 + } + ] +} \ No newline at end of file diff --git a/samples/features/sql-big-data-cluster/spark/dataloading/hello_Scala.ipynb b/samples/features/sql-big-data-cluster/spark/dataloading/hello_Scala.ipynb new file mode 100644 index 00000000..2ae643b9 --- /dev/null +++ b/samples/features/sql-big-data-cluster/spark/dataloading/hello_Scala.ipynb @@ -0,0 +1,49 @@ +{ + "metadata": { + "kernelspec": { + "name": "sparkkernel", + "display_name": "Spark | Scala" + }, + "language_info": { + "name": "scala", + "mimetype": "text/x-scala", + "codemirror_mode": "text/x-scala", + "pygments_lexer": "scala" + } + }, + "nbformat_minor": 2, + "nbformat": 4, + "cells": [ + { + "cell_type": "code", + "source": "object HelloWorld {\r\n def main(args: Array[String]): Unit= { println(\"Hello Spark Scala\")\r\n }\r\n}\r\n", + "metadata": {}, + "outputs": [ + { + "name": "stdout", + "text": "Starting Spark application\n", + "output_type": "stream" + }, + { + "data": { + "text/plain": "", + "text/html": "\n
IDYARN Application IDKindStateSpark UIDriver logCurrent session?
3application_1554316083160_0004sparkidleLinkLink✔
" + }, + "metadata": {}, + "output_type": "display_data" + }, + { + "name": "stdout", + "text": "SparkSession available as 'spark'.\n", + "output_type": "stream" + }, + { + "name": "stdout", + "text": "defined object HelloWorld\n", + "output_type": "stream" + } + ], + "execution_count": 2 + } + ] +} \ No newline at end of file diff --git a/samples/features/sql-big-data-cluster/spark/dataloading/hello_sparkR.ipynb b/samples/features/sql-big-data-cluster/spark/dataloading/hello_sparkR.ipynb new file mode 100644 index 00000000..2cf4e00a --- /dev/null +++ b/samples/features/sql-big-data-cluster/spark/dataloading/hello_sparkR.ipynb @@ -0,0 +1,31 @@ +{ + "metadata": { + "kernelspec": { + "name": "sparkrkernel", + "display_name": "Spark | R" + }, + "language_info": { + "name": "sparkR", + "mimetype": "text/x-rsrc", + "codemirror_mode": "text/x-rsrc", + "pygments_lexer": "r" + } + }, + "nbformat_minor": 2, + "nbformat": 4, + "cells": [ + { + "cell_type": "code", + "source": "print(\"Hello SparkR\")\r\n\r\nhead(iris)\r\n", + "metadata": {}, + "outputs": [ + { + "name": "stdout", + "text": "[1] \"Hello SparkR\"\n Sepal.Length Sepal.Width Petal.Length Petal.Width Species\n1 5.1 3.5 1.4 0.2 setosa\n2 4.9 3.0 1.4 0.2 setosa\n3 4.7 3.2 1.3 0.2 setosa\n4 4.6 3.1 1.5 0.2 setosa\n5 5.0 3.6 1.4 0.2 setosa\n6 5.4 3.9 1.7 0.4 setosa", + "output_type": "stream" + } + ], + "execution_count": 5 + } + ] +} \ No newline at end of file diff --git a/samples/features/sql-big-data-cluster/spark/spark-sql.ipynb b/samples/features/sql-big-data-cluster/spark/dataloading/spark-sql.ipynb similarity index 100% rename from samples/features/sql-big-data-cluster/spark/spark-sql.ipynb rename to samples/features/sql-big-data-cluster/spark/dataloading/spark-sql.ipynb diff --git a/samples/features/sql-big-data-cluster/spark/spark_to_sql/spark_to_sql_jdbc.ipynb b/samples/features/sql-big-data-cluster/spark/spark_to_sql/spark_to_sql_jdbc.ipynb new file mode 100644 index 00000000..0531f80d --- /dev/null +++ b/samples/features/sql-big-data-cluster/spark/spark_to_sql/spark_to_sql_jdbc.ipynb @@ -0,0 +1,78 @@ +{ + "metadata": { + "kernelspec": { + "name": "pyspark3kernel", + "display_name": "PySpark3" + }, + "language_info": { + "name": "pyspark3", + "mimetype": "text/x-python", + "codemirror_mode": { + "name": "python", + "version": 3 + }, + "pygments_lexer": "python3" + } + }, + "nbformat_minor": 2, + "nbformat": 4, + "cells": [ + { + "cell_type": "markdown", + "source": "# Read and write from Spark to SQL\r\nA typical big data scenario is large scale ETL in Spark and writing the processed data to SQLServer. The following samples shows \r\n- reading a HDFS file, \r\n- some basic processing on it and \r\n- then processed data to SQL Server table.\r\n\r\nNeed a database precreated in SQL for this sample. Here we are using database name \"MyTestDatabase\" that can be created using SQL statements below.\r\n\r\n``` sql\r\nCreate DATABASE MyTestDatabase\r\nGO \r\n``` \r\n ", + "metadata": {} + }, + { + "cell_type": "code", + "source": "\r\n#Read a file and then write it to the SQL table\r\ndatafile = \"/spark_data/AdultCensusIncome.csv\"\r\ndf = spark.read.format('csv').options(header='true', inferSchema='true', ignoreLeadingWhiteSpace='true', ignoreTrailingWhiteSpace='true').load(datafile)\r\ndf.show(5)\r\n", + "metadata": {}, + "outputs": [ + { + "output_type": "stream", + "name": "stdout", + "text": "+---+----------------+------+---------+-------------+------------------+-----------------+-------------+-----+------+------------+------------+--------------+--------------+------+\n|age| workclass|fnlwgt|education|education-num| marital-status| occupation| relationship| race| sex|capital-gain|capital-loss|hours-per-week|native-country|income|\n+---+----------------+------+---------+-------------+------------------+-----------------+-------------+-----+------+------------+------------+--------------+--------------+------+\n| 39| State-gov| 77516|Bachelors| 13| Never-married| Adm-clerical|Not-in-family|White| Male| 2174| 0| 40| United-States| <=50K|\n| 50|Self-emp-not-inc| 83311|Bachelors| 13|Married-civ-spouse| Exec-managerial| Husband|White| Male| 0| 0| 13| United-States| <=50K|\n| 38| Private|215646| HS-grad| 9| Divorced|Handlers-cleaners|Not-in-family|White| Male| 0| 0| 40| United-States| <=50K|\n| 53| Private|234721| 11th| 7|Married-civ-spouse|Handlers-cleaners| Husband|Black| Male| 0| 0| 40| United-States| <=50K|\n| 28| Private|338409|Bachelors| 13|Married-civ-spouse| Prof-specialty| Wife|Black|Female| 0| 0| 40| Cuba| <=50K|\n+---+----------------+------+---------+-------------+------------------+-----------------+-------------+-----+------+------------+------------+--------------+--------------+------+\nonly showing top 5 rows" + } + ], + "execution_count": 8 + }, + { + "cell_type": "code", + "source": "\r\n#Process this data. Very simple data cleanup steps. Replacing \"-\" with \"_\" in column names\r\ncolumns_new = [col.replace(\"-\", \"_\") for col in df.columns]\r\ndf = df.toDF(*columns_new)\r\ndf.show(5)\r\n\r\n", + "metadata": {}, + "outputs": [ + { + "output_type": "stream", + "name": "stdout", + "text": "+---+----------------+------+---------+-------------+------------------+-----------------+-------------+-----+------+------------+------------+--------------+--------------+------+\n|age| workclass|fnlwgt|education|education_num| marital_status| occupation| relationship| race| sex|capital_gain|capital_loss|hours_per_week|native_country|income|\n+---+----------------+------+---------+-------------+------------------+-----------------+-------------+-----+------+------------+------------+--------------+--------------+------+\n| 39| State-gov| 77516|Bachelors| 13| Never-married| Adm-clerical|Not-in-family|White| Male| 2174| 0| 40| United-States| <=50K|\n| 50|Self-emp-not-inc| 83311|Bachelors| 13|Married-civ-spouse| Exec-managerial| Husband|White| Male| 0| 0| 13| United-States| <=50K|\n| 38| Private|215646| HS-grad| 9| Divorced|Handlers-cleaners|Not-in-family|White| Male| 0| 0| 40| United-States| <=50K|\n| 53| Private|234721| 11th| 7|Married-civ-spouse|Handlers-cleaners| Husband|Black| Male| 0| 0| 40| United-States| <=50K|\n| 28| Private|338409|Bachelors| 13|Married-civ-spouse| Prof-specialty| Wife|Black|Female| 0| 0| 40| Cuba| <=50K|\n+---+----------------+------+---------+-------------+------------------+-----------------+-------------+-----+------+------------+------------+--------------+--------------+------+\nonly showing top 5 rows" + } + ], + "execution_count": 9 + }, + { + "cell_type": "code", + "source": "#Write from Spark to SQL table using JDBC\r\nprint(\"Use build in JDBC connector to write to SQLServer master instance in Big data \")\r\n\r\nservername = \"jdbc:sqlserver://mssql-master-pool-0.service-master-pool\"\r\ndbname = \"MyTestDatabase\"\r\nurl = servername + \";\" + \"databaseName=\" + dbname + \";\"\r\n\r\nc = \"dbo.AdultCensus\"\r\nuser = \"sa\"\r\npassword = \"****\"\r\n\r\nprint(\"url is \", url)\r\n\r\ntry:\r\n df.write \\\r\n .format(\"jdbc\") \\\r\n .mode(\"overwrite\") \\\r\n .option(\"url\", url) \\\r\n .option(\"dbtable\", dbtable) \\\r\n .option(\"user\", user) \\\r\n .option(\"password\", password)\\\r\n .save()\r\nexcept ValueError as error :\r\n print(\"JDBC Write failed\", error)\r\n\r\nprint(\"JDBC Write done \")\r\n\r\n\r\n", + "metadata": {}, + "outputs": [ + { + "output_type": "stream", + "name": "stdout", + "text": "Use build in JDBC connector to write to SQLServer master instance in Big data \nurl is jdbc:sqlserver://mssql-master-pool-0.service-master-pool;databaseName=MyTestDatabase;\nJDBC Write done" + } + ], + "execution_count": 10 + }, + { + "cell_type": "code", + "source": "#Read to Spark from SQL table using JDBC\r\nprint(\"read data from SQL server table \")\r\njdbcDF = spark.read \\\r\n .format(\"jdbc\") \\\r\n .option(\"url\", url\r\n ) \\\r\n .option(\"dbtable\", dbtable) \\\r\n .option(\"user\", user) \\\r\n .option(\"password\", password) \\\r\n .load()\r\n\r\njdbcDF.show(5)", + "metadata": {}, + "outputs": [ + { + "output_type": "stream", + "name": "stdout", + "text": "read data from SQL server table \n+---+----------------+------+---------+-------------+------------------+-----------------+-------------+-----+------+------------+------------+--------------+--------------+------+\n|age| workclass|fnlwgt|education|education_num| marital_status| occupation| relationship| race| sex|capital_gain|capital_loss|hours_per_week|native_country|income|\n+---+----------------+------+---------+-------------+------------------+-----------------+-------------+-----+------+------------+------------+--------------+--------------+------+\n| 39| State-gov| 77516|Bachelors| 13| Never-married| Adm-clerical|Not-in-family|White| Male| 2174| 0| 40| United-States| <=50K|\n| 50|Self-emp-not-inc| 83311|Bachelors| 13|Married-civ-spouse| Exec-managerial| Husband|White| Male| 0| 0| 13| United-States| <=50K|\n| 38| Private|215646| HS-grad| 9| Divorced|Handlers-cleaners|Not-in-family|White| Male| 0| 0| 40| United-States| <=50K|\n| 53| Private|234721| 11th| 7|Married-civ-spouse|Handlers-cleaners| Husband|Black| Male| 0| 0| 40| United-States| <=50K|\n| 28| Private|338409|Bachelors| 13|Married-civ-spouse| Prof-specialty| Wife|Black|Female| 0| 0| 40| Cuba| <=50K|\n+---+----------------+------+---------+-------------+------------------+-----------------+-------------+-----+------+------------+------------+--------------+--------------+------+\nonly showing top 5 rows" + } + ], + "execution_count": 13 + } + ] +} \ No newline at end of file diff --git a/samples/features/sql-big-data-cluster/spark/sparkml/Train_Score_Export_with_Spark.jpg b/samples/features/sql-big-data-cluster/spark/sparkml/Train_Score_Export_with_Spark.jpg new file mode 100644 index 00000000..0cc016f8 Binary files /dev/null and b/samples/features/sql-big-data-cluster/spark/sparkml/Train_Score_Export_with_Spark.jpg differ diff --git a/samples/features/sql-big-data-cluster/spark/sparkml/train_score_export_ml_models_with_spark.ipynb b/samples/features/sql-big-data-cluster/spark/sparkml/train_score_export_ml_models_with_spark.ipynb new file mode 100644 index 00000000..26567e6d --- /dev/null +++ b/samples/features/sql-big-data-cluster/spark/sparkml/train_score_export_ml_models_with_spark.ipynb @@ -0,0 +1,160 @@ +{ + "metadata": { + "kernelspec": { + "name": "pyspark3kernel", + "display_name": "PySpark3" + }, + "language_info": { + "name": "pyspark3", + "mimetype": "text/x-python", + "codemirror_mode": { + "name": "python", + "version": 3 + }, + "pygments_lexer": "python3" + } + }, + "nbformat_minor": 2, + "nbformat": 4, + "cells": [ + { + "cell_type": "markdown", + "source": "# Machine learning with SPARK in SQL Server 2019 Big Data Cluster\r\nSpark in Unified Big data compute engine that enables big data processing, Machine learning and AI\r\n\r\nKey Spark advantages are \r\n1. Distributed compute enging \r\n2. Choice of langauge (Python, R, Scala, Java)\r\n3. Single engine for Batch and Streaming jobs\r\n\r\nIn this tutorial we'll cover how we can use Spark to create and deploy machine learning models. The example is a python(PySpark) sample. The same can also be done using Scala and R ( SparkR) in Spark.\r\n\r\n\"drawing\"\r\n\r\n## Steps\r\n1. Explore your Data\r\n2. Data Prep and split Data as Training and Test set\r\n3. Model Training\r\n4. Model Scoring \r\n5. Persist as Spark Model\r\n6. Persist as Portable Model\r\n\r\nE2E machine learning involves several additional step e.g data exploration, feature selection and principal component analysis,model selection etc. Many of these steps are ignored here for brevity.\r\n\r\n\r\n\r\n", + "metadata": {} + }, + { + "cell_type": "markdown", + "source": "## Step 1 - Explore your data\r\n### Load the data\r\nFor this example we'll use **AdultCensusIncome** data from [here]( https://amldockerdatasets.azureedge.net/AdultCensusIncome.csv ). From your Azure Data Studio connect to the HDFS/Spark gateway and create a directory called spark_ml under HDFS. \r\nDownload [AdultCensusIncome.csv]( https://amldockerdatasets.azureedge.net/AdultCensusIncome.csv ) to your local machine and upload to HDFS.Upload AdultCensusIncome.csv to the folder we created.\r\n\r\n### Exploratory Analysis\r\n- Baisc exploration on the data\r\n- Labels & Features\r\n1. **Label** - This refers to predicted value. This is represented as a column in the data. Label is **income** \r\n2. **Features** - This refers to the characteristics that are used to predict. **age** and **hours_per_week**\r\n\r\nNote : In reality features are chosen by applying some correlations techniques to understand what best characterize the Label we are predicting.\r\n\r\n### The Model we will build\r\nIn AdultCensusIncome.csv contains several columsn like Income range, age, hours-per-week, education, occupation etc. We'll build a model that can predict income range would be >50K or <50K.\r\n", + "metadata": {} + }, + { + "cell_type": "code", + "source": "datafile = \"/spark_data/AdultCensusIncome.csv\"\r\n\r\n#Read the data to a spark data frame.\r\ndata_all = spark.read.format('csv').options(header='true', inferSchema='true', ignoreLeadingWhiteSpace='true', ignoreTrailingWhiteSpace='true').load(datafile)\r\nprint(\"Number of rows: {}, Number of coulumns : {}\".format(data_all.count(), len(data_all.columns)))\r\ndata_all.printSchema() \r\n\r\n#Replace \"-\" with \"_\" in column names\r\ncolumns_new = [col.replace(\"-\", \"_\") for col in data_all.columns]\r\ndata_all = data_all.toDF(*columns_new)\r\ndata_all.printSchema()\r\n", + "metadata": {}, + "outputs": [ + { + "output_type": "stream", + "name": "stdout", + "text": "Number of rows: 32561, Number of coulumns : 15\nroot\n |-- age: integer (nullable = true)\n |-- workclass: string (nullable = true)\n |-- fnlwgt: integer (nullable = true)\n |-- education: string (nullable = true)\n |-- education-num: integer (nullable = true)\n |-- marital-status: string (nullable = true)\n |-- occupation: string (nullable = true)\n |-- relationship: string (nullable = true)\n |-- race: string (nullable = true)\n |-- sex: string (nullable = true)\n |-- capital-gain: integer (nullable = true)\n |-- capital-loss: integer (nullable = true)\n |-- hours-per-week: integer (nullable = true)\n |-- native-country: string (nullable = true)\n |-- income: string (nullable = true)\n\nroot\n |-- age: integer (nullable = true)\n |-- workclass: string (nullable = true)\n |-- fnlwgt: integer (nullable = true)\n |-- education: string (nullable = true)\n |-- education_num: integer (nullable = true)\n |-- marital_status: string (nullable = true)\n |-- occupation: string (nullable = true)\n |-- relationship: string (nullable = true)\n |-- race: string (nullable = true)\n |-- sex: string (nullable = true)\n |-- capital_gain: integer (nullable = true)\n |-- capital_loss: integer (nullable = true)\n |-- hours_per_week: integer (nullable = true)\n |-- native_country: string (nullable = true)\n |-- income: string (nullable = true)" + } + ], + "execution_count": 3 + }, + { + "cell_type": "code", + "source": "#Basic data exploration\r\n\r\n##1. Sub set the data and print some important columns\r\nprint(\"Select few columns to see the data\")\r\ndata_all.select(['income','age','hours_per_week']).show(10)\r\n\r\n## Find the number of distict values\r\nprint(\"Number of distinct values for income\")\r\nds_sub = data_all.select('income').distinct()\r\nds_sub.show()\r\n\r\n##Add a numberic column(income_code) derived from income column\r\nprint(\"Added numeric column(income_code) derived from income column\")\r\nfrom pyspark.sql.functions import expr\r\n\r\ndf_new = data_all.withColumn(\"income_code\", expr(\"case \\\r\n when income like '%<=50K%' then 0 \\\r\n when income like '%>50K%' then 1 \\\r\n else 2 end \"))\r\n\r\ndf_new.select(['income','age','hours_per_week','income_code']).show(10)\r\n\r\n##Summary statistical operations on dataframe\r\nprint(\"Print a statistical summary of a few columns\")\r\ndf_new.select(['income','age','hours_per_week','income_code']).describe().show()\r\n\r\nprint(\"Calculate Co variance between a few columns to understand features to use\")\r\nmycov = df_new.stat.cov('income_code','hours_per_week')\r\nprint(\"Covariance between income and hours_per_week is\", round(mycov,1))\r\n\r\nmycov = df_new.stat.cov('income_code','age')\r\nprint(\"Covariance between income and age is\", round(mycov,1))\r\n\r\n", + "metadata": {}, + "outputs": [ + { + "output_type": "stream", + "name": "stdout", + "text": "Select few columns to see the data\n+------+---+--------------+\n|income|age|hours_per_week|\n+------+---+--------------+\n| <=50K| 39| 40|\n| <=50K| 50| 13|\n| <=50K| 38| 40|\n| <=50K| 53| 40|\n| <=50K| 28| 40|\n| <=50K| 37| 40|\n| <=50K| 49| 16|\n| >50K| 52| 45|\n| >50K| 31| 50|\n| >50K| 42| 40|\n+------+---+--------------+\nonly showing top 10 rows\n\nNumber of distinct values for income\n+------+\n|income|\n+------+\n| <=50K|\n| >50K|\n+------+\n\nAdded numeric column(income_code) derived from income column\n+------+---+--------------+-----------+\n|income|age|hours_per_week|income_code|\n+------+---+--------------+-----------+\n| <=50K| 39| 40| 0|\n| <=50K| 50| 13| 0|\n| <=50K| 38| 40| 0|\n| <=50K| 53| 40| 0|\n| <=50K| 28| 40| 0|\n| <=50K| 37| 40| 0|\n| <=50K| 49| 16| 0|\n| >50K| 52| 45| 1|\n| >50K| 31| 50| 1|\n| >50K| 42| 40| 1|\n+------+---+--------------+-----------+\nonly showing top 10 rows\n\nPrint a statistical summary of a few columns\n+-------+------+------------------+------------------+-------------------+\n|summary|income| age| hours_per_week| income_code|\n+-------+------+------------------+------------------+-------------------+\n| count| 32561| 32561| 32561| 32561|\n| mean| null| 38.58164675532078|40.437455852092995| 0.2408095574460244|\n| stddev| null|13.640432553581356|12.347428681731838|0.42758148856469247|\n| min| <=50K| 17| 1| 0|\n| max| >50K| 90| 99| 1|\n+-------+------+------------------+------------------+-------------------+\n\nCalculate Co variance between a few columns to understand features to use\nCovariance between income and hours_per_week is 1.2\nCovariance between income and age is 1.4" + } + ], + "execution_count": 4 + }, + { + "cell_type": "code", + "source": "# Choose feature columns and the label column.\r\nlabel = \"income\"\r\nxvars = [\"age\", \"hours_per_week\"] #all numeric\r\n\r\nprint(\"label = {}\".format(label))\r\nprint(\"features = {}\".format(xvars))\r\n\r\n#Check label counts to check data bias\r\nprint(\"Count of rows that are <=50K\", data_all[data_all.income==\"<=50K\"].count())\r\nprint(\"Count of rows that are >50K\", data_all[data_all.income==\">50K\"].count())\r\n\r\n\r\nselect_cols = xvars\r\nselect_cols.append(label)\r\ndata = data_all.select(select_cols)", + "metadata": {}, + "outputs": [ + { + "output_type": "stream", + "name": "stdout", + "text": "label = income\nfeatures = ['age', 'hours_per_week']\nCount of rows that are <=50K 24720\nCount of rows that are >50K 7841" + } + ], + "execution_count": 5 + }, + { + "cell_type": "markdown", + "source": "## Step 2 - Split as training and test set\r\nWe'll use 75% of rows to train the model and rest of the 25% to evaluate the model. Additionally we persist the train and test data sets to HDFS storage. The step is not necessary , but shown to demonstrate saving and loading with ORC format. Other format e.g. Parquet may also be used. Post this step you should see 2 directories created with the name \"AdultCensusIncomeTrain\" and \"AdultCensusIncomeTest\"\r\n", + "metadata": {} + }, + { + "cell_type": "code", + "source": "train, test = data.randomSplit([0.75, 0.25], seed=123)\r\n\r\nprint(\"train ({}, {})\".format(train.count(), len(train.columns)))\r\nprint(\"test ({}, {})\".format(test.count(), len(test.columns)))\r\n\r\ntrain_data_path = \"/spark_ml/AdultCensusIncomeTrain\"\r\ntest_data_path = \"/spark_ml/AdultCensusIncomeTest\"\r\n\r\ntrain.write.mode('overwrite').orc(train_data_path)\r\ntest.write.mode('overwrite').orc(test_data_path)\r\nprint(\"train and test datasets saved to {} and {}\".format(train_data_path, test_data_path))", + "metadata": {}, + "outputs": [ + { + "output_type": "stream", + "name": "stdout", + "text": "train (24469, 3)\ntest (8092, 3)\ntrain and test datasets saved to /spark_ml/AdultCensusIncomeTrain and /spark_ml/AdultCensusIncomeTest" + } + ], + "execution_count": 6 + }, + { + "cell_type": "markdown", + "source": "## Step 3 - Train a model\r\n[Spark ML pipeline] ( https://spark.apache.org/docs/2.3.1/ml-pipeline.html ) allow to sequence all steps as a workflow and make it easier to experiment with various algorithms and their parameters. The following code first constructs the stages and then puts these stages together in Ml pipeline. LogisticRegression is used to create the model.\r\n\r\n", + "metadata": {} + }, + { + "cell_type": "code", + "source": "from pyspark.ml import Pipeline, PipelineModel\r\nfrom pyspark.ml.feature import OneHotEncoder, StringIndexer, VectorAssembler\r\nfrom pyspark.ml.classification import LogisticRegression\r\n\r\nreg = 0.1\r\nprint(\"Using LogisticRegression model with Regularization Rate of {}.\".format(reg))\r\n\r\n# create a new Logistic Regression model.\r\nlr = LogisticRegression(regParam=reg)\r\n\r\ndtypes = dict(train.dtypes)\r\ndtypes.pop(label)\r\n\r\nsi_xvars = []\r\nohe_xvars = []\r\nfeatureCols = []\r\nfor idx,key in enumerate(dtypes):\r\n if dtypes[key] == \"string\":\r\n featureCol = \"-\".join([key, \"encoded\"])\r\n featureCols.append(featureCol)\r\n \r\n tmpCol = \"-\".join([key, \"tmp\"])\r\n si_xvars.append(StringIndexer(inputCol=key, outputCol=tmpCol, handleInvalid=\"skip\")) #, handleInvalid=\"keep\"\r\n ohe_xvars.append(OneHotEncoder(inputCol=tmpCol, outputCol=featureCol))\r\n else:\r\n featureCols.append(key)\r\n\r\n# string-index the label column into a column named \"label\"\r\nsi_label = StringIndexer(inputCol=label, outputCol='label')\r\n\r\n# assemble the encoded feature columns in to a column named \"features\"\r\nassembler = VectorAssembler(inputCols=featureCols, outputCol=\"features\")\r\n\r\n\r\nstages = []\r\nstages.extend(si_xvars)\r\nstages.extend(ohe_xvars)\r\nstages.append(si_label)\r\nstages.append(assembler)\r\nstages.append(lr)\r\npipe = Pipeline(stages=stages)\r\nprint(\"Pipeline Created\")\r\n\r\nmodel = pipe.fit(train)\r\nprint(\"Model Trained\")\r\nprint(\"Model is \", model)\r\nprint(\"Model Stages\", model.stages)", + "metadata": {}, + "outputs": [ + { + "output_type": "stream", + "name": "stdout", + "text": "Using LogisticRegression model with Regularization Rate of 0.1.\nPipeline Created\nModel Trained\nModel is PipelineModel_e5284bc61285\nModel Stages [StringIndexer_1ecf86c8d2ae, VectorAssembler_450ee37e6955, LogisticRegressionModel: uid = LogisticRegression_deb52c17940d, numClasses = 2, numFeatures = 2]" + } + ], + "execution_count": 7 + }, + { + "cell_type": "markdown", + "source": "## Step 4 - Model scoring\r\n\r\nPredict using the model and Evaluate the model accuracy\r\n \r\nThe code below use test data set to predict the outcome using the model created in the step above. We measure accuracy of the model with areaUnderROC and areaUnderPR metric.", + "metadata": {} + }, + { + "cell_type": "code", + "source": "from pyspark.ml.evaluation import BinaryClassificationEvaluator\r\n\r\n# make prediction\r\npred = model.transform(test)\r\n\r\n# evaluate. note only 2 metrics are supported out of the box by Spark ML.\r\nbce = BinaryClassificationEvaluator(rawPredictionCol='rawPrediction')\r\nau_roc = bce.setMetricName('areaUnderROC').evaluate(pred)\r\nau_prc = bce.setMetricName('areaUnderPR').evaluate(pred)\r\n\r\nprint(\"Area under ROC: {}\".format(au_roc))\r\nprint(\"Area Under PR: {}\".format(au_prc))\r\n\r\npred[pred.prediction==1.0][pred.income,pred.label,pred.prediction].show()", + "metadata": {}, + "outputs": [ + { + "output_type": "stream", + "name": "stdout", + "text": "Area under ROC: 0.7363559303440261\nArea Under PR: 0.39475773290351296\n+------+-----+----------+\n|income|label|prediction|\n+------+-----+----------+\n| <=50K| 0.0| 1.0|\n| <=50K| 0.0| 1.0|\n| <=50K| 0.0| 1.0|\n| <=50K| 0.0| 1.0|\n| <=50K| 0.0| 1.0|\n| <=50K| 0.0| 1.0|\n| <=50K| 0.0| 1.0|\n| >50K| 1.0| 1.0|\n| >50K| 1.0| 1.0|\n| <=50K| 0.0| 1.0|\n| <=50K| 0.0| 1.0|\n| <=50K| 0.0| 1.0|\n| <=50K| 0.0| 1.0|\n| >50K| 1.0| 1.0|\n| >50K| 1.0| 1.0|\n| <=50K| 0.0| 1.0|\n| >50K| 1.0| 1.0|\n| <=50K| 0.0| 1.0|\n| >50K| 1.0| 1.0|\n| <=50K| 0.0| 1.0|\n+------+-----+----------+\nonly showing top 20 rows" + } + ], + "execution_count": 8 + }, + { + "cell_type": "markdown", + "source": "## Step 5 - Persist the Spark Models\r\nFinally we persist the model in HDFS for later use. Post this step the created model get saved as /spark_ml/AdultCensus.mml\r\n\r\n", + "metadata": {} + }, + { + "cell_type": "code", + "source": "model_name = \"AdultCensus.mml\"\r\nmodel_fs = \"/spark_ml/\" + model_name\r\n\r\nmodel.write().overwrite().save(model_fs)\r\nprint(\"saved model to {}\".format(model_fs))\r\n\r\n# load the model file and check its same as the in-memory model\r\nmodel2 = PipelineModel.load(model_fs)\r\nassert str(model2) == str(model)\r\nprint(\"Successfully loaded from {}\".format(model_fs))", + "metadata": {}, + "outputs": [ + { + "output_type": "stream", + "name": "stdout", + "text": "saved model to /spark_ml/AdultCensus.mml\nSuccessfully loaded from /spark_ml/AdultCensus.mml" + } + ], + "execution_count": 9 + }, + { + "cell_type": "markdown", + "source": "## Step 6 - Persist as Portable Model\r\nHere we persist the Model in as Portable Mleap bundle for use outside Spark.", + "metadata": {} + }, + { + "cell_type": "code", + "source": "import os\r\nfrom mleap.pyspark.spark_support import SimpleSparkSerializer\r\n# serialize the model to a zip file in JSON format\r\nmodel_name_export = \"adult_census_pipeline.zip\"\r\nmodel_name_path = os.getcwd()\r\nmodel_file = os.path.join(model_name_path, model_name_export)\r\n\r\n# remove an old model file, if needed.\r\ntry:\r\n os.remove(model_file)\r\nexcept OSError:\r\n pass\r\n\r\nmodel_file_path = \"jar:file:{}\".format(model_file)\r\nmodel.serializeToBundle(model_file_path, model.transform(train))\r\n\r\nprint(\"persist the mleap bundle from local to hdfs\")\r\nfrom subprocess import Popen, PIPE\r\nproc = Popen([\"hadoop\", \"fs\", \"-put\", \"-f\", model_file, os.path.join(\"/spark_ml\", model_name_export)], stdout=PIPE, stderr=PIPE)\r\ns_output, s_err = proc.communicate()\r\n", + "metadata": {}, + "outputs": [ + { + "output_type": "stream", + "name": "stdout", + "text": "persist the mleap bundle from local to hdfs" + } + ], + "execution_count": 10 + } + ] +} \ No newline at end of file