mirror of
https://github.com/Microsoft/sql-server-samples.git
synced 2025-12-08 14:58:54 +00:00
Merge pull request #746 from shivsood/sample_fix
fixed hadoop path used when saving the local model to hdfs
This commit is contained in:
Umachandar Jayachandran
+431
-56
@@ -1,17 +1,17 @@
|
||||
{
|
||||
"metadata": {
|
||||
"kernelspec": {
|
||||
"name": "pyspark3kernel",
|
||||
"display_name": "PySpark3"
|
||||
"name": "pysparkkernel",
|
||||
"display_name": "PySpark"
|
||||
},
|
||||
"language_info": {
|
||||
"name": "pyspark3",
|
||||
"name": "pyspark",
|
||||
"mimetype": "text/x-python",
|
||||
"codemirror_mode": {
|
||||
"name": "python",
|
||||
"version": 3
|
||||
"version": 2
|
||||
},
|
||||
"pygments_lexer": "python3"
|
||||
"pygments_lexer": "python2"
|
||||
}
|
||||
},
|
||||
"nbformat_minor": 2,
|
||||
@@ -19,18 +19,80 @@
|
||||
"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<img src = \"Train_Score_Export_with_Spark.jpg\" style=\"float: center;\" alt=\"drawing\" width=\"900\">\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": {}
|
||||
"source": [
|
||||
"# Machine learning with SPARK in SQL Server 2019 Big Data Cluster\r\n",
|
||||
"Spark in Unified Big data compute engine that enables big data processing, Machine learning and AI\r\n",
|
||||
"\r\n",
|
||||
"Key Spark advantages are \r\n",
|
||||
"1. Distributed compute enging \r\n",
|
||||
"2. Choice of langauge (Python, R, Scala, Java)\r\n",
|
||||
"3. Single engine for Batch and Streaming jobs\r\n",
|
||||
"\r\n",
|
||||
"In 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",
|
||||
"<img src = \"Train_Score_Export_with_Spark.jpg\" style=\"float: center;\" alt=\"drawing\" width=\"900\">\r\n",
|
||||
"\r\n",
|
||||
"## Steps\r\n",
|
||||
"1. Explore your Data\r\n",
|
||||
"2. Data Prep and split Data as Training and Test set\r\n",
|
||||
"3. Model Training\r\n",
|
||||
"4. Model Scoring \r\n",
|
||||
"5. Persist as Spark Model\r\n",
|
||||
"6. Persist as Portable Model\r\n",
|
||||
"\r\n",
|
||||
"E2E 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": {
|
||||
"azdata_cell_guid": "3002b009-3bf8-4d80-b748-7a3c79151bba"
|
||||
}
|
||||
},
|
||||
{
|
||||
"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_data 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**, **hours_per_week**, and **education**\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": {}
|
||||
"source": [
|
||||
"## Step 1 - Explore your data\r\n",
|
||||
"### Load the data\r\n",
|
||||
"For 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_data under HDFS. \r\n",
|
||||
"Download [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\n",
|
||||
"1. **Label** - This refers to predicted value. This is represented as a column in the data. Label is **income** \r\n",
|
||||
"2. **Features** - This refers to the characteristics that are used to predict. **age**, **hours_per_week**, and **education**\r\n",
|
||||
"\r\n",
|
||||
"Note : 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\n",
|
||||
"In 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": {
|
||||
"azdata_cell_guid": "7c5aea92-c1bd-4195-a9b4-8e5a1a69de23"
|
||||
}
|
||||
},
|
||||
{
|
||||
"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": {},
|
||||
"source": [
|
||||
"datafile = \"/spark_data/AdultCensusIncome.csv\"\r\n",
|
||||
"\r\n",
|
||||
"#Read the data to a spark data frame.\r\n",
|
||||
"data_all = spark.read.format('csv').options(header='true', inferSchema='true', ignoreLeadingWhiteSpace='true', ignoreTrailingWhiteSpace='true').load(datafile)\r\n",
|
||||
"print(\"Number of rows: {}, Number of coulumns : {}\".format(data_all.count(), len(data_all.columns)))\r\n",
|
||||
"data_all.printSchema() \r\n",
|
||||
"\r\n",
|
||||
"#Replace \"-\" with \"_\" in column names\r\n",
|
||||
"columns_new = [col.replace(\"-\", \"_\") for col in data_all.columns]\r\n",
|
||||
"data_all = data_all.toDF(*columns_new)\r\n",
|
||||
"data_all.printSchema()\r\n",
|
||||
""
|
||||
],
|
||||
"metadata": {
|
||||
"azdata_cell_guid": "9fb741a4-ea63-45d0-ad37-373092b0ba36"
|
||||
},
|
||||
"outputs": [
|
||||
{
|
||||
"name": "stdout",
|
||||
@@ -40,7 +102,19 @@
|
||||
{
|
||||
"data": {
|
||||
"text/plain": "<IPython.core.display.HTML object>",
|
||||
"text/html": "<table>\n<tr><th>ID</th><th>YARN Application ID</th><th>Kind</th><th>State</th><th>Spark UI</th><th>Driver log</th><th>Current session?</th></tr><tr><td>20</td><td>application_1559313998190_0086</td><td>pyspark3</td><td>idle</td><td><a target=\"_blank\" href=\"https://10.193.17.7:30443/gateway/default/yarn/proxy/application_1559313998190_0086/\">Link</a></td><td><a target=\"_blank\" href=\"https://10.193.17.7:30443/gateway/default/yarn/container/container_1559313998190_0086_01_000001/root\">Link</a></td><td>✔</td></tr></table>"
|
||||
"text/html": "<table>\n<tr><th>ID</th><th>YARN Application ID</th><th>Kind</th><th>State</th><th>Spark UI</th><th>Driver log</th><th>Current session?</th></tr><tr><td>0</td><td>application_1582749467073_0001</td><td>pyspark</td><td>idle</td><td><a target=\"_blank\" href=\"https://10.193.21.202:30443/gateway/default/yarn/proxy/application_1582749467073_0001/\">Link</a></td><td><a target=\"_blank\" href=\"https://10.193.21.202:30443/gateway/default/yarn/container/container_1582749467073_0001_01_000001/root\">Link</a></td><td>✔</td></tr></table>"
|
||||
},
|
||||
"metadata": {},
|
||||
"output_type": "display_data"
|
||||
},
|
||||
{
|
||||
"data": {
|
||||
"text/plain": "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…",
|
||||
"application/vnd.jupyter.widget-view+json": {
|
||||
"version_major": 2,
|
||||
"version_minor": 0,
|
||||
"model_id": "91567484ce714d2ea856ce1ae23d09d7"
|
||||
}
|
||||
},
|
||||
"metadata": {},
|
||||
"output_type": "display_data"
|
||||
@@ -51,21 +125,20 @@
|
||||
"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)",
|
||||
"output_type": "stream"
|
||||
}
|
||||
],
|
||||
"execution_count": 2
|
||||
},
|
||||
{
|
||||
"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', 'education']).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', 'education', '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', 'education','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": [
|
||||
"data": {
|
||||
"text/plain": "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…",
|
||||
"application/vnd.jupyter.widget-view+json": {
|
||||
"version_major": 2,
|
||||
"version_minor": 0,
|
||||
"model_id": "aa11297baa07447c980a3ce97657fe75"
|
||||
}
|
||||
},
|
||||
"metadata": {},
|
||||
"output_type": "display_data"
|
||||
},
|
||||
{
|
||||
"name": "stdout",
|
||||
"text": "Select few columns to see the data\n+------+---+--------------+---------+\n|income|age|hours_per_week|education|\n+------+---+--------------+---------+\n| <=50K| 39| 40|Bachelors|\n| <=50K| 50| 13|Bachelors|\n| <=50K| 38| 40| HS-grad|\n| <=50K| 53| 40| 11th|\n| <=50K| 28| 40|Bachelors|\n| <=50K| 37| 40| Masters|\n| <=50K| 49| 16| 9th|\n| >50K| 52| 45| HS-grad|\n| >50K| 31| 50| Masters|\n| >50K| 42| 40|Bachelors|\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|education|income_code|\n+------+---+--------------+---------+-----------+\n| <=50K| 39| 40|Bachelors| 0|\n| <=50K| 50| 13|Bachelors| 0|\n| <=50K| 38| 40| HS-grad| 0|\n| <=50K| 53| 40| 11th| 0|\n| <=50K| 28| 40|Bachelors| 0|\n| <=50K| 37| 40| Masters| 0|\n| <=50K| 49| 16| 9th| 0|\n| >50K| 52| 45| HS-grad| 1|\n| >50K| 31| 50| Masters| 1|\n| >50K| 42| 40|Bachelors| 1|\n+------+---+--------------+---------+-----------+\nonly showing top 10 rows\n\nPrint a statistical summary of a few columns\n+-------+------+------------------+------------------+------------+-------------------+\n|summary|income| age| hours_per_week| education| income_code|\n+-------+------+------------------+------------------+------------+-------------------+\n| count| 32561| 32561| 32561| 32561| 32561|\n| mean| null| 38.58164675532078|40.437455852092995| null| 0.2408095574460244|\n| stddev| null|13.640432553581356|12.347428681731838| null|0.42758148856469247|\n| min| <=50K| 17| 1| 10th| 0|\n| max| >50K| 90| 99|Some-college| 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",
|
||||
"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)",
|
||||
"output_type": "stream"
|
||||
}
|
||||
],
|
||||
@@ -73,30 +146,104 @@
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"source": "# Choose feature columns and the label column.\r\nlabel = \"income\"\r\nxvars = [\"age\", \"hours_per_week\", 'education'] #numeric and string\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": {},
|
||||
"source": [
|
||||
"#Basic data exploration\r\n",
|
||||
"\r\n",
|
||||
"##1. Sub set the data and print some important columns\r\n",
|
||||
"print(\"Select few columns to see the data\")\r\n",
|
||||
"data_all.select(['income','age','hours_per_week', 'education']).show(10)\r\n",
|
||||
"\r\n",
|
||||
"## Find the number of distict values\r\n",
|
||||
"print(\"Number of distinct values for income\")\r\n",
|
||||
"ds_sub = data_all.select('income').distinct()\r\n",
|
||||
"ds_sub.show()\r\n",
|
||||
"\r\n",
|
||||
"##Add a numberic column(income_code) derived from income column\r\n",
|
||||
"print(\"Added numeric column(income_code) derived from income column\")\r\n",
|
||||
"from pyspark.sql.functions import expr\r\n",
|
||||
"\r\n",
|
||||
"df_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\n",
|
||||
"df_new.select(['income', 'age', 'hours_per_week', 'education', 'income_code']).show(10)\r\n",
|
||||
"\r\n",
|
||||
"##Summary statistical operations on dataframe\r\n",
|
||||
"print(\"Print a statistical summary of a few columns\")\r\n",
|
||||
"df_new.select(['income','age','hours_per_week', 'education','income_code']).describe().show()\r\n",
|
||||
"\r\n",
|
||||
"print(\"Calculate Co variance between a few columns to understand features to use\")\r\n",
|
||||
"mycov = df_new.stat.cov('income_code','hours_per_week')\r\n",
|
||||
"print(\"Covariance between income and hours_per_week is\", round(mycov,1))\r\n",
|
||||
"\r\n",
|
||||
"mycov = df_new.stat.cov('income_code','age')\r\n",
|
||||
"print(\"Covariance between income and age is\", round(mycov,1))\r\n",
|
||||
"\r\n",
|
||||
""
|
||||
],
|
||||
"metadata": {
|
||||
"azdata_cell_guid": "bed46457-a3b0-4972-afc0-f73107af7769"
|
||||
},
|
||||
"outputs": [
|
||||
{
|
||||
"data": {
|
||||
"text/plain": "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…",
|
||||
"application/vnd.jupyter.widget-view+json": {
|
||||
"version_major": 2,
|
||||
"version_minor": 0,
|
||||
"model_id": "793343b1c2c545da975144505a58cb27"
|
||||
}
|
||||
},
|
||||
"metadata": {},
|
||||
"output_type": "display_data"
|
||||
},
|
||||
{
|
||||
"name": "stdout",
|
||||
"text": "label = income\nfeatures = ['age', 'hours_per_week', 'education']\nCount of rows that are <=50K 24720\nCount of rows that are >50K 7841",
|
||||
"text": "Select few columns to see the data\n+------+---+--------------+---------+\n|income|age|hours_per_week|education|\n+------+---+--------------+---------+\n| <=50K| 39| 40|Bachelors|\n| <=50K| 50| 13|Bachelors|\n| <=50K| 38| 40| HS-grad|\n| <=50K| 53| 40| 11th|\n| <=50K| 28| 40|Bachelors|\n| <=50K| 37| 40| Masters|\n| <=50K| 49| 16| 9th|\n| >50K| 52| 45| HS-grad|\n| >50K| 31| 50| Masters|\n| >50K| 42| 40|Bachelors|\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|education|income_code|\n+------+---+--------------+---------+-----------+\n| <=50K| 39| 40|Bachelors| 0|\n| <=50K| 50| 13|Bachelors| 0|\n| <=50K| 38| 40| HS-grad| 0|\n| <=50K| 53| 40| 11th| 0|\n| <=50K| 28| 40|Bachelors| 0|\n| <=50K| 37| 40| Masters| 0|\n| <=50K| 49| 16| 9th| 0|\n| >50K| 52| 45| HS-grad| 1|\n| >50K| 31| 50| Masters| 1|\n| >50K| 42| 40|Bachelors| 1|\n+------+---+--------------+---------+-----------+\nonly showing top 10 rows\n\nPrint a statistical summary of a few columns\n+-------+------+------------------+------------------+------------+-------------------+\n|summary|income| age| hours_per_week| education| income_code|\n+-------+------+------------------+------------------+------------+-------------------+\n| count| 32561| 32561| 32561| 32561| 32561|\n| mean| null| 38.58164675532078|40.437455852092995| null| 0.2408095574460244|\n| stddev| null|13.640432553581356|12.347428681731838| null|0.42758148856469247|\n| min| <=50K| 17| 1| 10th| 0|\n| max| >50K| 90| 99|Some-college| 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",
|
||||
"output_type": "stream"
|
||||
}
|
||||
],
|
||||
"execution_count": 4
|
||||
},
|
||||
{
|
||||
"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": {},
|
||||
"source": [
|
||||
"# Choose feature columns and the label column.\r\n",
|
||||
"label = \"income\"\r\n",
|
||||
"xvars = [\"age\", \"hours_per_week\", 'education'] #numeric and string\r\n",
|
||||
"\r\n",
|
||||
"print(\"label = {}\".format(label))\r\n",
|
||||
"print(\"features = {}\".format(xvars))\r\n",
|
||||
"\r\n",
|
||||
"#Check label counts to check data bias\r\n",
|
||||
"print(\"Count of rows that are <=50K\", data_all[data_all.income==\"<=50K\"].count())\r\n",
|
||||
"print(\"Count of rows that are >50K\", data_all[data_all.income==\">50K\"].count())\r\n",
|
||||
"\r\n",
|
||||
"\r\n",
|
||||
"select_cols = xvars\r\n",
|
||||
"select_cols.append(label)\r\n",
|
||||
"data = data_all.select(select_cols)"
|
||||
],
|
||||
"metadata": {
|
||||
"azdata_cell_guid": "86f0640c-c2f4-4b4c-ae98-6e6806d4e77f"
|
||||
},
|
||||
"outputs": [
|
||||
{
|
||||
"data": {
|
||||
"text/plain": "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…",
|
||||
"application/vnd.jupyter.widget-view+json": {
|
||||
"version_major": 2,
|
||||
"version_minor": 0,
|
||||
"model_id": "b0be4b8053684bbaadc38b1acfb4ea00"
|
||||
}
|
||||
},
|
||||
"metadata": {},
|
||||
"output_type": "display_data"
|
||||
},
|
||||
{
|
||||
"name": "stdout",
|
||||
"text": "train (24469, 4)\ntest (8092, 4)\ntrain and test datasets saved to /spark_ml/AdultCensusIncomeTrain and /spark_ml/AdultCensusIncomeTest",
|
||||
"text": "label = income\nfeatures = ['age', 'hours_per_week', 'education']\nCount of rows that are <=50K 24720\nCount of rows that are >50K 7841",
|
||||
"output_type": "stream"
|
||||
}
|
||||
],
|
||||
@@ -104,17 +251,49 @@
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"source": "## Step 3 - Train a model\r\n[Spark ML pipelines](https://spark.apache.org/docs/latest/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.",
|
||||
"metadata": {}
|
||||
"source": [
|
||||
"## Step 2 - Split as training and test set\r\n",
|
||||
"We'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": {
|
||||
"azdata_cell_guid": "bcaaaac9-6efc-4dfe-87b0-ef375a9a0758"
|
||||
}
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"source": "from pyspark.ml import Pipeline, PipelineModel\r\nfrom pyspark.ml.feature import OneHotEncoderEstimator, 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(OneHotEncoderEstimator(inputCols=[tmpCol], outputCols=[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": {},
|
||||
"source": [
|
||||
"train, test = data.randomSplit([0.75, 0.25], seed=123)\r\n",
|
||||
"\r\n",
|
||||
"print(\"train ({}, {})\".format(train.count(), len(train.columns)))\r\n",
|
||||
"print(\"test ({}, {})\".format(test.count(), len(test.columns)))\r\n",
|
||||
"\r\n",
|
||||
"train_data_path = \"/spark_ml/AdultCensusIncomeTrain\"\r\n",
|
||||
"test_data_path = \"/spark_ml/AdultCensusIncomeTest\"\r\n",
|
||||
"\r\n",
|
||||
"train.write.mode('overwrite').orc(train_data_path)\r\n",
|
||||
"test.write.mode('overwrite').orc(test_data_path)\r\n",
|
||||
"print(\"train and test datasets saved to {} and {}\".format(train_data_path, test_data_path))"
|
||||
],
|
||||
"metadata": {
|
||||
"azdata_cell_guid": "7e487878-958b-4cd8-97fe-0199c29059a8"
|
||||
},
|
||||
"outputs": [
|
||||
{
|
||||
"data": {
|
||||
"text/plain": "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…",
|
||||
"application/vnd.jupyter.widget-view+json": {
|
||||
"version_major": 2,
|
||||
"version_minor": 0,
|
||||
"model_id": "e48983c1085b4e1e936352672e8368e4"
|
||||
}
|
||||
},
|
||||
"metadata": {},
|
||||
"output_type": "display_data"
|
||||
},
|
||||
{
|
||||
"name": "stdout",
|
||||
"text": "Using LogisticRegression model with Regularization Rate of 0.1.\nPipeline Created\nModel Trained\nModel is PipelineModel_1adfacc01e7a\nModel Stages [StringIndexer_ee8506a28443, OneHotEncoderEstimator_cb5dbefb5cce, StringIndexer_38769cda5ab3, VectorAssembler_a3c2d358bd55, LogisticRegressionModel: uid = LogisticRegression_18837c9488f5, numClasses = 2, numFeatures = 17]",
|
||||
"text": "train (24469, 4)\ntest (8092, 4)\ntrain and test datasets saved to /spark_ml/AdultCensusIncomeTrain and /spark_ml/AdultCensusIncomeTest",
|
||||
"output_type": "stream"
|
||||
}
|
||||
],
|
||||
@@ -122,17 +301,84 @@
|
||||
},
|
||||
{
|
||||
"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": {}
|
||||
"source": [
|
||||
"## Step 3 - Train a model\r\n",
|
||||
"[Spark ML pipelines](https://spark.apache.org/docs/latest/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."
|
||||
],
|
||||
"metadata": {
|
||||
"azdata_cell_guid": "eb03e212-5450-47bf-8fb9-e4d412f1ab74"
|
||||
}
|
||||
},
|
||||
{
|
||||
"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": {},
|
||||
"source": [
|
||||
"from pyspark.ml import Pipeline, PipelineModel\r\n",
|
||||
"from pyspark.ml.feature import OneHotEncoderEstimator, StringIndexer, VectorAssembler\r\n",
|
||||
"from pyspark.ml.classification import LogisticRegression\r\n",
|
||||
"\r\n",
|
||||
"reg = 0.1\r\n",
|
||||
"print(\"Using LogisticRegression model with Regularization Rate of {}.\".format(reg))\r\n",
|
||||
"\r\n",
|
||||
"# create a new Logistic Regression model.\r\n",
|
||||
"lr = LogisticRegression(regParam=reg)\r\n",
|
||||
"\r\n",
|
||||
"dtypes = dict(train.dtypes)\r\n",
|
||||
"dtypes.pop(label)\r\n",
|
||||
"\r\n",
|
||||
"si_xvars = []\r\n",
|
||||
"ohe_xvars = []\r\n",
|
||||
"featureCols = []\r\n",
|
||||
"for 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(OneHotEncoderEstimator(inputCols=[tmpCol], outputCols=[featureCol]))\r\n",
|
||||
" else:\r\n",
|
||||
" featureCols.append(key)\r\n",
|
||||
"\r\n",
|
||||
"# string-index the label column into a column named \"label\"\r\n",
|
||||
"si_label = StringIndexer(inputCol=label, outputCol='label')\r\n",
|
||||
"\r\n",
|
||||
"# assemble the encoded feature columns in to a column named \"features\"\r\n",
|
||||
"assembler = VectorAssembler(inputCols=featureCols, outputCol=\"features\")\r\n",
|
||||
"\r\n",
|
||||
"\r\n",
|
||||
"stages = []\r\n",
|
||||
"stages.extend(si_xvars)\r\n",
|
||||
"stages.extend(ohe_xvars)\r\n",
|
||||
"stages.append(si_label)\r\n",
|
||||
"stages.append(assembler)\r\n",
|
||||
"stages.append(lr)\r\n",
|
||||
"pipe = Pipeline(stages=stages)\r\n",
|
||||
"print(\"Pipeline Created\")\r\n",
|
||||
"\r\n",
|
||||
"model = pipe.fit(train)\r\n",
|
||||
"print(\"Model Trained\")\r\n",
|
||||
"print(\"Model is \", model)\r\n",
|
||||
"print(\"Model Stages\", model.stages)"
|
||||
],
|
||||
"metadata": {
|
||||
"azdata_cell_guid": "3e3c1258-3073-4ede-b3e4-84776b67ba3e"
|
||||
},
|
||||
"outputs": [
|
||||
{
|
||||
"data": {
|
||||
"text/plain": "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…",
|
||||
"application/vnd.jupyter.widget-view+json": {
|
||||
"version_major": 2,
|
||||
"version_minor": 0,
|
||||
"model_id": "3acc417b6c9b47acadcbc71acf0bc3eb"
|
||||
}
|
||||
},
|
||||
"metadata": {},
|
||||
"output_type": "display_data"
|
||||
},
|
||||
{
|
||||
"name": "stdout",
|
||||
"text": "Area under ROC: 0.7964496884726682\nArea Under PR: 0.5358180243123482\n+------+-----+----------+\n|income|label|prediction|\n+------+-----+----------+\n| <=50K| 0.0| 1.0|\n| >50K| 1.0| 1.0|\n| <=50K| 0.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| <=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| >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| 1.0| 1.0|\n| <=50K| 0.0| 1.0|\n| >50K| 1.0| 1.0|\n+------+-----+----------+\nonly showing top 20 rows",
|
||||
"text": "Using LogisticRegression model with Regularization Rate of 0.1.\nPipeline Created\nModel Trained\nModel is PipelineModel_dd10dea16fcc\nModel Stages [StringIndexer_a72809f2344f, OneHotEncoderEstimator_bff5847e2bae, StringIndexer_e62df3c26c82, VectorAssembler_e84f4c37608c, LogisticRegressionModel: uid = LogisticRegression_96385e6ff190, numClasses = 2, numFeatures = 17]",
|
||||
"output_type": "stream"
|
||||
}
|
||||
],
|
||||
@@ -140,17 +386,54 @@
|
||||
},
|
||||
{
|
||||
"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": {}
|
||||
"source": [
|
||||
"## Step 4 - Model scoring\r\n",
|
||||
"\r\n",
|
||||
"Predict using the model and Evaluate the model accuracy\r\n",
|
||||
" \r\n",
|
||||
"The 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": {
|
||||
"azdata_cell_guid": "c03ed2fd-6a23-40c3-b71c-608f14cde6af"
|
||||
}
|
||||
},
|
||||
{
|
||||
"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": {},
|
||||
"source": [
|
||||
"from pyspark.ml.evaluation import BinaryClassificationEvaluator\r\n",
|
||||
"\r\n",
|
||||
"# make prediction\r\n",
|
||||
"pred = model.transform(test)\r\n",
|
||||
"\r\n",
|
||||
"# evaluate. note only 2 metrics are supported out of the box by Spark ML.\r\n",
|
||||
"bce = BinaryClassificationEvaluator(rawPredictionCol='rawPrediction')\r\n",
|
||||
"au_roc = bce.setMetricName('areaUnderROC').evaluate(pred)\r\n",
|
||||
"au_prc = bce.setMetricName('areaUnderPR').evaluate(pred)\r\n",
|
||||
"\r\n",
|
||||
"print(\"Area under ROC: {}\".format(au_roc))\r\n",
|
||||
"print(\"Area Under PR: {}\".format(au_prc))\r\n",
|
||||
"\r\n",
|
||||
"pred[pred.prediction==1.0][pred.income,pred.label,pred.prediction].show()"
|
||||
],
|
||||
"metadata": {
|
||||
"azdata_cell_guid": "b1f5b3e5-a279-4d74-886f-e89789cb3f95"
|
||||
},
|
||||
"outputs": [
|
||||
{
|
||||
"data": {
|
||||
"text/plain": "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…",
|
||||
"application/vnd.jupyter.widget-view+json": {
|
||||
"version_major": 2,
|
||||
"version_minor": 0,
|
||||
"model_id": "381712d1e2e8407db531b071056ed753"
|
||||
}
|
||||
},
|
||||
"metadata": {},
|
||||
"output_type": "display_data"
|
||||
},
|
||||
{
|
||||
"name": "stdout",
|
||||
"text": "saved model to /spark_ml/AdultCensus.mml\nSuccessfully loaded from /spark_ml/AdultCensus.mml",
|
||||
"text": "Area under ROC: 0.7964496884726682\nArea Under PR: 0.5358180243123482\n+------+-----+----------+\n|income|label|prediction|\n+------+-----+----------+\n| <=50K| 0.0| 1.0|\n| >50K| 1.0| 1.0|\n| <=50K| 0.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| <=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| >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| 1.0| 1.0|\n| <=50K| 0.0| 1.0|\n| >50K| 1.0| 1.0|\n+------+-----+----------+\nonly showing top 20 rows",
|
||||
"output_type": "stream"
|
||||
}
|
||||
],
|
||||
@@ -158,21 +441,113 @@
|
||||
},
|
||||
{
|
||||
"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": {}
|
||||
"source": [
|
||||
"## Step 5 - Persist the Spark Models\r\n",
|
||||
"Finally 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": {
|
||||
"azdata_cell_guid": "095b7da6-6af7-4e15-b984-450ae22c3621"
|
||||
}
|
||||
},
|
||||
{
|
||||
"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": {},
|
||||
"source": [
|
||||
"model_name = \"AdultCensus.mml\"\r\n",
|
||||
"model_fs = \"/spark_ml/\" + model_name\r\n",
|
||||
"\r\n",
|
||||
"model.write().overwrite().save(model_fs)\r\n",
|
||||
"print(\"saved model to {}\".format(model_fs))\r\n",
|
||||
"\r\n",
|
||||
"# load the model file and check its same as the in-memory model\r\n",
|
||||
"model2 = PipelineModel.load(model_fs)\r\n",
|
||||
"assert str(model2) == str(model)\r\n",
|
||||
"print(\"Successfully loaded from {}\".format(model_fs))"
|
||||
],
|
||||
"metadata": {
|
||||
"azdata_cell_guid": "53dcb180-3b03-451e-b2cc-58ce16a62c13"
|
||||
},
|
||||
"outputs": [
|
||||
{
|
||||
"data": {
|
||||
"text/plain": "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…",
|
||||
"application/vnd.jupyter.widget-view+json": {
|
||||
"version_major": 2,
|
||||
"version_minor": 0,
|
||||
"model_id": "fdb264031f4c444aa0f52718d6c7ac86"
|
||||
}
|
||||
},
|
||||
"metadata": {},
|
||||
"output_type": "display_data"
|
||||
},
|
||||
{
|
||||
"name": "stdout",
|
||||
"text": "persist the mleap bundle from local to hdfs",
|
||||
"text": "saved model to /spark_ml/AdultCensus.mml\nSuccessfully loaded from /spark_ml/AdultCensus.mml",
|
||||
"output_type": "stream"
|
||||
}
|
||||
],
|
||||
"execution_count": 9
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"source": [
|
||||
"## Step 6 - Persist as Portable Model\r\n",
|
||||
"Here we persist the Model in as Portable Mleap bundle for use outside Spark."
|
||||
],
|
||||
"metadata": {
|
||||
"azdata_cell_guid": "92161ae4-60b6-4725-b0c4-f7b09c0cedf2"
|
||||
}
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"source": [
|
||||
"import os\r\n",
|
||||
"from mleap.pyspark.spark_support import SimpleSparkSerializer\r\n",
|
||||
"# serialize the model to a zip file in JSON format\r\n",
|
||||
"model_name_export = \"adult_census_pipeline.zip\"\r\n",
|
||||
"model_name_path = os.getcwd()\r\n",
|
||||
"model_file = os.path.join(model_name_path, model_name_export)\r\n",
|
||||
"\r\n",
|
||||
"# remove an old model file, if needed.\r\n",
|
||||
"try:\r\n",
|
||||
" os.remove(model_file)\r\n",
|
||||
"except OSError:\r\n",
|
||||
" pass\r\n",
|
||||
"\r\n",
|
||||
"model_file_path = \"jar:file:{}\".format(model_file)\r\n",
|
||||
"model.serializeToBundle(model_file_path, model.transform(train))\r\n",
|
||||
"\r\n",
|
||||
"print(\"persist the mleap bundle from local to hdfs\")\r\n",
|
||||
"\r\n",
|
||||
"from subprocess import Popen, PIPE\r\n",
|
||||
"proc = Popen([\"/opt/hadoop/bin/hdfs\", \"dfs\", \"-put\", \"-f\", model_file, os.path.join(\"/spark_ml\", model_name_export)], stdout=PIPE, stderr=PIPE)\r\n",
|
||||
"s_output, s_err = proc.communicate()\r\n",
|
||||
""
|
||||
],
|
||||
"metadata": {
|
||||
"azdata_cell_guid": "e06fa992-1569-4fc0-95e3-372ae7dbf889"
|
||||
},
|
||||
"outputs": [
|
||||
{
|
||||
"data": {
|
||||
"text/plain": "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…",
|
||||
"application/vnd.jupyter.widget-view+json": {
|
||||
"version_major": 2,
|
||||
"version_minor": 0,
|
||||
"model_id": "a3c85cc2e90c49a4a2cdd303243fd8d2"
|
||||
}
|
||||
},
|
||||
"metadata": {},
|
||||
"output_type": "display_data"
|
||||
},
|
||||
{
|
||||
"name": "stdout",
|
||||
"text": "persist the mleap bundle from local to hdfs\n/tmp/nm-local-dir/usercache/root/appcache/application_1582749467073_0001/container_1582749467073_0001_01_000001/adult_census_pipeline.zip\nadult_census_pipeline.zip\n/opt/hadoop/bin/hdfs/spark_ml/adult_census_pipeline.zip",
|
||||
"output_type": "stream"
|
||||
}
|
||||
],
|
||||
"execution_count": 27
|
||||
}
|
||||
]
|
||||
}
|
||||
Reference in New Issue
Block a user