Project_1B_ Project_Template.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Part I. ETL Pipeline for Pre-Processing the Files"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## PLEASE RUN THE FOLLOWING CODE FOR PRE-PROCESSING THE FILES"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Import Python packages "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Import Python packages \n",
    "import pandas as pd\n",
    "import cassandra\n",
    "import re\n",
    "import os\n",
    "import glob\n",
    "import numpy as np\n",
    "import json\n",
    "import csv\n",
    "from prettytable import PrettyTable"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Creating list of filepaths to process original event csv data files"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "/workspace/home\n"
     ]
    }
   ],
   "source": [
    "# checking your current working directory\n",
    "print(os.getcwd())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['.workspace-config.json',\n",
       " '.ipynb_checkpoints',\n",
       " 'event_data',\n",
       " 'images',\n",
       " 'event_datafile_new.csv',\n",
       " 'Project_1B_ Project_Template.ipynb']"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# print all the dir and file in current directory\n",
    "os.listdir()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "/workspace/home/event_data\n"
     ]
    }
   ],
   "source": [
    "# Get directory path to event data\n",
    "filepath = os.getcwd() + '/event_data'\n",
    "print(filepath)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['2018-11-10-events.csv',\n",
       " '2018-11-20-events.csv',\n",
       " '2018-11-09-events.csv',\n",
       " '2018-11-11-events.csv',\n",
       " '2018-11-07-events.csv',\n",
       " '2018-11-27-events.csv',\n",
       " '2018-11-21-events.csv',\n",
       " '2018-11-15-events.csv',\n",
       " '2018-11-17-events.csv',\n",
       " '2018-11-06-events.csv',\n",
       " '2018-11-24-events.csv',\n",
       " '2018-11-23-events.csv',\n",
       " '2018-11-04-events.csv',\n",
       " '2018-11-30-events.csv',\n",
       " '2018-11-29-events.csv',\n",
       " '2018-11-01-events.csv',\n",
       " '2018-11-16-events.csv',\n",
       " '2018-11-02-events.csv',\n",
       " '2018-11-13-events.csv',\n",
       " '2018-11-08-events.csv',\n",
       " '2018-11-12-events.csv',\n",
       " '2018-11-28-events.csv',\n",
       " '2018-11-19-events.csv',\n",
       " '2018-11-18-events.csv',\n",
       " '2018-11-26-events.csv',\n",
       " '2018-11-25-events.csv',\n",
       " '2018-11-05-events.csv',\n",
       " '2018-11-03-events.csv',\n",
       " '2018-11-14-events.csv',\n",
       " '2018-11-22-events.csv']"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# list all the files in event_data folder\n",
    "os.listdir(filepath)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Root: /workspace/home/event_data\n",
      "\n",
      "Dirs: []\n",
      "\n",
      "Files: ['2018-11-10-events.csv', '2018-11-20-events.csv', '2018-11-09-events.csv', '2018-11-11-events.csv', '2018-11-07-events.csv', '2018-11-27-events.csv', '2018-11-21-events.csv', '2018-11-15-events.csv', '2018-11-17-events.csv', '2018-11-06-events.csv', '2018-11-24-events.csv', '2018-11-23-events.csv', '2018-11-04-events.csv', '2018-11-30-events.csv', '2018-11-29-events.csv', '2018-11-01-events.csv', '2018-11-16-events.csv', '2018-11-02-events.csv', '2018-11-13-events.csv', '2018-11-08-events.csv', '2018-11-12-events.csv', '2018-11-28-events.csv', '2018-11-19-events.csv', '2018-11-18-events.csv', '2018-11-26-events.csv', '2018-11-25-events.csv', '2018-11-05-events.csv', '2018-11-03-events.csv', '2018-11-14-events.csv', '2018-11-22-events.csv']\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# print all the dirs and files in filepath\n",
    "for root, dirs, files in os.walk(filepath):\n",
    "    print(\"Root: {}\\n\".format(root))\n",
    "    print(\"Dirs: {}\\n\".format(dirs))\n",
    "    print(\"Files: {}\\n\".format(files))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "# get each filepath for each file under event_data\n",
    "for root, dirs, files in os.walk(filepath):\n",
    "    file_path_list = glob.glob(os.path.join(root,'*'))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Processing the files to create the data file csv that will be used for Apache Casssandra tables"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "8056\n"
     ]
    }
   ],
   "source": [
    "# get all the rows from all the file\n",
    "full_data_row = []\n",
    "for f in file_path_list:\n",
    "    # it is recommended to open the file with the newline='' argument value on all platforms to \n",
    "    # disable universal newline translation.\n",
    "    with open(f, 'r', encoding='utf-8', newline='') as csvFile:\n",
    "        csvReader = csv.reader(csvFile)\n",
    "        next(csvReader)\n",
    "        for line in csvReader:\n",
    "            full_data_row.append(line)\n",
    "            \n",
    "print(len(full_data_row))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create new csv file that will be used to insert data into the Apache Cassandra tables\n",
    "csv.register_dialect('myDialect', quoting=csv.QUOTE_ALL, skipinitialspace=True)\n",
    "\n",
    "with open('event_datafile_new.csv', 'w', encoding = 'utf8', newline='') as f:\n",
    "    writer = csv.writer(f, dialect='myDialect')\n",
    "    writer.writerow(['artist','firstName','gender','itemInSession','lastName','length',\\\n",
    "                'level','location','sessionId','song','userId'])\n",
    "    for row in full_data_row:\n",
    "        # check if artist is empty\n",
    "        if (row[0] == ''):\n",
    "            continue\n",
    "        writer.writerow((row[0], row[2], row[3], row[4], row[5], row[6], row[7], row[8], row[12], row[13], row[16]))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "6821\n"
     ]
    }
   ],
   "source": [
    "# check the number of rows in your csv file\n",
    "with open('event_datafile_new.csv', 'r', encoding = 'utf8') as f:\n",
    "    print(sum(1 for line in f))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Part II. Complete the Apache Cassandra coding portion of your project. \n",
    "\n",
    "## Now you are ready to work with the CSV file titled <font color=red>event_datafile_new.csv</font>, located within the Workspace directory.  The event_datafile_new.csv contains the following columns: \n",
    "- artist \n",
    "- firstName\n",
    "- gender\n",
    "- itemInSession\n",
    "- lastName\n",
    "- length\n",
    "- level\n",
    "- location\n",
    "- sessionId\n",
    "- song\n",
    "- userId\n",
    "\n",
    "The image below is a screenshot of what the denormalized data should appear like in the <font color=red>**event_datafile_new.csv**</font> after the code above is run:<br>\n",
    "\n",
    "<img src=\"images/image_event_datafile_new.jpg\">"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Begin writing your Apache Cassandra code in the cells below"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Creating a Cluster"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "# connect to a cassandra instance\n",
    "from cassandra.cluster import Cluster\n",
    "try: \n",
    "    cluster = Cluster(['127.0.0.1'])\n",
    "    # To establish connection and begin executing queries, need a session\n",
    "    session = cluster.connect()\n",
    "except Exception as e:\n",
    "    print(e)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Create Keyspace"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "try:\n",
    "    session.execute(\"\"\"\n",
    "    CREATE KEYSPACE IF NOT EXISTS sparkify \n",
    "    WITH REPLICATION = \n",
    "    { 'class' : 'SimpleStrategy', 'replication_factor' : 1 }\"\"\"\n",
    ")\n",
    "\n",
    "except Exception as e:\n",
    "    print(e)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Set Keyspace"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "try:\n",
    "    session.set_keyspace('sparkify')\n",
    "except Exception as e:\n",
    "    print(e)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Now we need to create tables to run the following queries. Remember, with Apache Cassandra you model the database tables on the queries you want to run."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create queries to ask the following three questions of the data\n",
    "\n",
    "#### 1. Give me the artist, song title and song's length in the music app history that was heard during  sessionId = 338, and itemInSession  = 4\n",
    "\n",
    "\n",
    "#### 2. Give me only the following: name of artist, song (sorted by itemInSession) and user (first and last name) for userid = 10, sessionid = 182\n",
    "    \n",
    "\n",
    "#### 3. Give me every user name (first and last) in my music app history who listened to the song 'All Hands Against His Own'\n",
    "\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Query 1. Give me the artist, song title and song's length in the music app history that was heard during a given  sessionId and itemInSession\n",
    "\n",
    "**Table Name:** `song_details_session_item`\n",
    "\n",
    "**Columns:** \n",
    "- `sessionId`\n",
    "- `itemInSession`\n",
    "- `artist`\n",
    "- `song`\n",
    "- `songLength`\n",
    "\n",
    "**Primary Key:** `sessionId, itemInSession`\n",
    "\n",
    "The reason behind choosing `sessionId` and `itemInSession` as primary key is that we have to filter the data based on the given `sessionId` and `itemInSession` value. Here `sessionId` is the partition column and `itemInSession` is the clustering column and both this column can make primary key unique."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "query = \"create table if not exists song_details_session_item \"\n",
    "query += \"(sessionId int, itemInSession int, artist text, song text, songLength float, \\\n",
    "primary key(sessionId, itemInSession))\"\n",
    "\n",
    "try:\n",
    "    session.execute(query)\n",
    "except Exception as e:\n",
    "    print(e)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "scrolled": false
   },
   "outputs": [],
   "source": [
    "file = 'event_datafile_new.csv'\n",
    "\n",
    "with open(file, encoding = 'utf8') as f:\n",
    "    csvreader = csv.reader(f)\n",
    "    next(csvreader) # skip header\n",
    "    for line in csvreader:\n",
    "        query = \"insert into song_details_session_item(sessionId, itemInSession, artist, song, songLength)\"\n",
    "        query = query + \"values (%s, %s, %s, %s, %s)\"\n",
    "        try:\n",
    "            session.execute(query, (int(line[8]), int(line[3]), line[0], line[9], float(line[5])))\n",
    "        except Exception as e:\n",
    "            print(e)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Do a SELECT with condition  sessionId = 338 and itemInSession  = 4 \n",
    "\n",
    "The following query will return artist, song and songLength where the sessionId=338 and itemInSession=4"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+---------------------------------+--------------------+\n",
      "|   Artist  |               Song              |     songLength     |\n",
      "+-----------+---------------------------------+--------------------+\n",
      "| Faithless | Music Matters (Mark Knight Dub) | 495.30731201171875 |\n",
      "+-----------+---------------------------------+--------------------+\n"
     ]
    }
   ],
   "source": [
    "query = \"select artist, song, songLength from song_details_session_item where sessionId=338 \\\n",
    "and itemInSession=4\"\n",
    "\n",
    "try:\n",
    "    rows = session.execute(query)\n",
    "except Exception as e:\n",
    "    print(e)\n",
    "    \n",
    "table = PrettyTable()\n",
    "table.field_names = [\"Artist\", \"Song\", \"songLength\"]\n",
    "for row in rows:\n",
    "    table.add_row([row.artist, row.song, row.songlength])\n",
    "    \n",
    "print(table)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Query 2. Give me only the following: name of artist, song (sorted by itemInSession) and user (first and last name) for given userid and sessionid\n",
    "\n",
    "**Table Name:** `artist_song_user_on_userid_session`\n",
    "\n",
    "**Columns:**\n",
    "- `userId`\n",
    "- `sessionId`\n",
    "- `itemInSession`\n",
    "- `artist`\n",
    "- `song`\n",
    "- `firstName`\n",
    "- `lastName`\n",
    "\n",
    "**Primary Key:** `userId, sessionId, itemInSession`\n",
    "\n",
    "The combination of `userId` and `sessionId` will not make the primary key unique. So, we need another column to add to make the primary key unique. And also we have to sort the rows based on `itemInSession` and we know that the clustering column sort data within a partition in ascending order. So, we can add `itemInSession` as clustering column to make the primary key unique and also to sort the data based on `itemInSession` value. Here `userId` is the ppartition column and `sessionId` and `itemInSession` are clustering columns."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "query = \"create table if not exists artist_song_user_on_userid_session \"\n",
    "query += \"(userId int, sessionId int, itemInSession int, artist text, song text, firstName text, \\\n",
    "lastName text, primary key(userId, sessionId, itemInSession))\"\n",
    "\n",
    "try:\n",
    "    session.execute(query)\n",
    "except Exception as e:\n",
    "    print(e)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "file = 'event_datafile_new.csv'\n",
    "\n",
    "with open(file, encoding = 'utf8') as f:\n",
    "    csvreader = csv.reader(f)\n",
    "    next(csvreader) # skip header\n",
    "    for line in csvreader:\n",
    "        query = \"insert into artist_song_user_on_userid_session (userId, sessionId, itemInSession, \\\n",
    "        artist, song, firstName, lastName)\"\n",
    "        query = query + \"values (%s, %s, %s, %s, %s, %s, %s)\"\n",
    "        try:\n",
    "            session.execute(query, (int(line[10]), int(line[8]), int(line[3]), line[0], line[9], \n",
    "                                    line[1], line[4]))\n",
    "        except Exception as e:\n",
    "            print(e)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Do a SELECT with condition userId=10  sessionId = 182\n",
    "\n",
    "The following query return artist, song(sorted based on itemInSession in ascending order), firstName and lastName from table where userId=10 and sessionId=182"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+------------------------------------------------------+-------------+\n",
      "|       Artist      |                         Song                         |     User    |\n",
      "+-------------------+------------------------------------------------------+-------------+\n",
      "|  Down To The Bone |                  Keep On Keepin' On                  | Sylvie Cruz |\n",
      "|    Three Drives   |                     Greece 2000                      | Sylvie Cruz |\n",
      "| Sebastien Tellier |                      Kilometer                       | Sylvie Cruz |\n",
      "|   Lonnie Gordon   | Catch You Baby (Steve Pitron & Max Sanna Radio Edit) | Sylvie Cruz |\n",
      "+-------------------+------------------------------------------------------+-------------+\n"
     ]
    }
   ],
   "source": [
    "query = \"select artist, song, firstName, lastName from \\\n",
    "artist_song_user_on_userid_session where userId=10 and sessionId=182\"\n",
    "\n",
    "try:\n",
    "    rows = session.execute(query)\n",
    "except Exception as e:\n",
    "    print(e)\n",
    "    \n",
    "table = PrettyTable()\n",
    "table.field_names = [\"Artist\", \"Song\", \"User\"]\n",
    "for row in rows:\n",
    "    table.add_row([row.artist, row.song, row.firstname+\" \"+row.lastname])\n",
    "    \n",
    "print(table)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 3. Give me every user name (first and last) in my music app history who listened to the song 'All Hands Against His Own'\n",
    "\n",
    "**Table Name:** `user_on_song`\n",
    "\n",
    "**Columns:**\n",
    "- `song`\n",
    "- `userId`\n",
    "- `firstName`\n",
    "- `lastName`\n",
    "\n",
    "**Primary Key:** `song, userId`\n",
    "\n",
    "Here, we want the user name who listened to a particular song. So, here only `song` cant be a primary key alone, as it dont make primary key unique. As there may more than one user who listen to a particular song. So, we can add `userId` to `song` to make our primary key. Although `song` and `userId` may not uniquely identify a row, but in this case that wont be a problem. As, in apache cassandra when we try to insert data, if there is already data with the same primary key then data will be updated and no error will be thrown. And, in this case we are only interested to the user name. Even though data will be updated but we will have the same user name as for a particular `userId` `firstName` and `lastName` is same always. So, here `song` is the partition key and `userId` is the clustering column."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "query = \"create table if not exists user_on_song \"\n",
    "query += \"(song text, userId int, firstName text, lastName text, primary key(song, userId))\"\n",
    "\n",
    "try:\n",
    "    session.execute(query)\n",
    "except Exception as e:\n",
    "    print(e)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "file = 'event_datafile_new.csv'\n",
    "\n",
    "with open(file, encoding = 'utf8') as f:\n",
    "    csvreader = csv.reader(f)\n",
    "    next(csvreader) # skip header\n",
    "    for line in csvreader:\n",
    "        query = \"insert into user_on_song (song, userId, firstName, lastName)\"\n",
    "        query = query + \"values (%s, %s, %s, %s)\"\n",
    "        try:\n",
    "            session.execute(query, (line[9], int(line[10]), line[1], line[4]))\n",
    "        except Exception as e:\n",
    "            print(e)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Do a SELECT with condition  song='All Hands Against His Own'\n",
    "\n",
    "The following query will return user firstName and lastName who listen to the given song."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------+\n",
      "|       User       |\n",
      "+------------------+\n",
      "| Jacqueline Lynch |\n",
      "|   Tegan Levine   |\n",
      "|   Sara Johnson   |\n",
      "+------------------+\n"
     ]
    }
   ],
   "source": [
    "query = \"select firstName, lastName from user_on_song where song='All Hands Against His Own'\"\n",
    "\n",
    "try:\n",
    "    rows = session.execute(query)\n",
    "except Exception as e:\n",
    "    print(e)\n",
    "    \n",
    "table = PrettyTable()\n",
    "table.field_names = [\"User\"]\n",
    "for row in rows:\n",
    "    table.add_row([row.firstname+\" \"+row.lastname])\n",
    "    \n",
    "print(table)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Drop the tables before closing out the sessions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Dropped table: song_details_session_item\n",
      "Dropped table: artist_song_user_on_userid_session\n",
      "Dropped table: user_on_song\n"
     ]
    }
   ],
   "source": [
    "for table in ['song_details_session_item','artist_song_user_on_userid_session','user_on_song']:\n",
    "    query = \"DROP TABLE IF EXISTS \" + table\n",
    "    try:\n",
    "        session.execute(query)\n",
    "        print(\"Dropped table: \" + table)\n",
    "    except Exception as e:\n",
    "        print(e)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Close the session and cluster connection¶"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "session.shutdown()\n",
    "cluster.shutdown()"
   ]
  }
 ],
 "metadata": {
  "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.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}