spark_lynda_1.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[id: string, last_name: string, email: string, gender: string, department: string, start_date: string, salary: string, job_title: string, region_id: string]"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from pyspark.context import SparkContext\n",
    "from pyspark.sql.session import SparkSession\n",
    "sc = SparkContext('local')\n",
    "spark = SparkSession(sc)\n",
    "\n",
    "emp_df = spark.read.csv('employee.txt', header = True)\n",
    "emp_df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "StructType(List(StructField(id,StringType,true),StructField(last_name,StringType,true),StructField(email,StringType,true),StructField(gender,StringType,true),StructField(department,StringType,true),StructField(start_date,StringType,true),StructField(salary,StringType,true),StructField(job_title,StringType,true),StructField(region_id,StringType,true)))"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "emp_df.schema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- id: string (nullable = true)\n",
      " |-- last_name: string (nullable = true)\n",
      " |-- email: string (nullable = true)\n",
      " |-- gender: string (nullable = true)\n",
      " |-- department: string (nullable = true)\n",
      " |-- start_date: string (nullable = true)\n",
      " |-- salary: string (nullable = true)\n",
      " |-- job_title: string (nullable = true)\n",
      " |-- region_id: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "emp_df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['id',\n",
       " 'last_name',\n",
       " 'email',\n",
       " 'gender',\n",
       " 'department',\n",
       " 'start_date',\n",
       " 'salary',\n",
       " 'job_title',\n",
       " 'region_id']"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "emp_df.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(id='1', last_name=\"'Kelley'\", email=\"'rkelley0@soundcloud.com'\", gender=\"'Female'\", department=\"'Computers'\", start_date=\"'10/2/2009'\", salary='67470', job_title=\"'Structural Engineer'\", region_id='2'),\n",
       " Row(id='2', last_name=\"'Armstrong'\", email=\"'sarmstrong1@infoseek.co.jp'\", gender=\"'Male'\", department=\"'Sports'\", start_date=\"'3/31/2008'\", salary='71869', job_title=\"'Financial Advisor'\", region_id='2'),\n",
       " Row(id='3', last_name=\"'Carr'\", email=\"'fcarr2@woothemes.com'\", gender=\"'Male'\", department=\"'Automotive'\", start_date=\"'7/12/2009'\", salary='101768', job_title=\"'Recruiting Manager'\", region_id='3'),\n",
       " Row(id='4', last_name=\"'Murray'\", email=\"'jmurray3@gov.uk'\", gender=\"'Female'\", department=\"'Jewelery'\", start_date=\"'12/25/2014'\", salary='96897', job_title=\"'Desktop Support Technician'\", region_id='3'),\n",
       " Row(id='5', last_name=\"'Ellis'\", email=\"'jellis4@sciencedirect.com'\", gender=\"'Female'\", department=\"'Grocery'\", start_date=\"'9/19/2002'\", salary='63702', job_title=\"'Software Engineer III'\", region_id='7')]"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "emp_df.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1000"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "emp_df.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "85"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sample_df = emp_df.sample(False, 0.1)\n",
    "sample_df.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "478"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "emp_mgrs_df = emp_df.filter('salary >= 100000')\n",
    "emp_mgrs_df.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+\n",
      "|salary|\n",
      "+------+\n",
      "|101768|\n",
      "|118497|\n",
      "|108657|\n",
      "|108093|\n",
      "|121966|\n",
      "|141139|\n",
      "|106659|\n",
      "|148952|\n",
      "|109890|\n",
      "|115274|\n",
      "|144724|\n",
      "|126103|\n",
      "|144965|\n",
      "|113507|\n",
      "|120579|\n",
      "|107222|\n",
      "|125668|\n",
      "|113857|\n",
      "|108378|\n",
      "|133424|\n",
      "+------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "emp_mgrs_df.select('salary').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(id=1, features=DenseVector([10.0, 10000.0, 1.0]))]"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from pyspark.ml.feature import MinMaxScaler\n",
    "from pyspark.ml.linalg import Vectors\n",
    "\n",
    "features_df = spark.createDataFrame([\n",
    "    (1, Vectors.dense([10.0, 10000.0, 1.0]),),\n",
    "    (2, Vectors.dense([20.0, 30000.0, 2.0]),),\n",
    "    (3, Vectors.dense([30.0, 40000.0, 3.0]),)\n",
    "], ['id', 'features'])\n",
    "\n",
    "features_df.take(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(id=1, features=DenseVector([10.0, 10000.0, 1.0]), sfeatures=DenseVector([0.0, 0.0, 0.0]))]"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "feature_scaler = MinMaxScaler(inputCol='features', outputCol='sfeatures')\n",
    "smodel = feature_scaler.fit(features_df)\n",
    "sfeatures_df = smodel.transform(features_df)\n",
    "sfeatures_df.take(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------+--------------------+\n",
      "|          features|           sfeatures|\n",
      "+------------------+--------------------+\n",
      "|[10.0,10000.0,1.0]|       [0.0,0.0,0.0]|\n",
      "|[20.0,30000.0,2.0]|[0.5,0.6666666666...|\n",
      "|[30.0,40000.0,3.0]|       [1.0,1.0,1.0]|\n",
      "+------------------+--------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sfeatures_df.select('features', 'sfeatures').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(id=1, features=DenseVector([10.0, 10000.0, 1.0]))]"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from pyspark.ml.feature import StandardScaler\n",
    "\n",
    "features_df = spark.createDataFrame([\n",
    "    (1, Vectors.dense([10.0, 10000.00, 1.0]),),\n",
    "    (2, Vectors.dense([20.0, 30000.00, 2.0]),),\n",
    "    (3, Vectors.dense([30.0, 40000.00, 3.0]),)\n",
    "], ['id', 'features'])\n",
    "features_df.take(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(id=1, features=DenseVector([10.0, 10000.0, 1.0]), sfeatures=DenseVector([-1.0, -1.0911, -1.0]))]"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "feature_stand_scaler = StandardScaler(inputCol='features', outputCol='sfeatures', withStd=True, withMean=True)\n",
    "stand_smodel = feature_stand_scaler.fit(features_df)\n",
    "stand_sfeatures_df = stand_smodel.transform(features_df)\n",
    "stand_sfeatures_df.take(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+------------------+--------------------+\n",
      "| id|          features|           sfeatures|\n",
      "+---+------------------+--------------------+\n",
      "|  1|[10.0,10000.0,1.0]|[-1.0,-1.09108945...|\n",
      "|  2|[20.0,30000.0,2.0]|[0.0,0.2182178902...|\n",
      "|  3|[30.0,40000.0,3.0]|[1.0,0.8728715609...|\n",
      "+---+------------------+--------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "stand_sfeatures_df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+\n",
      "|features|\n",
      "+--------+\n",
      "|  -800.0|\n",
      "|   -10.5|\n",
      "|    -1.7|\n",
      "|     0.0|\n",
      "|     8.2|\n",
      "|    90.1|\n",
      "+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.ml.feature import Bucketizer\n",
    "splits = [-float('inf'), -10.0, 0.0, 10.0, float('inf')]\n",
    "\n",
    "b_data = [(-800.0,), (-10.5,), (-1.7,), (0.0,), (8.2,), (90.1,)]\n",
    "b_df = spark.createDataFrame(b_data, ['features'])\n",
    "b_df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+---------+\n",
      "|features|bfeatures|\n",
      "+--------+---------+\n",
      "|  -800.0|      0.0|\n",
      "|   -10.5|      0.0|\n",
      "|    -1.7|      1.0|\n",
      "|     0.0|      2.0|\n",
      "|     8.2|      2.0|\n",
      "|    90.1|      3.0|\n",
      "+--------+---------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.ml.feature import Bucketizer\n",
    "bucketizer = Bucketizer(splits = splits, inputCol = 'features', outputCol='bfeatures')\n",
    "bucketed_df = bucketizer.transform(b_df)\n",
    "bucketed_df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+--------------------+\n",
      "| id|            sentence|\n",
      "+---+--------------------+\n",
      "|  1|This is an introd...|\n",
      "|  2|MLlib includes li...|\n",
      "|  3|It also contains ...|\n",
      "+---+--------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.ml.feature import Tokenizer\n",
    "\n",
    "sentences_df = spark.createDataFrame([\n",
    "    (1, 'This is an introduction to Spark MLlib'),\n",
    "    (2, 'MLlib includes libraries for classification and regression'),\n",
    "    (3, 'It also contains supporting tools for pipelines')], ['id', 'sentence'])\n",
    "sentences_df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+--------------------+--------------------+\n",
      "| id|            sentence|               words|\n",
      "+---+--------------------+--------------------+\n",
      "|  1|This is an introd...|[this, is, an, in...|\n",
      "|  2|MLlib includes li...|[mllib, includes,...|\n",
      "|  3|It also contains ...|[it, also, contai...|\n",
      "+---+--------------------+--------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sent_token = Tokenizer(inputCol='sentence', outputCol='words')\n",
    "sent_tokenized_df = sent_token.transform(sentences_df)\n",
    "sent_tokenized_df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[id: bigint, sentence: string]"
      ]
     },
     "execution_count": 20,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from pyspark.ml.feature import HashingTF, IDF\n",
    "\n",
    "sentences_df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(id=1, sentence='This is an introduction to Spark MLlib')]"
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sentences_df.take(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(id=1, sentence='This is an introduction to Spark MLlib', words=['this', 'is', 'an', 'introduction', 'to', 'spark', 'mllib'])]"
      ]
     },
     "execution_count": 22,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sent_tokenized_df.take(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+--------------------+--------------------+--------------------+\n",
      "| id|            sentence|               words|         rawFeatures|\n",
      "+---+--------------------+--------------------+--------------------+\n",
      "|  1|This is an introd...|[this, is, an, in...|(20,[1,5,6,8,12,1...|\n",
      "|  2|MLlib includes li...|[mllib, includes,...|(20,[1,6,9,12,13,...|\n",
      "|  3|It also contains ...|[it, also, contai...|(20,[0,8,10,12,15...|\n",
      "+---+--------------------+--------------------+--------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "hashingTF = HashingTF(inputCol = 'words', outputCol = 'rawFeatures', numFeatures = 20)\n",
    "sent_hfTF_df = hashingTF.transform(sent_tokenized_df)\n",
    "sent_hfTF_df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+--------------------+--------------------+--------------------+--------------------+\n",
      "| id|            sentence|               words|         rawFeatures|        idf_features|\n",
      "+---+--------------------+--------------------+--------------------+--------------------+\n",
      "|  1|This is an introd...|[this, is, an, in...|(20,[1,5,6,8,12,1...|(20,[1,5,6,8,12,1...|\n",
      "|  2|MLlib includes li...|[mllib, includes,...|(20,[1,6,9,12,13,...|(20,[1,6,9,12,13,...|\n",
      "|  3|It also contains ...|[it, also, contai...|(20,[0,8,10,12,15...|(20,[0,8,10,12,15...|\n",
      "+---+--------------------+--------------------+--------------------+--------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "idf = IDF(inputCol = 'rawFeatures', outputCol = 'idf_features')\n",
    "idfModel = idf.fit(sent_hfTF_df)\n",
    "tfidf_df = idfModel.transform(sent_hfTF_df)\n",
    "tfidf_df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(id=1, sentence='This is an introduction to Spark MLlib', words=['this', 'is', 'an', 'introduction', 'to', 'spark', 'mllib'], rawFeatures=SparseVector(20, {1: 2.0, 5: 1.0, 6: 1.0, 8: 1.0, 12: 1.0, 13: 1.0}), idf_features=SparseVector(20, {1: 0.5754, 5: 0.6931, 6: 0.2877, 8: 0.2877, 12: 0.0, 13: 0.2877}))]"
      ]
     },
     "execution_count": 25,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "tfidf_df.take(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[col1: int, col2: int, col3: int]"
      ]
     },
     "execution_count": 26,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from pyspark.ml.feature import VectorAssembler\n",
    "from pyspark.ml.clustering import KMeans\n",
    "\n",
    "cluster_df = spark.read.csv('clustering_dataset.csv', header=True, inferSchema = True)\n",
    "cluster_df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+----+----+\n",
      "|col1|col2|col3|\n",
      "+----+----+----+\n",
      "|   7|   4|   1|\n",
      "|   7|   7|   9|\n",
      "|   7|   9|   6|\n",
      "|   1|   6|   5|\n",
      "|   6|   7|   7|\n",
      "|   7|   9|   4|\n",
      "|   7|  10|   6|\n",
      "|   7|   8|   2|\n",
      "|   8|   3|   8|\n",
      "|   4|  10|   5|\n",
      "|   7|   4|   5|\n",
      "|   7|   8|   4|\n",
      "|   2|   5|   1|\n",
      "|   2|   6|   2|\n",
      "|   2|   3|   8|\n",
      "|   3|   9|   1|\n",
      "|   4|   2|   9|\n",
      "|   1|   7|   1|\n",
      "|   6|   2|   3|\n",
      "|   4|   1|   9|\n",
      "|   4|   8|   5|\n",
      "|   6|   6|   7|\n",
      "|   4|   6|   2|\n",
      "|   8|   1|   1|\n",
      "|   7|   5|  10|\n",
      "|  17|  25|  21|\n",
      "|  15|  23|  32|\n",
      "|  42|  25|  45|\n",
      "|  41|  47|  21|\n",
      "|  37|  20|  27|\n",
      "|  40|  18|  26|\n",
      "|  41|  28|  50|\n",
      "|  32|  25|  40|\n",
      "|  24|  29|  35|\n",
      "|  47|  18|  47|\n",
      "|  36|  42|  45|\n",
      "|  49|  29|  15|\n",
      "|  47|  39|  22|\n",
      "|  38|  27|  25|\n",
      "|  45|  23|  40|\n",
      "|  23|  36|  19|\n",
      "|  47|  40|  50|\n",
      "|  37|  30|  40|\n",
      "|  42|  48|  41|\n",
      "|  29|  31|  21|\n",
      "|  36|  39|  48|\n",
      "|  50|  24|  31|\n",
      "|  42|  44|  37|\n",
      "|  37|  39|  46|\n",
      "|  22|  40|  30|\n",
      "|  17|  29|  41|\n",
      "|  85| 100|  69|\n",
      "|  68|  76|  67|\n",
      "|  76|  70|  93|\n",
      "|  62|  66|  91|\n",
      "|  83|  93|  76|\n",
      "|  95|  72|  63|\n",
      "|  75|  94|  95|\n",
      "|  83|  72|  80|\n",
      "|  93|  87|  76|\n",
      "|  86|  93|  63|\n",
      "|  97|  82|  75|\n",
      "|  61|  74|  74|\n",
      "|  84|  90| 100|\n",
      "|  77|  67|  97|\n",
      "|  61|  82|  73|\n",
      "|  81|  60|  69|\n",
      "|  67|  80|  98|\n",
      "|  94|  82|  60|\n",
      "|  69|  73|  74|\n",
      "|  74|  96|  80|\n",
      "|  86|  62|  61|\n",
      "|  88|  68|  95|\n",
      "|  99|  67|  80|\n",
      "|  76|  95|  70|\n",
      "+----+----+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "cluster_df.show(75)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+----+----+--------------+\n",
      "|col1|col2|col3|      features|\n",
      "+----+----+----+--------------+\n",
      "|   7|   4|   1| [7.0,4.0,1.0]|\n",
      "|   7|   7|   9| [7.0,7.0,9.0]|\n",
      "|   7|   9|   6| [7.0,9.0,6.0]|\n",
      "|   1|   6|   5| [1.0,6.0,5.0]|\n",
      "|   6|   7|   7| [6.0,7.0,7.0]|\n",
      "|   7|   9|   4| [7.0,9.0,4.0]|\n",
      "|   7|  10|   6|[7.0,10.0,6.0]|\n",
      "|   7|   8|   2| [7.0,8.0,2.0]|\n",
      "|   8|   3|   8| [8.0,3.0,8.0]|\n",
      "|   4|  10|   5|[4.0,10.0,5.0]|\n",
      "|   7|   4|   5| [7.0,4.0,5.0]|\n",
      "|   7|   8|   4| [7.0,8.0,4.0]|\n",
      "|   2|   5|   1| [2.0,5.0,1.0]|\n",
      "|   2|   6|   2| [2.0,6.0,2.0]|\n",
      "|   2|   3|   8| [2.0,3.0,8.0]|\n",
      "|   3|   9|   1| [3.0,9.0,1.0]|\n",
      "|   4|   2|   9| [4.0,2.0,9.0]|\n",
      "|   1|   7|   1| [1.0,7.0,1.0]|\n",
      "|   6|   2|   3| [6.0,2.0,3.0]|\n",
      "|   4|   1|   9| [4.0,1.0,9.0]|\n",
      "+----+----+----+--------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "vectorAssembler = VectorAssembler(inputCols = ['col1', 'col2', 'col3'], outputCol = 'features')\n",
    "vcluster_df = vectorAssembler.transform(cluster_df)\n",
    "vcluster_df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [],
   "source": [
    "kmeans = KMeans().setK(3)\n",
    "kmeans = kmeans.setSeed(1)\n",
    "kmodel = kmeans.fit(vcluster_df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[array([35.88461538, 31.46153846, 34.42307692]),\n",
       " array([5.12, 5.84, 4.84]),\n",
       " array([80.        , 79.20833333, 78.29166667])]"
      ]
     },
     "execution_count": 31,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "centers = kmodel.clusterCenters()\n",
    "centers"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[array([5.12, 5.84, 4.84]),\n",
       " array([35.88461538, 31.46153846, 34.42307692]),\n",
       " array([80.        , 79.20833333, 78.29166667])]"
      ]
     },
     "execution_count": 32,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from pyspark.ml.clustering import BisectingKMeans\n",
    "bkmeans = BisectingKMeans().setK(3)\n",
    "bkmeans = bkmeans.setSeed(1)\n",
    "bkmodel = bkmeans.fit(vcluster_df)\n",
    "bkcenters = bkmodel.clusterCenters()\n",
    "bkcenters"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(_c0=5.1, _c1=3.5, _c2=1.4, _c3=0.2, _c4='Iris-setosa')]"
      ]
     },
     "execution_count": 33,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from pyspark.sql.functions import *\n",
    "from pyspark.ml.feature import StringIndexer\n",
    "\n",
    "iris_df = spark.read.csv('iris.data.txt', inferSchema=True)\n",
    "iris_df.take(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(sepal_length=5.1, sepal_width=3.5, petal_length=1.4, petal_width=0.2, species='Iris-setosa')]"
      ]
     },
     "execution_count": 34,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "iris_df = iris_df.select(col('_c0').alias('sepal_length'), \n",
    "                        col('_c1').alias('sepal_width'),\n",
    "                        col('_c2').alias('petal_length'), \n",
    "                        col('_c3').alias('petal_width'), \n",
    "                        col('_c4').alias('species')\n",
    "                        )\n",
    "iris_df.take(1)                 "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(sepal_length=5.1, sepal_width=3.5, petal_length=1.4, petal_width=0.2, species='Iris-setosa', features=DenseVector([5.1, 3.5, 1.4, 0.2]))]"
      ]
     },
     "execution_count": 35,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "vectorAssembler = VectorAssembler(inputCols = ['sepal_length', 'sepal_width', 'petal_length', 'petal_width'], outputCol='features')\n",
    "\n",
    "viris_df = vectorAssembler.transform(iris_df)\n",
    "viris_df.take(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------+-----------+------------+-----------+-----------+-----------------+-----+\n",
      "|sepal_length|sepal_width|petal_length|petal_width|    species|         features|label|\n",
      "+------------+-----------+------------+-----------+-----------+-----------------+-----+\n",
      "|         5.1|        3.5|         1.4|        0.2|Iris-setosa|[5.1,3.5,1.4,0.2]|  0.0|\n",
      "|         4.9|        3.0|         1.4|        0.2|Iris-setosa|[4.9,3.0,1.4,0.2]|  0.0|\n",
      "+------------+-----------+------------+-----------+-----------+-----------------+-----+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "indexer = StringIndexer(inputCol = 'species', outputCol='label')\n",
    "iviris_df = indexer.fit(viris_df).transform(viris_df)\n",
    "iviris_df.show(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[sepal_length: double, sepal_width: double, petal_length: double, petal_width: double, species: string, features: vector, label: double]"
      ]
     },
     "execution_count": 37,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "iviris_df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(sepal_length=5.1, sepal_width=3.5, petal_length=1.4, petal_width=0.2, species='Iris-setosa', features=DenseVector([5.1, 3.5, 1.4, 0.2]), label=0.0)]"
      ]
     },
     "execution_count": 38,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "iviris_df.take(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.classification import NaiveBayes\n",
    "from pyspark.ml.evaluation import MulticlassClassificationEvaluator\n",
    "\n",
    "splits = iviris_df.randomSplit([0.6, 0.4], 1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {},
   "outputs": [],
   "source": [
    "train_df = splits[0]\n",
    "test_df = splits[1]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "92"
      ]
     },
     "execution_count": 41,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "train_df.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "58"
      ]
     },
     "execution_count": 42,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "test_df.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {},
   "outputs": [],
   "source": [
    "nb = NaiveBayes(modelType='multinomial')\n",
    "nbmodel = nb.fit(train_df)\n",
    "\n",
    "predictions_df = nbmodel.transform(test_df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(sepal_length=4.5, sepal_width=2.3, petal_length=1.3, petal_width=0.3, species='Iris-setosa', features=DenseVector([4.5, 2.3, 1.3, 0.3]), label=0.0, rawPrediction=DenseVector([-10.3605, -11.0141, -11.7112]), probability=DenseVector([0.562, 0.2924, 0.1456]), prediction=0.0)]"
      ]
     },
     "execution_count": 44,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "predictions_df.take(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.5862068965517241"
      ]
     },
     "execution_count": 45,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "evaluator = MulticlassClassificationEvaluator(labelCol = 'label', predictionCol='prediction', metricName='accuracy')\n",
    "nbaccuracy = evaluator.evaluate(predictions_df)\n",
    "nbaccuracy"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.9482758620689655"
      ]
     },
     "execution_count": 46,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from pyspark.ml.classification import MultilayerPerceptronClassifier\n",
    "\n",
    "layers = [4, 5, 5, 3]\n",
    "mlp = MultilayerPerceptronClassifier(layers=layers, seed=1)\n",
    "mlp_model = mlp.fit(train_df)\n",
    "mlp_predictions = mlp_model.transform(test_df)\n",
    "mlp_evaluator = MulticlassClassificationEvaluator(metricName='accuracy')\n",
    "mlp_accuracy = mlp_evaluator.evaluate(mlp_predictions)\n",
    "mlp_accuracy"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.9310344827586207"
      ]
     },
     "execution_count": 48,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from pyspark.ml.classification import DecisionTreeClassifier\n",
    "\n",
    "dt = DecisionTreeClassifier(labelCol='label', featuresCol='features')\n",
    "dt_model = dt.fit(train_df)\n",
    "dt_predictions = dt_model.transform(test_df)\n",
    "dt_evaluator = MulticlassClassificationEvaluator(labelCol='label', predictionCol='prediction', metricName='accuracy')\n",
    "dt_accuracy = dt_evaluator.evaluate(dt_predictions)\n",
    "dt_accuracy"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+-----+-------+-----+------+\n",
      "|   AT|    V|     AP|   RH|    PE|\n",
      "+-----+-----+-------+-----+------+\n",
      "|14.96|41.76|1024.07|73.17|463.26|\n",
      "|25.18|62.96|1020.04|59.08|444.37|\n",
      "| 5.11| 39.4|1012.16|92.14|488.56|\n",
      "|20.86|57.32|1010.24|76.64|446.48|\n",
      "|10.82| 37.5|1009.23|96.62| 473.9|\n",
      "|26.27|59.44|1012.23|58.77|443.67|\n",
      "|15.89|43.96|1014.02|75.24|467.35|\n",
      "| 9.48|44.71|1019.12|66.43|478.42|\n",
      "|14.64| 45.0|1021.78|41.25|475.98|\n",
      "|11.74|43.56|1015.14|70.72| 477.5|\n",
      "|17.99|43.72|1008.64|75.04|453.02|\n",
      "|20.14|46.93|1014.66|64.22|453.99|\n",
      "|24.34| 73.5|1011.31|84.15|440.29|\n",
      "|25.71|58.59|1012.77|61.83|451.28|\n",
      "|26.19|69.34|1009.48|87.59|433.99|\n",
      "|21.42|43.79|1015.76|43.08|462.19|\n",
      "|18.21| 45.0|1022.86|48.84|467.54|\n",
      "|11.04|41.74| 1022.6|77.51| 477.2|\n",
      "|14.45|52.75|1023.97|63.59|459.85|\n",
      "|13.97|38.47|1015.15|55.28| 464.3|\n",
      "+-----+-----+-------+-----+------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.ml.regression import LinearRegression\n",
    "pp_df = spark.read.csv('power_plant.csv', header=True, inferSchema=True)\n",
    "pp_df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[AT: double, V: double, AP: double, RH: double, PE: double]"
      ]
     },
     "execution_count": 52,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pp_df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(AT=14.96, V=41.76, AP=1024.07, RH=73.17, PE=463.26, features=DenseVector([14.96, 41.76, 1024.07, 73.17]))]"
      ]
     },
     "execution_count": 53,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "vectorAssembler = VectorAssembler(inputCols = ['AT', 'V', 'AP', 'RH'], outputCol = 'features')\n",
    "vpp_df = vectorAssembler.transform(pp_df)\n",
    "vpp_df.take(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+-----+-------+-----+------+--------------------+\n",
      "|   AT|    V|     AP|   RH|    PE|            features|\n",
      "+-----+-----+-------+-----+------+--------------------+\n",
      "|14.96|41.76|1024.07|73.17|463.26|[14.96,41.76,1024...|\n",
      "|25.18|62.96|1020.04|59.08|444.37|[25.18,62.96,1020...|\n",
      "| 5.11| 39.4|1012.16|92.14|488.56|[5.11,39.4,1012.1...|\n",
      "|20.86|57.32|1010.24|76.64|446.48|[20.86,57.32,1010...|\n",
      "|10.82| 37.5|1009.23|96.62| 473.9|[10.82,37.5,1009....|\n",
      "|26.27|59.44|1012.23|58.77|443.67|[26.27,59.44,1012...|\n",
      "|15.89|43.96|1014.02|75.24|467.35|[15.89,43.96,1014...|\n",
      "| 9.48|44.71|1019.12|66.43|478.42|[9.48,44.71,1019....|\n",
      "|14.64| 45.0|1021.78|41.25|475.98|[14.64,45.0,1021....|\n",
      "|11.74|43.56|1015.14|70.72| 477.5|[11.74,43.56,1015...|\n",
      "|17.99|43.72|1008.64|75.04|453.02|[17.99,43.72,1008...|\n",
      "|20.14|46.93|1014.66|64.22|453.99|[20.14,46.93,1014...|\n",
      "|24.34| 73.5|1011.31|84.15|440.29|[24.34,73.5,1011....|\n",
      "|25.71|58.59|1012.77|61.83|451.28|[25.71,58.59,1012...|\n",
      "|26.19|69.34|1009.48|87.59|433.99|[26.19,69.34,1009...|\n",
      "|21.42|43.79|1015.76|43.08|462.19|[21.42,43.79,1015...|\n",
      "|18.21| 45.0|1022.86|48.84|467.54|[18.21,45.0,1022....|\n",
      "|11.04|41.74| 1022.6|77.51| 477.2|[11.04,41.74,1022...|\n",
      "|14.45|52.75|1023.97|63.59|459.85|[14.45,52.75,1023...|\n",
      "|13.97|38.47|1015.15|55.28| 464.3|[13.97,38.47,1015...|\n",
      "+-----+-----+-------+-----+------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "vpp_df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "ename": "NameError",
     "evalue": "name 'LinearRegression' is not defined",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mNameError\u001b[0m                                 Traceback (most recent call last)",
      "\u001b[0;32m<ipython-input-1-21204b12720d>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m()\u001b[0m\n\u001b[0;32m----> 1\u001b[0;31m \u001b[0mlr\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mLinearRegression\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mfeaturesCol\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0;34m'features'\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mlabelCol\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;34m'PE'\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m      2\u001b[0m \u001b[0mlr_model\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mlr\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mfit\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mvpp_df\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m      3\u001b[0m \u001b[0mprint\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m\"Coefficients: %s\"\u001b[0m \u001b[0;34m%\u001b[0m \u001b[0mstr\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mlr_Model\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mcoefficients\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;31mNameError\u001b[0m: name 'LinearRegression' is not defined"
     ]
    }
   ],
   "source": [
    "lr = LinearRegression(featuresCol = 'features', labelCol='PE')\n",
    "lr_model = lr.fit(vpp_df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "454.6092744523414"
      ]
     },
     "execution_count": 56,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "lr_model.intercept"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 57,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "4.557126016749488"
      ]
     },
     "execution_count": 57,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "lr_model.summary.rootMeanSquaredError"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 58,
   "metadata": {},
   "outputs": [],
   "source": [
    "lr_model.save('lr1.model')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 60,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(AT=14.96, V=41.76, AP=1024.07, RH=73.17, PE=463.26)]"
      ]
     },
     "execution_count": 60,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from pyspark.ml.regression import DecisionTreeRegressor\n",
    "from pyspark.ml.evaluation import RegressionEvaluator\n",
    "\n",
    "pp_df = spark.read.csv('power_plant.csv', header=True, inferSchema=True)\n",
    "pp_df.take(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 62,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(AT=14.96, V=41.76, AP=1024.07, RH=73.17, PE=463.26, features=DenseVector([14.96, 41.76, 1024.07, 73.17]))]"
      ]
     },
     "execution_count": 62,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "vectorAssembler = VectorAssembler(inputCols = ['AT', 'V', 'AP', 'RH'], outputCol = 'features')\n",
    "vpp_df = vectorAssembler.transform(pp_df)\n",
    "vpp_df.take(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 63,
   "metadata": {},
   "outputs": [],
   "source": [
    "splits = vpp_df.randomSplit([0.7, 0.3])\n",
    "train_df = splits[0]\n",
    "test_df = splits[1]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 67,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "4.571310350538227"
      ]
     },
     "execution_count": 67,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "dt = DecisionTreeRegressor(featuresCol ='features', labelCol = 'PE')\n",
    "dt_model = dt.fit(train_df)\n",
    "dt_predictions = dt_model.transform(test_df)\n",
    "dt_evaluator = RegressionEvaluator(labelCol = 'PE', predictionCol = 'prediction', metricName = 'rmse')\n",
    "rmse = dt_evaluator.evaluate(dt_predictions)\n",
    "rmse"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 68,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "4.137733340128398"
      ]
     },
     "execution_count": 68,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from pyspark.ml.regression import GBTRegressor\n",
    "gbt = GBTRegressor(featuresCol = 'features', labelCol = 'PE')\n",
    "gbt_model = gbt.fit(train_df)\n",
    "gbt_predictions = gbt_model.transform(test_df)\n",
    "gbt_evaluator = RegressionEvaluator(labelCol = 'PE', predictionCol = 'prediction', metricName = 'rmse')\n",
    "rmse = gbt_evaluator.evaluate(gbt_predictions)\n",
    "rmse"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "conda_python3",
   "language": "python",
   "name": "conda_python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}