Spark_ML/sparkml/3_creating_features_quiz.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Creating Features Quiz\n",
    "Use this Jupyter notebook to find the answers to the quiz in the previous section. There is an answer key in the next part of the lesson."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "\n",
    "# TODOS: \n",
    "# 1) import any other libraries you might need\n",
    "# 2) run the cells below to read dataset and build body length feature\n",
    "# 3) write code to answer the quiz questions \n",
    "from pyspark.ml.feature import RegexTokenizer, VectorAssembler, Normalizer, StandardScaler, MinMaxScaler\n",
    "from pyspark.sql.functions import col, lit, concat, udf\n",
    "from pyspark.sql.types import IntegerType"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "WARNING: An illegal reflective access operation has occurred\n",
      "WARNING: Illegal reflective access by org.apache.spark.unsafe.Platform (file:/usr/local/lib/python3.9/site-packages/pyspark/jars/spark-unsafe_2.12-3.2.0.jar) to constructor java.nio.DirectByteBuffer(long,int)\n",
      "WARNING: Please consider reporting this to the maintainers of org.apache.spark.unsafe.Platform\n",
      "WARNING: Use --illegal-access=warn to enable warnings of further illegal reflective access operations\n",
      "WARNING: All illegal access operations will be denied in a future release\n",
      "Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties\n",
      "Setting default log level to \"WARN\".\n",
      "To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).\n",
      "22/01/13 19:29:13 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n",
      "22/01/13 19:29:14 WARN Utils: Service 'SparkUI' could not bind on port 4040. Attempting port 4041.\n",
      "22/01/13 19:29:14 WARN Utils: Service 'SparkUI' could not bind on port 4041. Attempting port 4042.\n",
      "22/01/13 19:29:14 WARN Utils: Service 'SparkUI' could not bind on port 4042. Attempting port 4043.\n"
     ]
    }
   ],
   "source": [
    "spark = SparkSession.builder \\\n",
    "    .master(\"local\") \\\n",
    "    .appName(\"Creating Features\") \\\n",
    "    .getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Read Dataset"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "stack_overflow_data = 'Train_onetag_small.json'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "DataFrame[Body: string, Id: bigint, Tags: string, Title: string, oneTag: string]"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df = spark.read.json(stack_overflow_data)\n",
    "df.persist()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Build Body Length Feature"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "regexTokenizer = RegexTokenizer(inputCol=\"Body\", outputCol=\"words\", pattern=\"\\\\W\")\n",
    "df = regexTokenizer.transform(df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "body_length = udf(lambda x: len(x), IntegerType())\n",
    "df = df.withColumn(\"BodyLength\", body_length(df.words))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Traceback (most recent call last):                                  (0 + 1) / 1]\n",
      "  File \"/usr/local/lib/python3.9/site-packages/pyspark/python/lib/pyspark.zip/pyspark/daemon.py\", line 186, in manager\n",
      "  File \"/usr/local/lib/python3.9/site-packages/pyspark/python/lib/pyspark.zip/pyspark/daemon.py\", line 74, in worker\n",
      "  File \"/usr/local/lib/python3.9/site-packages/pyspark/python/lib/pyspark.zip/pyspark/worker.py\", line 663, in main\n",
      "    if read_int(infile) == SpecialLengths.END_OF_STREAM:\n",
      "  File \"/usr/local/lib/python3.9/site-packages/pyspark/python/lib/pyspark.zip/pyspark/serializers.py\", line 564, in read_int\n",
      "    raise EOFError\n",
      "EOFError\n",
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "Row(Body=\"<p>I'd like to check if an uploaded file is an image file (e.g png, jpg, jpeg, gif, bmp) or another file. The problem is that I'm using Uploadify to upload the files, which changes the mime type and gives a 'text/octal' or something as the mime type, no matter which file type you upload.</p>\\n\\n<p>Is there a way to check if the uploaded file is an image apart from checking the file extension using PHP?</p>\\n\", Id=1, Tags='php image-processing file-upload upload mime-types', Title='How to check if an uploaded file is an image without mime type?', oneTag='php', words=['p', 'i', 'd', 'like', 'to', 'check', 'if', 'an', 'uploaded', 'file', 'is', 'an', 'image', 'file', 'e', 'g', 'png', 'jpg', 'jpeg', 'gif', 'bmp', 'or', 'another', 'file', 'the', 'problem', 'is', 'that', 'i', 'm', 'using', 'uploadify', 'to', 'upload', 'the', 'files', 'which', 'changes', 'the', 'mime', 'type', 'and', 'gives', 'a', 'text', 'octal', 'or', 'something', 'as', 'the', 'mime', 'type', 'no', 'matter', 'which', 'file', 'type', 'you', 'upload', 'p', 'p', 'is', 'there', 'a', 'way', 'to', 'check', 'if', 'the', 'uploaded', 'file', 'is', 'an', 'image', 'apart', 'from', 'checking', 'the', 'file', 'extension', 'using', 'php', 'p'], BodyLength=83)"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Question 1\n",
    "Select the question with Id = 1112. How many words does its body contain (check the BodyLength column)?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+\n",
      "|BodyLength|\n",
      "+----------+\n",
      "|        63|\n",
      "+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# TODO: write your code to answer question 1\n",
    "df.select(\"BodyLength\").where(df.Id == 1112).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Question 2\n",
    "Create a new column that concatenates the question title and body. Apply the same functions we used before to compute the number of words in this combined column. What's the value in this new column for Id = 5123?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "# TODO: write your code to answer question 2\n",
    "# Make a new col named \"Desc\" that concated by col \"Title\" and \"Body\"\n",
    "# df.select(concat(\"Title\", lit(' '), \"Body\")).show()\n",
    "df = df.withColumn(\"Desc\", concat(\"Title\", lit(' '), \"Body\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Tokenize words2\n",
    "regexTokenizer = RegexTokenizer(inputCol=\"Desc\", outputCol=\"words2\", pattern=\"\\\\W\")\n",
    "df = regexTokenizer.transform(df)\n",
    "# Count words2\n",
    "df = df.withColumn(\"DescLength\", body_length(df.words2))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+\n",
      "|DescLength|\n",
      "+----------+\n",
      "|       135|\n",
      "+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(\"DescLength\").where(df.Id == 5123).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Create a Vector\n",
    "Create a vector from the combined Title + Body length column. In the next few questions, you'll try different normalizer/scaler methods on this new column."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "# TODO: write your code to create this vector\n",
    "assembler = VectorAssembler(inputCols=[\"DescLength\"], outputCol=\"DescVec\")\n",
    "df = assembler.transform(df)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Question 3\n",
    "Using the Normalizer method what's the normalized value for question Id = 512?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "# TODO: write your code to answer question 3\n",
    "normalizer= Normalizer(inputCol=\"DescVec\", outputCol=\"DescVecNormalizer\")\n",
    "df = normalizer.transform(df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 4:>                                                          (0 + 1) / 1]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------+\n",
      "|DescVecNormalizer|\n",
      "+-----------------+\n",
      "|            [1.0]|\n",
      "+-----------------+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "df.select(\"DescVecNormalizer\").where(df.Id == 512).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Question 4\n",
    "Using the StandardScaler method (scaling both the mean and the standard deviation) what's the normalized value for question Id = 512?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "# TODO: write your code to answer question 4\n",
    "standardScaler = StandardScaler(inputCol=\"DescVec\", outputCol=\"DescStandardScalar\", withMean=True, withStd=True)\n",
    "standardScalerModel = standardScaler.fit(df)\n",
    "df = standardScalerModel.transform(df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 8:>                                                          (0 + 1) / 1]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+\n",
      "|  DescStandardScalar|\n",
      "+--------------------+\n",
      "|[-0.6417314460998...|\n",
      "+--------------------+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "df.select(\"DescStandardScalar\").where(df.Id == 512).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Question 5\n",
    "Using the MinMAxScaler method what's the normalized value for question Id = 512?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "# TODO: write your code to answer question 5\n",
    "minMaxScaler = MinMaxScaler(inputCol=\"DescVec\", outputCol=\"DescMinMaxScalar\")\n",
    "minMaxScalerModel = minMaxScaler.fit(df)\n",
    "df = minMaxScalerModel.transform(df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 12:>                                                         (0 + 1) / 1]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+\n",
      "|    DescMinMaxScalar|\n",
      "+--------------------+\n",
      "|[0.00624833820792...|\n",
      "+--------------------+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "df.select(\"DescMinMaxScalar\").where(df.Id == 512).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "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.9.9"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}