Lambda_Architecture_using Spark_HDFS.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true,
    "jupyter": {
     "outputs_hidden": true
    },
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# Importing necessary libraries \n",
    "import json\n",
    "import pandas as pd\n",
    "from pyspark.sql import Row\n",
    "import pprint\n",
    "from pyspark.sql.functions import desc"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "jupyter": {
     "outputs_hidden": true
    }
   },
   "source": [
    "## Linux commands for the entire pipeline"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 1. Docker commands to bring up/ tear down/ check out the Docker cluster: \n",
    "    docker-compose up -d  \n",
    "    docker-compose down  \n",
    "    docker-compose ps -a   \n",
    "     \n",
    "    \n",
    "#### 2. Command to create the kafka topic with a partition and replication for kafka cluster using zookeeper as its cluster manager:   \n",
    "    docker-compose exec kafka kafka-topics --create --topic assessments --partitions 1 --replication-factor 1 --if-not-exists --zookeeper zookeeper:32181  \n",
    "    \n",
    "    \n",
    "#### 3. Command to publish the assessments json data to the kafka topic using kafkacat:  \n",
    "    docker-compose exec mids bash -c \"cat /w205/project-2-alicehua11/assessments-attempts-20180128-121051-nested.json | jq '.[]' -c | kafkacat -P -b kafka:29092 -t assessment && echo 'Produced assessment.'\"  \n",
    "    \n",
    "    \n",
    "#### 4. Command to bring up Jupyter notebook by creating a symbolic link in Spark container to /205 mount point:\n",
    "    docker-compose exec spark bash  \n",
    "    ln -s /w205 w205  \n",
    "    exit  \n",
    "    docker-compose exec spark env PYSPARK_DRIVER_PYTHON=jupyter PYSPARK_DRIVER_PYTHON_OPTS='notebook --no-browser --port 8888 --ip 0.0.0.0 --allow-root' pyspark\n",
    "    \n",
    "#### 5. Command to check out Hadoop HDFS parquet files  \n",
    "    docker-compose exec cloudera hadoop fs -ls /tmp/\n",
    "    "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Part 1: Create a data frame by subscribing to the kakfa topic"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": true,
    "jupyter": {
     "outputs_hidden": true
    },
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# Ask for the entire Kafka topic from beginning to end \n",
    "raw_assessments = spark.read.format(\"kafka\").option(\"kafka.bootstrap.servers\", \"kafka:29092\").option(\"subscribe\",\"assessment\").option(\"startingOffsets\", \"earliest\").option(\"endingOffsets\", \"latest\").load() "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[key: binary, value: binary, topic: string, partition: int, offset: bigint, timestamp: timestamp, timestampType: int]"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Cache the PySpark dataframe to persists the data in memory for subsequent actions without reevaluation\n",
    "raw_assessments.cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+--------------------+----------+---------+------+--------------------+-------------+\n",
      "| key|               value|     topic|partition|offset|           timestamp|timestampType|\n",
      "+----+--------------------+----------+---------+------+--------------------+-------------+\n",
      "|null|[7B 22 6B 65 65 6...|assessment|        0|     0|1969-12-31 23:59:...|            0|\n",
      "|null|[7B 22 6B 65 65 6...|assessment|        0|     1|1969-12-31 23:59:...|            0|\n",
      "|null|[7B 22 6B 65 65 6...|assessment|        0|     2|1969-12-31 23:59:...|            0|\n",
      "|null|[7B 22 6B 65 65 6...|assessment|        0|     3|1969-12-31 23:59:...|            0|\n",
      "|null|[7B 22 6B 65 65 6...|assessment|        0|     4|1969-12-31 23:59:...|            0|\n",
      "|null|[7B 22 6B 65 65 6...|assessment|        0|     5|1969-12-31 23:59:...|            0|\n",
      "|null|[7B 22 6B 65 65 6...|assessment|        0|     6|1969-12-31 23:59:...|            0|\n",
      "|null|[7B 22 6B 65 65 6...|assessment|        0|     7|1969-12-31 23:59:...|            0|\n",
      "|null|[7B 22 6B 65 65 6...|assessment|        0|     8|1969-12-31 23:59:...|            0|\n",
      "|null|[7B 22 6B 65 65 6...|assessment|        0|     9|1969-12-31 23:59:...|            0|\n",
      "|null|[7B 22 6B 65 65 6...|assessment|        0|    10|1969-12-31 23:59:...|            0|\n",
      "|null|[7B 22 6B 65 65 6...|assessment|        0|    11|1969-12-31 23:59:...|            0|\n",
      "|null|[7B 22 6B 65 65 6...|assessment|        0|    12|1969-12-31 23:59:...|            0|\n",
      "|null|[7B 22 6B 65 65 6...|assessment|        0|    13|1969-12-31 23:59:...|            0|\n",
      "|null|[7B 22 6B 65 65 6...|assessment|        0|    14|1969-12-31 23:59:...|            0|\n",
      "|null|[7B 22 6B 65 65 6...|assessment|        0|    15|1969-12-31 23:59:...|            0|\n",
      "|null|[7B 22 6B 65 65 6...|assessment|        0|    16|1969-12-31 23:59:...|            0|\n",
      "|null|[7B 22 6B 65 65 6...|assessment|        0|    17|1969-12-31 23:59:...|            0|\n",
      "|null|[7B 22 6B 65 65 6...|assessment|        0|    18|1969-12-31 23:59:...|            0|\n",
      "|null|[7B 22 6B 65 65 6...|assessment|        0|    19|1969-12-31 23:59:...|            0|\n",
      "+----+--------------------+----------+---------+------+--------------------+-------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# A quick peak at the dataframe, notice the topic that was created called assessment with the partition in Kafka\n",
    "raw_assessments.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- key: binary (nullable = true)\n",
      " |-- value: binary (nullable = true)\n",
      " |-- topic: string (nullable = true)\n",
      " |-- partition: integer (nullable = true)\n",
      " |-- offset: long (nullable = true)\n",
      " |-- timestamp: timestamp (nullable = true)\n",
      " |-- timestampType: integer (nullable = true)\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "pyspark.sql.dataframe.DataFrame"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Check out the schema of this PySpark dataframe\n",
    "raw_assessments.printSchema()\n",
    "type(raw_assessments)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Part 2: Convert the json data as a string into dataframe"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": true,
    "jupyter": {
     "outputs_hidden": true
    },
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# Selecting the value column from the earlier dataframe and cast it hexadecimal codes into string for analysis\n",
    "assessments = raw_assessments.select(raw_assessments.value.cast('string'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- value: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Recheck the schema to see if value is casted correctly\n",
    "assessments.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+\n",
      "|               value|\n",
      "+--------------------+\n",
      "|{\"keen_timestamp\"...|\n",
      "|{\"keen_timestamp\"...|\n",
      "|{\"keen_timestamp\"...|\n",
      "|{\"keen_timestamp\"...|\n",
      "|{\"keen_timestamp\"...|\n",
      "|{\"keen_timestamp\"...|\n",
      "|{\"keen_timestamp\"...|\n",
      "|{\"keen_timestamp\"...|\n",
      "|{\"keen_timestamp\"...|\n",
      "|{\"keen_timestamp\"...|\n",
      "|{\"keen_timestamp\"...|\n",
      "|{\"keen_timestamp\"...|\n",
      "|{\"keen_timestamp\"...|\n",
      "|{\"keen_timestamp\"...|\n",
      "|{\"keen_timestamp\"...|\n",
      "|{\"keen_timestamp\"...|\n",
      "|{\"keen_timestamp\"...|\n",
      "|{\"keen_timestamp\"...|\n",
      "|{\"keen_timestamp\"...|\n",
      "|{\"keen_timestamp\"...|\n",
      "+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Take a quick peak at the newly casted dataframe\n",
    "assessments.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Row(value='{\"keen_timestamp\":\"1516717442.735266\",\"max_attempts\":\"1.0\",\"started_at\":\"2018-01-23T14:23:19.082Z\",\"base_exam_id\":\"37f0a30a-7464-11e6-aa92-a8667f27e5dc\",\"user_exam_id\":\"6d4089e4-bde5-4a22-b65f-18bce9ab79c8\",\"sequences\":{\"questions\":[{\"user_incomplete\":true,\"user_correct\":false,\"options\":[{\"checked\":true,\"at\":\"2018-01-23T14:23:24.670Z\",\"id\":\"49c574b4-5c82-4ffd-9bd1-c3358faf850d\",\"submitted\":1,\"correct\":true},{\"checked\":true,\"at\":\"2018-01-23T14:23:25.914Z\",\"id\":\"f2528210-35c3-4320-acf3-9056567ea19f\",\"submitted\":1,\"correct\":true},{\"checked\":false,\"correct\":true,\"id\":\"d1bf026f-554f-4543-bdd2-54dcf105b826\"}],\"user_submitted\":true,\"id\":\"7a2ed6d3-f492-49b3-b8aa-d080a8aad986\",\"user_result\":\"missed_some\"},{\"user_incomplete\":false,\"user_correct\":false,\"options\":[{\"checked\":true,\"at\":\"2018-01-23T14:23:30.116Z\",\"id\":\"a35d0e80-8c49-415d-b8cb-c21a02627e2b\",\"submitted\":1},{\"checked\":false,\"correct\":true,\"id\":\"bccd6e2e-2cef-4c72-8bfa-317db0ac48bb\"},{\"checked\":true,\"at\":\"2018-01-23T14:23:41.791Z\",\"id\":\"7e0b639a-2ef8-4604-b7eb-5018bd81a91b\",\"submitted\":1,\"correct\":true}],\"user_submitted\":true,\"id\":\"bbed4358-999d-4462-9596-bad5173a6ecb\",\"user_result\":\"incorrect\"},{\"user_incomplete\":false,\"user_correct\":true,\"options\":[{\"checked\":false,\"at\":\"2018-01-23T14:23:52.510Z\",\"id\":\"a9333679-de9d-41ff-bb3d-b239d6b95732\"},{\"checked\":false,\"id\":\"85795acc-b4b1-4510-bd6e-41648a3553c9\"},{\"checked\":true,\"at\":\"2018-01-23T14:23:54.223Z\",\"id\":\"c185ecdb-48fb-4edb-ae4e-0204ac7a0909\",\"submitted\":1,\"correct\":true},{\"checked\":true,\"at\":\"2018-01-23T14:23:53.862Z\",\"id\":\"77a66c83-d001-45cd-9a5a-6bba8eb7389e\",\"submitted\":1,\"correct\":true}],\"user_submitted\":true,\"id\":\"e6ad8644-96b1-4617-b37b-a263dded202c\",\"user_result\":\"correct\"},{\"user_incomplete\":false,\"user_correct\":true,\"options\":[{\"checked\":false,\"id\":\"59b9fc4b-f239-4850-b1f9-912d1fd3ca13\"},{\"checked\":false,\"id\":\"2c29e8e8-d4a8-406e-9cdf-de28ec5890fe\"},{\"checked\":false,\"id\":\"62feee6e-9b76-4123-bd9e-c0b35126b1f1\"},{\"checked\":true,\"at\":\"2018-01-23T14:24:00.807Z\",\"id\":\"7f13df9c-fcbe-4424-914f-2206f106765c\",\"submitted\":1,\"correct\":true}],\"user_submitted\":true,\"id\":\"95194331-ac43-454e-83de-ea8913067055\",\"user_result\":\"correct\"}],\"attempt\":1,\"id\":\"5b28a462-7a3b-42e0-b508-09f3906d1703\",\"counts\":{\"incomplete\":1,\"submitted\":4,\"incorrect\":1,\"all_correct\":false,\"correct\":2,\"total\":4,\"unanswered\":0}},\"keen_created_at\":\"1516717442.735266\",\"certification\":\"false\",\"keen_id\":\"5a6745820eb8ab00016be1f1\",\"exam_name\":\"Normal Forms and All That Jazz Master Class\"}')"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Check out the first row of the dataframe\n",
    "assessments.first()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Part 3: Extract / unrolls the json data into new dataframes to answer you business questions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": true,
    "jupyter": {
     "outputs_hidden": true
    },
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# Using RDD method map() and lambda to massively process\n",
    "# the string content of value column into dictionary key, value to into PySpark dataframe\n",
    "extracted_assessments = assessments.rdd.map(lambda x: Row(**json.loads(x.value))).toDF()\n",
    "\n",
    "\n",
    "# Using a flatMap() and custom lambda to unnest sequence col\n",
    "def my_lambda_questions(x):\n",
    "    raw_dict = json.loads(x.value)\n",
    "    my_list = []\n",
    "    my_count = 0\n",
    "    for l in raw_dict[\"sequences\"][\"questions\"]:\n",
    "        my_count += 1\n",
    "        my_dict = {\"exam_name\" : raw_dict[\"exam_name\"], \"my_count\" : my_count}\n",
    "        my_list.append(Row(**my_dict))\n",
    "    return my_list\n",
    "\n",
    "# Get the unrolled questions per assessment \n",
    "questions = assessments.rdd.flatMap(my_lambda_questions).toDF()\n",
    "\n",
    "\n",
    "# Using flatMap() and custom lamda to unnest the score of sequence\n",
    "def lambda_correct_total(x):\n",
    "    raw_dict = json.loads(x.value)\n",
    "    my_list = []\n",
    "    if \"sequences\" in raw_dict:\n",
    "        if \"counts\" in raw_dict[\"sequences\"]:\n",
    "            if \"correct\" in raw_dict[\"sequences\"][\"counts\"] and \"total\" in raw_dict[\"sequences\"][\"counts\"]:\n",
    "                my_dict = {\"exam_name\": raw_dict[\"exam_name\"],\n",
    "                           \"correct\": raw_dict[\"sequences\"][\"counts\"][\"correct\"], \n",
    "                           \"total\": raw_dict[\"sequences\"][\"counts\"][\"total\"],\n",
    "                           \"incorrect\": raw_dict[\"sequences\"][\"counts\"][\"incorrect\"],\n",
    "                           \"unanswered\": raw_dict[\"sequences\"][\"counts\"][\"unanswered\"],\n",
    "                           \"incomplete\": raw_dict[\"sequences\"][\"counts\"][\"incomplete\"]}\n",
    "                my_list.append(Row(**my_dict))\n",
    "    return my_list\n",
    "\n",
    "#Get the unrolled score dataframe\n",
    "correct_total = assessments.rdd.flatMap(lambda_correct_total).toDF()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Part 4: Register dataframes as temporary tables to allow in memory queries against them"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": true,
    "jupyter": {
     "outputs_hidden": true
    },
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# Register the PySpark DataFrame as a Spark \"TempTable\" aka \"View\" in BigQuery\n",
    "extracted_assessments.registerTempTable('extracted_assessments')\n",
    "questions.registerTempTable('questions')\n",
    "correct_total.registerTempTable('ct')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Part 5: Perform SQL queries against the datframes you registered "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Question 1: How many assessments are in the dataset?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------+\n",
      "|total_assessments|\n",
      "+-----------------+\n",
      "|             3280|\n",
      "+-----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql(\"select count(*) as total_assessments from extracted_assessments\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Question 2: What is the least common courses taken?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------------------------------------+------------+\n",
      "|exam_name                                        |course_count|\n",
      "+-------------------------------------------------+------------+\n",
      "|Nulls, Three-valued Logic and Missing Information|1           |\n",
      "|Native Web Apps for Android                      |1           |\n",
      "|Learning to Visualize Data with D3.js            |1           |\n",
      "|Operating Red Hat Enterprise Linux Servers       |1           |\n",
      "|The Closed World Assumption                      |2           |\n",
      "|Arduino Prototyping Techniques                   |2           |\n",
      "|What's New in JavaScript                         |2           |\n",
      "|Learning Spring Programming                      |2           |\n",
      "|Hibernate and JPA Fundamentals                   |2           |\n",
      "|Understanding the Grails 3 Domain Model          |2           |\n",
      "+-------------------------------------------------+------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql(\"select exam_name, count(*) as course_count from extracted_assessments group by exam_name order by course_count limit 10\").show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Question 3:  What are and the most common courses taken?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------------------------------------------------+------------+\n",
      "|exam_name                                                  |course_count|\n",
      "+-----------------------------------------------------------+------------+\n",
      "|Learning Git                                               |394         |\n",
      "|Introduction to Python                                     |162         |\n",
      "|Intermediate Python Programming                            |158         |\n",
      "|Introduction to Java 8                                     |158         |\n",
      "|Learning to Program with R                                 |128         |\n",
      "|Introduction to Machine Learning                           |119         |\n",
      "|Software Architecture Fundamentals Understanding the Basics|109         |\n",
      "|Beginning C# Programming                                   |95          |\n",
      "|Learning Eclipse                                           |85          |\n",
      "|Learning Apache Maven                                      |80          |\n",
      "+-----------------------------------------------------------+------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql(\"select exam_name, count(*) as course_count from extracted_assessments group by exam_name order by course_count desc limit 10\").show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Question 4: What are some courses with the most number of questions?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------------------------------+--------+\n",
      "|exam_name                                 |num_ques|\n",
      "+------------------------------------------+--------+\n",
      "|Operating Red Hat Enterprise Linux Servers|20      |\n",
      "|Great Bash                                |10      |\n",
      "|Learning Linux System Administration      |8       |\n",
      "|Understanding the Grails 3 Domain Model   |7       |\n",
      "|Learning to Program with R                |7       |\n",
      "|What's New in JavaScript                  |7       |\n",
      "|Introduction to Data Science with R       |7       |\n",
      "|Being a Better Introvert                  |7       |\n",
      "|Arduino Inputs                            |6       |\n",
      "|Using Web Components                      |6       |\n",
      "+------------------------------------------+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql(\"select exam_name, max(my_count) as num_ques from questions group by exam_name order by num_ques desc limit 10\").show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Question 5: What are the summary statistics of average scores for all courses?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+------------------+\n",
      "|summary|         avg_score|\n",
      "+-------+------------------+\n",
      "|  count|               102|\n",
      "|   mean| 61.43039215686273|\n",
      "| stddev|14.974404559911491|\n",
      "|    min|              20.0|\n",
      "|    max|             100.0|\n",
      "+-------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "avg_score = spark.sql(\"select exam_name, round((avg(correct / total)*100), 2) as avg_score from ct group by exam_name order by avg_score desc\")\n",
    "avg_score.select(\"avg_score\").describe().show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Question 6: What are the courses with the highest number of perfect score?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------------------------+-----+\n",
      "|exam_name                            |count|\n",
      "+-------------------------------------+-----+\n",
      "|Learning Git                         |130  |\n",
      "|Introduction to Java 8               |94   |\n",
      "|Introduction to Machine Learning     |46   |\n",
      "|Beginning Programming with JavaScript|29   |\n",
      "|Advanced Machine Learning            |24   |\n",
      "|Intermediate Python Programming      |23   |\n",
      "|Git Fundamentals for Web Developers  |21   |\n",
      "|Learning Apache Maven                |21   |\n",
      "|Learning Eclipse                     |21   |\n",
      "|Learning to Program with R           |20   |\n",
      "+-------------------------------------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "score = spark.sql(\"select exam_name, (correct / total)*100 as score from ct order by score desc\")\n",
    "score.filter(score.score == \"100.0\").groupBy(score.exam_name).count().sort(desc('count')).limit(10).show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Question 7: What are the courses with highest number of scores lower than 50?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------------------------------------------------+-----+\n",
      "|exam_name                                                  |count|\n",
      "+-----------------------------------------------------------+-----+\n",
      "|Learning Git                                               |107  |\n",
      "|Introduction to Python                                     |67   |\n",
      "|Learning to Program with R                                 |58   |\n",
      "|Intermediate Python Programming                            |56   |\n",
      "|Software Architecture Fundamentals Understanding the Basics|42   |\n",
      "|Beginning C# Programming                                   |28   |\n",
      "|Learning C# Best Practices                                 |23   |\n",
      "|Introduction to Data Science with R                        |23   |\n",
      "|Introduction to Machine Learning                           |22   |\n",
      "|Mastering Advanced Git                                     |20   |\n",
      "+-----------------------------------------------------------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "score.filter(score.score < \"50.0\").groupBy(score.exam_name).count().sort(desc('count')).limit(10).show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Question 8: What are the courses with highest number of incomplete and unanswered questions?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------------------------------------------------+---------------+---------------+\n",
      "|exam_name                                                  |sum(incomplete)|sum(unanswered)|\n",
      "+-----------------------------------------------------------+---------------+---------------+\n",
      "|Learning to Program with R                                 |153            |23             |\n",
      "|Introduction to Python                                     |126            |42             |\n",
      "|Introduction to Data Science with R                        |72             |2              |\n",
      "|Beginning C# Programming                                   |64             |8              |\n",
      "|Intermediate Python Programming                            |50             |27             |\n",
      "|Introduction to Machine Learning                           |49             |1              |\n",
      "|Learning Eclipse                                           |49             |7              |\n",
      "|Software Architecture Fundamentals Understanding the Basics|41             |32             |\n",
      "|HTML5 The Basics                                           |41             |7              |\n",
      "|Learning Apache Maven                                      |32             |13             |\n",
      "+-----------------------------------------------------------+---------------+---------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "incomplete = correct_total.select(\"exam_name\", \"incomplete\", \"unanswered\").groupBy(\"exam_name\").sum().sort(desc(\"sum(incomplete)\")).limit(10)\n",
    "incomplete.show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------------------------------------------------+---------------+---------------+\n",
      "|exam_name                                                     |sum(incomplete)|sum(unanswered)|\n",
      "+--------------------------------------------------------------+---------------+---------------+\n",
      "|Learning Git                                                  |0              |119            |\n",
      "|Introduction to Python                                        |126            |42             |\n",
      "|Software Architecture Fundamentals Understanding the Basics   |41             |32             |\n",
      "|Intermediate Python Programming                               |50             |27             |\n",
      "|Mastering Advanced Git                                        |14             |24             |\n",
      "|Learning to Program with R                                    |153            |23             |\n",
      "|Learning C# Best Practices                                    |30             |21             |\n",
      "|Great Bash                                                    |9              |18             |\n",
      "|Learning Linux System Administration                          |29             |15             |\n",
      "|JavaScript: The Good Parts Master Class with Douglas Crockford|0              |14             |\n",
      "+--------------------------------------------------------------+---------------+---------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "unanswered = correct_total.select(\"exam_name\", \"incomplete\", \"unanswered\").groupBy(\"exam_name\").sum().sort(desc(\"sum(unanswered)\")).limit(10)\n",
    "unanswered.show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Part 6: Write PySpark dataframe to Hadoop HDFS in Parquet format to create a batch and serving layers scale up SQL"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "collapsed": true,
    "jupyter": {
     "outputs_hidden": true
    }
   },
   "outputs": [],
   "source": [
    "raw_assessments.write.mode('overwrite').parquet(\"/tmp/raw_assessments\")\n",
    "assessments.write.mode(\"overwrite\").parquet(\"/tmp/assessments\")\n",
    "extracted_assessments.write.mode(\"overwrite\").parquet(\"/tmp/extracted_assessments\")\n",
    "questions.write.mode(\"overwrite\").parquet(\"/tmp/questions\")\n",
    "correct_total.write.mode(\"overwrite\").parquet(\"/tmp/correct_total\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Appendix to view the json data in pprint format"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "collapsed": true,
    "jupyter": {
     "outputs_hidden": true
    },
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "p = pprint.PrettyPrinter(indent=1)\n",
    "f = open(\"assessment-attempts-20180128-121051-nested.json\",\"r\")\n",
    "s = f.read()\n",
    "json_data = json.loads(s)\n",
    "f.close()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{'base_exam_id': '94b741b2-fc67-4db4-adc2-aafae130848f',\n",
      " 'certification': 'false',\n",
      " 'exam_name': 'Intermediate C# Programming',\n",
      " 'keen_created_at': '1512281212.867346',\n",
      " 'keen_id': '5a23947cf84983000111f60a',\n",
      " 'keen_timestamp': '1512281212.867346',\n",
      " 'max_attempts': '1.0',\n",
      " 'sequences': {'attempt': 1,\n",
      "               'counts': {'all_correct': False,\n",
      "                          'correct': 1,\n",
      "                          'incomplete': 0,\n",
      "                          'incorrect': 3,\n",
      "                          'submitted': 4,\n",
      "                          'total': 4,\n",
      "                          'unanswered': 0},\n",
      "               'id': '4e0847c6-88fe-4c45-8ff9-820ad406f1f9',\n",
      "               'questions': [{'id': '12ea2f83-fec4-4032-a1af-92ed8e1cd2d1',\n",
      "                              'options': [{'at': '2017-12-03T06:05:34.832Z',\n",
      "                                           'checked': True,\n",
      "                                           'id': '3ebac309-aa96-4c0a-a478-82fc9c748b8c',\n",
      "                                           'submitted': 1},\n",
      "                                          {'checked': False,\n",
      "                                           'id': 'd97c8e01-7f6f-4c7a-ab0b-61fd4d63833d'},\n",
      "                                          {'checked': False,\n",
      "                                           'id': '888dda7f-f3d7-4be8-882b-edffe2e09419'},\n",
      "                                          {'checked': False,\n",
      "                                           'correct': True,\n",
      "                                           'id': '1a38c00a-417b-443b-867f-91e62ecd32d2'}],\n",
      "                              'user_correct': False,\n",
      "                              'user_incomplete': False,\n",
      "                              'user_result': 'incorrect',\n",
      "                              'user_submitted': True},\n",
      "                             {'id': '2df572cb-9325-4e14-bbdb-97c46da850eb',\n",
      "                              'options': [{'checked': False,\n",
      "                                           'id': 'e2ca7d6f-d763-4d9b-a321-a670eec2d083'},\n",
      "                                          {'checked': False,\n",
      "                                           'id': '8ab2a3db-66f5-4574-90a0-96c946b96843'},\n",
      "                                          {'at': '2017-12-03T06:05:59.017Z',\n",
      "                                           'checked': True,\n",
      "                                           'id': 'e2dd2051-937c-4f8c-b11e-2262dd86444c',\n",
      "                                           'submitted': 1},\n",
      "                                          {'checked': False,\n",
      "                                           'correct': True,\n",
      "                                           'id': 'c9750dce-087b-4c02-9115-d3b8b8819824'}],\n",
      "                              'user_correct': False,\n",
      "                              'user_incomplete': False,\n",
      "                              'user_result': 'incorrect',\n",
      "                              'user_submitted': True},\n",
      "                             {'id': '9114d53a-747a-4f9e-92f3-77704540809d',\n",
      "                              'options': [{'checked': False,\n",
      "                                           'id': '0e9c2000-7818-404b-ac4c-6875c40b5b6a'},\n",
      "                                          {'at': '2017-12-03T06:06:51.926Z',\n",
      "                                           'checked': True,\n",
      "                                           'correct': True,\n",
      "                                           'id': 'aa59975b-8eda-4f20-9b2a-6987b7af8ace',\n",
      "                                           'submitted': 1},\n",
      "                                          {'checked': False,\n",
      "                                           'id': 'd86af131-7a4a-4897-a3d1-8f6473ba0fba'},\n",
      "                                          {'checked': False,\n",
      "                                           'id': '291f35b1-30fa-4cf6-97fe-ca1f65b1d624'}],\n",
      "                              'user_correct': True,\n",
      "                              'user_incomplete': False,\n",
      "                              'user_result': 'correct',\n",
      "                              'user_submitted': True},\n",
      "                             {'id': '84baa11f-127d-4cfc-ac0a-87c8f173477f',\n",
      "                              'options': [{'at': '2017-12-03T06:06:39.787Z',\n",
      "                                           'checked': True,\n",
      "                                           'correct': True,\n",
      "                                           'id': 'efbcf9a2-0270-4612-8865-df8be4ca9e21',\n",
      "                                           'submitted': 1},\n",
      "                                          {'at': '2017-12-03T06:06:47.694Z',\n",
      "                                           'checked': True,\n",
      "                                           'id': 'cff3ab0b-adc4-4608-a7d4-087d48e8d9cc',\n",
      "                                           'submitted': 1},\n",
      "                                          {'checked': False,\n",
      "                                           'correct': True,\n",
      "                                           'id': '2db6b333-8001-4659-8083-c379721a34da'},\n",
      "                                          {'checked': False,\n",
      "                                           'id': 'c2cb6e77-06b5-479d-9b6e-df26e2dd05b8'}],\n",
      "                              'user_correct': False,\n",
      "                              'user_incomplete': False,\n",
      "                              'user_result': 'incorrect',\n",
      "                              'user_submitted': True}]},\n",
      " 'started_at': '2017-12-03T06:05:22.730Z',\n",
      " 'user_exam_id': 'f8986885-833b-46f9-be2c-aa0c8b9817bc'}\n"
     ]
    }
   ],
   "source": [
    "# this will pretty print the json in alphabetic order which may or may not match the file order\n",
    "p.pprint(json_data[79])"
   ]
  }
 ],
 "metadata": {
  "environment": {
   "name": "tf2-2-3-gpu.2-3.m55",
   "type": "gcloud",
   "uri": "gcr.io/deeplearning-platform-release/tf2-2-3-gpu.2-3:m55"
  },
  "kernelspec": {
   "display_name": "Python 3",
   "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.6.1"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}