Apache project 3(1).ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Assignment 3\n",
    "\n",
    "Welcome to Assignment 3. This will be even more fun. Now we will calculate statistical measures on the test data you have created.\n",
    "\n",
    "YOU ARE NOT ALLOWED TO USE ANY OTHER 3RD PARTY LIBRARIES LIKE PANDAS. PLEASE ONLY MODIFY CONTENT INSIDE THE FUNCTION SKELETONS\n",
    "Please read why: https://www.coursera.org/learn/exploring-visualizing-iot-data/discussions/weeks/3/threads/skjCbNgeEeapeQ5W6suLkA\n",
    ". Just make sure you hit the play button on each cell from top to down. There are seven functions you have to implement. Please also make sure than on each change on a function you hit the play button again on the corresponding cell to make it available to the rest of this notebook.\n",
    "Please also make sure to only implement the function bodies and DON'T add any additional code outside functions since this might confuse the autograder.\n",
    "\n",
    "So the function below is used to make it easy for you to create a data frame from a cloudant data frame using the so called \"DataSource\" which is some sort of a plugin which allows ApacheSpark to use different data sources.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "All functions can be implemented using DataFrames, ApacheSparkSQL or RDDs. We are only interested in the result. You are given the reference to the data frame in the \"df\" parameter and in case you want to use SQL just use the \"spark\" parameter which is a reference to the global SparkSession object. Finally if you want to use RDDs just use \"df.rdd\" for obtaining a reference to the underlying RDD object. \n",
    "\n",
    "Let's start with the first function. Please calculate the minimal temperature for the test data set you have created. We've provided a little skeleton for you in case you want to use SQL. You can use this skeleton for all subsequent functions. Everything can be implemented using SQL only if you like."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "def minTemperature(df,spark):\n",
    "    df = spark.sql(\"select * from washing where temperature IS NOT NULL\")\n",
    "    row1 = df.agg({\"temperature\": \"min\"}).collect()[0]\n",
    "    min_ = row1[\"min(temperature)\"]\n",
    "    #some reference: https://spark.apache.org/docs/latest/api/python/pyspark.sql.html#pyspark.sql.DataFrame\n",
    "    return min_"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Please now do the same for the mean of the temperature"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "def meanTemperature(df,spark):\n",
    "    df = spark.sql(\"select * from washing where temperature IS NOT NULL\")\n",
    "    mean_ = spark.sql(\"select avg(temperature) from washing\").collect()[0][0] # selecting 1st row and first column\n",
    "    return mean_"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Please now do the same for the maximum of the temperature"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "def maxTemperature(df,spark):\n",
    "    df = spark.sql(\"select * from washing where temperature IS NOT NULL\")\n",
    "    row1 = df.agg({\"temperature\": \"max\"}).collect()[0]\n",
    "    max_ = row1[\"max(temperature)\"]\n",
    "    #some reference: https://spark.apache.org/docs/latest/api/python/pyspark.sql.html#pyspark.sql.DataFrame\n",
    "    return max_"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Please now do the same for the standard deviation of the temperature"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "def sdTemperature(df,spark):\n",
    "    df = spark.sql(\"select * from washing where temperature IS NOT NULL\")\n",
    "    std_ = spark.sql(\"select std(temperature) from washing\").collect()[0][0]\n",
    "    return std_"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Please now do the same for the skew of the temperature. Since the SQL statement for this is a bit more complicated we've provided a skeleton for you. You have to insert custom code at four position in order to make the function work. Alternatively you can also remove everything and implement if on your own. Note that we are making use of two previously defined functions, so please make sure they are correct. Also note that we are making use of python's string formatting capabilitis where the results of the two function calls to \"meanTemperature\" and \"sdTemperature\" are inserted at the \"%s\" symbols in the SQL string."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [],
   "source": [
    "def skewTemperature(df,spark):    \n",
    "    return spark.sql(\"\"\"\n",
    "    SELECT \n",
    "    (\n",
    "        1/(count(temperature))\n",
    "    ) *\n",
    "    (SUM (\n",
    "    POWER(temperature-%s,3))/POWER(%s,3)\n",
    "    )\n",
    "\n",
    "    as sktemperature from washing where temperature IS NOT NULL\n",
    "                    \"\"\" %(meanTemperature(df,spark),sdTemperature(df,spark))).first().sktemperature"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Kurtosis is the 4th statistical moment, so if you are smart you can make use of the code for skew which is the 3rd statistical moment. Actually only two things are different."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [],
   "source": [
    "def kurtosisTemperature(df,spark):    \n",
    "    return spark.sql(\"\"\"\n",
    "    SELECT \n",
    "    (\n",
    "        1/(count(temperature))\n",
    "    ) *\n",
    "    (SUM (\n",
    "    POWER(temperature-%s,4))/POWER(%s,4)\n",
    "    )\n",
    "\n",
    "    as sktemperature from washing where temperature IS NOT NULL\n",
    "                    \"\"\" %(meanTemperature(df,spark),sdTemperature(df,spark))).first().sktemperature"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Just a hint. This can be solved easily using SQL as well, but as shown in the lecture also using RDDs."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {},
   "outputs": [],
   "source": [
    "def correlationTemperatureHardness(df,spark):\n",
    "    from pyspark.mllib.stat import Statistics\n",
    "    d = Statistics.corr((spark.sql(\"SELECT temperature,hardness  FROM washing where temperature IS NOT NULL and hardness IS NOT NULL\").rdd.map(lambda row: (row[0],row[1]))))\n",
    "    return d.item(1)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### PLEASE DON'T REMOVE THIS BLOCK - THE FOLLOWING CODE IS NOT GRADED\n",
    "#axx\n",
    "### PLEASE DON'T REMOVE THIS BLOCK - THE FOLLOWING CODE IS NOT GRADED"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now it is time to grab a PARQUET file and create a dataframe out of it. Using SparkSQL you can handle it like a database. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "--2019-06-23 13:39:09--  https://github.com/IBM/coursera/blob/master/coursera_ds/washing.parquet?raw=true\n",
      "Resolving github.com (github.com)... 140.82.118.3\n",
      "Connecting to github.com (github.com)|140.82.118.3|:443... connected.\n",
      "HTTP request sent, awaiting response... 302 Found\n",
      "Location: https://github.com/IBM/coursera/raw/master/coursera_ds/washing.parquet [following]\n",
      "--2019-06-23 13:39:09--  https://github.com/IBM/coursera/raw/master/coursera_ds/washing.parquet\n",
      "Reusing existing connection to github.com:443.\n",
      "HTTP request sent, awaiting response... 302 Found\n",
      "Location: https://raw.githubusercontent.com/IBM/coursera/master/coursera_ds/washing.parquet [following]\n",
      "--2019-06-23 13:39:10--  https://raw.githubusercontent.com/IBM/coursera/master/coursera_ds/washing.parquet\n",
      "Resolving raw.githubusercontent.com (raw.githubusercontent.com)... 151.101.16.133\n",
      "Connecting to raw.githubusercontent.com (raw.githubusercontent.com)|151.101.16.133|:443... connected.\n",
      "HTTP request sent, awaiting response... 200 OK\n",
      "Length: 112048 (109K) [application/octet-stream]\n",
      "Saving to: 'washing.parquet?raw=true'\n",
      "\n",
      "washing.parquet?raw 100%[===================>] 109.42K  --.-KB/s    in 0.009s  \n",
      "\n",
      "2019-06-23 13:39:10 (11.9 MB/s) - 'washing.parquet?raw=true' saved [112048/112048]\n",
      "\n"
     ]
    }
   ],
   "source": [
    "!wget https://github.com/IBM/coursera/blob/master/coursera_ds/washing.parquet?raw=true\n",
    "!mv washing.parquet?raw=true washing.parquet"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+--------------------+-----+--------+----------+---------+--------+-----+-----------+-------------+-------+\n",
      "|                 _id|                _rev|count|flowrate|fluidlevel|frequency|hardness|speed|temperature|           ts|voltage|\n",
      "+--------------------+--------------------+-----+--------+----------+---------+--------+-----+-----------+-------------+-------+\n",
      "|0d86485d0f88d1f9d...|1-57940679fb8a713...|    4|      11|acceptable|     null|      77| null|        100|1547808723923|   null|\n",
      "|0d86485d0f88d1f9d...|1-15ff3a0b304d789...|    2|    null|      null|     null|    null| 1046|       null|1547808729917|   null|\n",
      "|0d86485d0f88d1f9d...|1-97c2742b68c7b07...|    4|    null|      null|       71|    null| null|       null|1547808731918|    236|\n",
      "|0d86485d0f88d1f9d...|1-eefb903dbe45746...|   19|      11|acceptable|     null|      75| null|         86|1547808738999|   null|\n",
      "|0d86485d0f88d1f9d...|1-5f68b4c72813c25...|    7|    null|      null|       75|    null| null|       null|1547808740927|    235|\n",
      "|0d86485d0f88d1f9d...|1-cd4b6c57ddbe77e...|    5|    null|      null|     null|    null| 1014|       null|1547808744923|   null|\n",
      "|0d86485d0f88d1f9d...|1-a35b25b5bf43aaf...|   32|      11|acceptable|     null|      73| null|         84|1547808752028|   null|\n",
      "|0d86485d0f88d1f9d...|1-b717f7289a8476d...|   48|      11|acceptable|     null|      79| null|         84|1547808768065|   null|\n",
      "|0d86485d0f88d1f9d...|1-c2f1f8fcf178b2f...|   18|    null|      null|       73|    null| null|       null|1547808773944|    228|\n",
      "|0d86485d0f88d1f9d...|1-15033dd9eebb4a8...|   59|      11|acceptable|     null|      72| null|         96|1547808779093|   null|\n",
      "|0d86485d0f88d1f9d...|1-753dae825f9a6c2...|   62|      11|acceptable|     null|      73| null|         88|1547808782113|   null|\n",
      "|0d86485d0f88d1f9d...|1-b168089f44f03f0...|   13|    null|      null|     null|    null| 1097|       null|1547808784940|   null|\n",
      "|0d86485d0f88d1f9d...|1-403b687c6be0dea...|   23|    null|      null|       80|    null| null|       null|1547808788955|    236|\n",
      "|0d86485d0f88d1f9d...|1-195551e0455a24b...|   72|      11|acceptable|     null|      77| null|         87|1547808792134|   null|\n",
      "|0d86485d0f88d1f9d...|1-060a39fc6c2ddee...|   26|    null|      null|       62|    null| null|       null|1547808797959|    233|\n",
      "|0d86485d0f88d1f9d...|1-2234514bffee465...|   27|    null|      null|       61|    null| null|       null|1547808800960|    226|\n",
      "|0d86485d0f88d1f9d...|1-4265898bb401db0...|   82|      11|acceptable|     null|      79| null|         96|1547808802154|   null|\n",
      "|0d86485d0f88d1f9d...|1-2fbf7ca9a0425a0...|   94|      11|acceptable|     null|      73| null|         90|1547808814186|   null|\n",
      "|0d86485d0f88d1f9d...|1-203c0ee6d7fbd21...|   97|      11|acceptable|     null|      77| null|         88|1547808817190|   null|\n",
      "|0d86485d0f88d1f9d...|1-47e1965db94fcab...|  104|      11|acceptable|     null|      75| null|         80|1547808824198|   null|\n",
      "+--------------------+--------------------+-----+--------+----------+---------+--------+-----+-----------+-------------+-------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df = spark.read.parquet('washing.parquet')\n",
    "df.createOrReplaceTempView('washing')\n",
    "df.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now it is time to connect to the object store and read a PARQUET file and create a dataframe out of it. We've created that data for you already. Using SparkSQL you can handle it like a database."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+--------------------+-----+--------+----------+---------+--------+-----+-----------+-------------+-------+\n",
      "|                 _id|                _rev|count|flowrate|fluidlevel|frequency|hardness|speed|temperature|           ts|voltage|\n",
      "+--------------------+--------------------+-----+--------+----------+---------+--------+-----+-----------+-------------+-------+\n",
      "|0d86485d0f88d1f9d...|1-57940679fb8a713...|    4|      11|acceptable|     null|      77| null|        100|1547808723923|   null|\n",
      "|0d86485d0f88d1f9d...|1-15ff3a0b304d789...|    2|    null|      null|     null|    null| 1046|       null|1547808729917|   null|\n",
      "|0d86485d0f88d1f9d...|1-97c2742b68c7b07...|    4|    null|      null|       71|    null| null|       null|1547808731918|    236|\n",
      "|0d86485d0f88d1f9d...|1-eefb903dbe45746...|   19|      11|acceptable|     null|      75| null|         86|1547808738999|   null|\n",
      "|0d86485d0f88d1f9d...|1-5f68b4c72813c25...|    7|    null|      null|       75|    null| null|       null|1547808740927|    235|\n",
      "|0d86485d0f88d1f9d...|1-cd4b6c57ddbe77e...|    5|    null|      null|     null|    null| 1014|       null|1547808744923|   null|\n",
      "|0d86485d0f88d1f9d...|1-a35b25b5bf43aaf...|   32|      11|acceptable|     null|      73| null|         84|1547808752028|   null|\n",
      "|0d86485d0f88d1f9d...|1-b717f7289a8476d...|   48|      11|acceptable|     null|      79| null|         84|1547808768065|   null|\n",
      "|0d86485d0f88d1f9d...|1-c2f1f8fcf178b2f...|   18|    null|      null|       73|    null| null|       null|1547808773944|    228|\n",
      "|0d86485d0f88d1f9d...|1-15033dd9eebb4a8...|   59|      11|acceptable|     null|      72| null|         96|1547808779093|   null|\n",
      "|0d86485d0f88d1f9d...|1-753dae825f9a6c2...|   62|      11|acceptable|     null|      73| null|         88|1547808782113|   null|\n",
      "|0d86485d0f88d1f9d...|1-b168089f44f03f0...|   13|    null|      null|     null|    null| 1097|       null|1547808784940|   null|\n",
      "|0d86485d0f88d1f9d...|1-403b687c6be0dea...|   23|    null|      null|       80|    null| null|       null|1547808788955|    236|\n",
      "|0d86485d0f88d1f9d...|1-195551e0455a24b...|   72|      11|acceptable|     null|      77| null|         87|1547808792134|   null|\n",
      "|0d86485d0f88d1f9d...|1-060a39fc6c2ddee...|   26|    null|      null|       62|    null| null|       null|1547808797959|    233|\n",
      "|0d86485d0f88d1f9d...|1-2234514bffee465...|   27|    null|      null|       61|    null| null|       null|1547808800960|    226|\n",
      "|0d86485d0f88d1f9d...|1-4265898bb401db0...|   82|      11|acceptable|     null|      79| null|         96|1547808802154|   null|\n",
      "|0d86485d0f88d1f9d...|1-2fbf7ca9a0425a0...|   94|      11|acceptable|     null|      73| null|         90|1547808814186|   null|\n",
      "|0d86485d0f88d1f9d...|1-203c0ee6d7fbd21...|   97|      11|acceptable|     null|      77| null|         88|1547808817190|   null|\n",
      "|0d86485d0f88d1f9d...|1-47e1965db94fcab...|  104|      11|acceptable|     null|      75| null|         80|1547808824198|   null|\n",
      "+--------------------+--------------------+-----+--------+----------+---------+--------+-----+-----------+-------------+-------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "import ibmos2spark\n",
    "\n",
    "# @hidden_cell\n",
    "credentials = {\n",
    "    'endpoint': 'https://s3-api.us-geo.objectstorage.service.networklayer.com',\n",
    "    'api_key': 'PUJMZf9PLqN4y-6NUtVlEuq6zFoWhfuecFVMYLBrkxrT',\n",
    "    'service_id': 'iam-ServiceId-9cd8e66e-3bb4-495a-807a-588692cca4d0',\n",
    "    'iam_service_endpoint': 'https://iam.bluemix.net/oidc/token'}\n",
    "\n",
    "configuration_name = 'os_b0f1407510994fd1b793b85137baafb8_configs'\n",
    "cos = ibmos2spark.CloudObjectStorage(sc, credentials, configuration_name, 'bluemix_cos')\n",
    "\n",
    "from pyspark.sql import SparkSession\n",
    "spark = SparkSession.builder.getOrCreate()\n",
    "# Since JSON data can be semi-structured and contain additional metadata, it is possible that you might face issues with the DataFrame layout.\n",
    "# Please read the documentation of 'SparkSession.read()' to learn more about the possibilities to adjust the data loading.\n",
    "# PySpark documentation: http://spark.apache.org/docs/2.0.2/api/python/pyspark.sql.html#pyspark.sql.DataFrameReader.json\n",
    "\n",
    "df = spark.read.parquet(cos.url('washing.parquet', 'courseradsnew-donotdelete-pr-1hffrnl2pprwut'))\n",
    "df.createOrReplaceTempView('washing')\n",
    "df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "80"
      ]
     },
     "execution_count": 31,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "minTemperature(df,spark)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "90.03800298062593"
      ]
     },
     "execution_count": 32,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "meanTemperature(df,spark)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "100"
      ]
     },
     "execution_count": 33,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "maxTemperature(df,spark)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "6.1007610586219725"
      ]
     },
     "execution_count": 34,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sdTemperature(df,spark)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.01039837458333609"
      ]
     },
     "execution_count": 35,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "skewTemperature(df,spark)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1.773427150876989"
      ]
     },
     "execution_count": 36,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "kurtosisTemperature(df,spark)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.017754069047439067"
      ]
     },
     "execution_count": 42,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "correlationTemperatureHardness(df,spark)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Congratulations, you are done, please download this notebook as python file using the export function and submit is to the gader using the filename \"assignment3.1.py\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "array([[1.        , 0.01775407]])"
      ]
     },
     "execution_count": 39,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "d[0:1]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.017754069047439067"
      ]
     },
     "execution_count": 40,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "d.item(1)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3.6 with Spark",
   "language": "python3",
   "name": "python36"
  },
  "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.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}