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": [
    "#### 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"
   ]
  },
  {
   "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": [
      "/home/workspace\n"
     ]
    }
   ],
   "source": [
    "# checking your current working directory\n",
    "print(os.getcwd())\n",
    "\n",
    "# Get your current folder and subfolder event data\n",
    "filepath = os.getcwd() + '/event_data'\n",
    "\n",
    "# Create a for loop to create a list of files and collect each filepath\n",
    "for root, dirs, files in os.walk(filepath):\n",
    "    \n",
    "# join the file path and roots with the subdirectories using glob\n",
    "    file_path_list = glob.glob(os.path.join(root,'*'))\n",
    "    #print(file_path_list)"
   ]
  },
  {
   "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": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "# initiating an empty list of rows that will be generated from each file\n",
    "full_data_rows_list = [] \n",
    "    \n",
    "# for every filepath in the file path list \n",
    "for f in file_path_list:\n",
    "\n",
    "# reading csv file \n",
    "    with open(f, 'r', encoding = 'utf8', newline='') as csvfile: \n",
    "        # creating a csv reader object \n",
    "        csvreader = csv.reader(csvfile) \n",
    "        next(csvreader)\n",
    "        \n",
    " # extracting each data row one by one and append it        \n",
    "        for line in csvreader:\n",
    "            #print(line)\n",
    "            full_data_rows_list.append(line) \n",
    "            \n",
    "# uncomment the code below if you would like to get total number of rows \n",
    "#print(len(full_data_rows_list))\n",
    "# uncomment the code below if you would like to check to see what the list of event data rows will look like\n",
    "#print(full_data_rows_list)\n",
    "\n",
    "# creating a smaller event data csv file called event_datafile_full csv that will be used to insert data into the \\\n",
    "# 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_rows_list:\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": 4,
   "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.\n",
    "\n",
    "The event_datafile_new.csv contains the following columns: \n",
    "- artist \n",
    "- firstName of user\n",
    "- gender of user\n",
    "- item number in session\n",
    "- last name of user\n",
    "- length of the song\n",
    "- level (paid or free song)\n",
    "- location of the user\n",
    "- sessionId\n",
    "- song title\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": [
    "#### Creating a Cluster"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "# This should make a connection to a Cassandra instance your local machine \n",
    "# (127.0.0.1)\n",
    "\n",
    "from cassandra.cluster import Cluster\n",
    "cluster = Cluster()\n",
    "\n",
    "# To establish connection and begin executing queries, need a session\n",
    "session = cluster.connect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Create Keyspace"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "try:\n",
    "    session.execute(\"\"\"\n",
    "    CREATE KEYSPACE IF NOT EXISTS udacity \n",
    "    WITH REPLICATION = { 'class' : 'SimpleStrategy', 'replication_factor' : 1 }\n",
    "    \"\"\"\n",
    ")\n",
    "\n",
    "except Exception as e:\n",
    "    print(e)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Set Keyspace"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "try:\n",
    "    session.set_keyspace('udacity')\n",
    "except Exception as e:\n",
    "    print(e)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Helper Functions: read_line and run_query"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "def read_line(line):\n",
    "    \"\"\"\n",
    "    Decode the lines read from the csv file and store them in a dictionary to make it easier to manipulate.\n",
    "    \n",
    "    Args:\n",
    "        line (:obj:`list`): One object of class list representing one row of csv file.\n",
    "    \n",
    "    Returns:\n",
    "        record (:obj:`dict`): Dictionary with the row decoded in different named fields.\n",
    "    \"\"\"\n",
    "    \n",
    "    record = {\n",
    "        \"artist\" : line[0],\n",
    "        \"firstname\" : line[1],\n",
    "        \"gender\" : line[2],\n",
    "        \"iteminsession\" : int(line[3]),\n",
    "        \"lastname\" : line[4],\n",
    "        \"length\" : float(line[5]),\n",
    "        \"level\" : line[6],\n",
    "        \"location\" : line[7],\n",
    "        \"sessionid\" : int(line[8]),\n",
    "        \"song\" : line[9],\n",
    "        \"userid\" : int(line[10])\n",
    "    }\n",
    "    return record"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "def run_query(session, query):\n",
    "    \"\"\"\n",
    "    Execute the query passed as argument in Apache Cassandra using the session to the database.\n",
    "    \n",
    "    Args:\n",
    "        session (:obj:`session`): One object of class session to the Apache Cassandra database.\n",
    "        query (str): Query to be executed in Apache Cassandra\n",
    "    \n",
    "    Returns:\n",
    "        df (:obj:`DataFrame`): Pandas DataFrame representing the results of the query in Apache Cassandra.\n",
    "    \"\"\"\n",
    "    try:\n",
    "        rows = session.execute(query)\n",
    "        df = pd.DataFrame(list(rows))\n",
    "    except Exception as e:\n",
    "        print(e)\n",
    "    \n",
    "    return df"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Table 1: session_songs\n",
    "_Give me the artist, song title and song's length in the music app history that was heard during  sessionId = 338, and itemInSession  = 4_"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "''' \n",
    "    Table created to provide an fast query on songs played by users on a specific track of one session. \n",
    "    It is specifically designed to supply the requirements of a query like stated in item 1:\n",
    "        -> 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",
    "    The INSERT and SELECT are also designed to deliver the analytics team request on that query.\n",
    "'''\n",
    "\n",
    "query = \"CREATE TABLE IF NOT EXISTS session_songs \"\n",
    "query = query + \"(sessionid int, iteminsession int, artist text, song text, length float, PRIMARY KEY (sessionid, iteminsession))\"\n",
    "try:\n",
    "    session.execute(query)\n",
    "except Exception as e:\n",
    "    print(e)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "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 session_songs (sessionid, iteminsession, artist, song, length)\"\n",
    "        query = query + \" VALUES (%s, %s, %s, %s, %s)\"\n",
    "        row = read_line(line)\n",
    "        session.execute(query, (row[\"sessionid\"], row[\"iteminsession\"], row[\"artist\"], row[\"song\"], row[\"length\"]))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Description of the model\n",
    "In the session_songs table the Primary Key has two fields: session id and item session id. sessionid is used as the partition key and iteminsession as clustering keys. Partitioning is done by session and within that partition, rows are ordered by the iteminsession. Since we need to know a specific play occurance of one song, we created the table with this level of detail. Session Id along with the Item Session Id seems to be the most detailed information available in event datafile and that was what we decided to keep in session_songs in order to address the requirement of this query."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>sessionid</th>\n",
       "      <th>iteminsession</th>\n",
       "      <th>artist</th>\n",
       "      <th>song</th>\n",
       "      <th>length</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>338</td>\n",
       "      <td>4</td>\n",
       "      <td>Faithless</td>\n",
       "      <td>Music Matters (Mark Knight Dub)</td>\n",
       "      <td>495.307312</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   sessionid  iteminsession     artist                             song  \\\n",
       "0        338              4  Faithless  Music Matters (Mark Knight Dub)   \n",
       "\n",
       "       length  \n",
       "0  495.307312  "
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "query = \"select sessionid, iteminsession, artist, song, length from session_songs WHERE sessionid = 338 and iteminsession = 4\"\n",
    "run_query(session, query)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Table 2: users_plays\n",
    "_Give me only the following: name of artist, song (sorted by itemInSession) and user (first and last name) for userid = 10, sessionid = 182_"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "''' \n",
    "    Table created to provide an fast query on songs played by users on a specific session. \n",
    "    It is specifically designed to supply the requirements of a query like stated in item 2:\n",
    "        -> Give me only the following: name of artist, song (sorted by itemInSession) and user (first and last name) for userid = 10, sessionid = 182\n",
    "    The INSERT and SELECT are also designed to deliver the analytics team request on that query.\n",
    "'''\n",
    "\n",
    "query = \"CREATE TABLE IF NOT EXISTS users_plays \"\n",
    "query = query + \"(userid int, sessionid int, iteminsession int, artist text, song text, firstname text, lastname text, PRIMARY KEY (userid, sessionid, iteminsession))\"\n",
    "try:\n",
    "    session.execute(query)\n",
    "except Exception as e:\n",
    "    print(e)             "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "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 users_plays (userid, sessionid, iteminsession, artist, song, firstname, lastname)\"\n",
    "        query = query + \" VALUES (%s, %s, %s, %s, %s, %s, %s)\"\n",
    "        row = read_line(line)\n",
    "        session.execute(query, (row[\"userid\"], row[\"sessionid\"], row[\"iteminsession\"], row[\"artist\"], row[\"song\"], row[\"firstname\"], row[\"lastname\"]))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Description of the model\n",
    "In the users_plays table the Primary Key has three fields: user id, session id and item session id. User id is used as the partition key while session id and iteminsession as clustering keys. Partitioning is done by user id and within that partition, rows are ordered by the session id and iteminsession. It is important for this query to obtain the songs ordered by item session id and that made us to model the table with session id and item session id as clustering keys. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>artist</th>\n",
       "      <th>iteminsession</th>\n",
       "      <th>song</th>\n",
       "      <th>firstname</th>\n",
       "      <th>lastname</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>Down To The Bone</td>\n",
       "      <td>0</td>\n",
       "      <td>Keep On Keepin' On</td>\n",
       "      <td>Sylvie</td>\n",
       "      <td>Cruz</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>Three Drives</td>\n",
       "      <td>1</td>\n",
       "      <td>Greece 2000</td>\n",
       "      <td>Sylvie</td>\n",
       "      <td>Cruz</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>Sebastien Tellier</td>\n",
       "      <td>2</td>\n",
       "      <td>Kilometer</td>\n",
       "      <td>Sylvie</td>\n",
       "      <td>Cruz</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>Lonnie Gordon</td>\n",
       "      <td>3</td>\n",
       "      <td>Catch You Baby (Steve Pitron &amp; Max Sanna Radio...</td>\n",
       "      <td>Sylvie</td>\n",
       "      <td>Cruz</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "              artist  iteminsession  \\\n",
       "0   Down To The Bone              0   \n",
       "1       Three Drives              1   \n",
       "2  Sebastien Tellier              2   \n",
       "3      Lonnie Gordon              3   \n",
       "\n",
       "                                                song firstname lastname  \n",
       "0                                 Keep On Keepin' On    Sylvie     Cruz  \n",
       "1                                        Greece 2000    Sylvie     Cruz  \n",
       "2                                          Kilometer    Sylvie     Cruz  \n",
       "3  Catch You Baby (Steve Pitron & Max Sanna Radio...    Sylvie     Cruz  "
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "query = \"select artist, iteminsession, song, firstname, lastname from users_plays WHERE userid = 10 and sessionid = 182\"\n",
    "run_query(session, query)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Table 3: song_fans\n",
    "_Give me every user name (first and last) in my music app history who listened to the song 'All Hands Against His Own'_"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "''' \n",
    "    Table created to provide an fast query on users that played a specific song. \n",
    "    It is specifically designed to supply the requirements of a query like stated in item 3:\n",
    "        -> Give me every user name (first and last) in my music app history who listened to the song 'All Hands Against His Own'\n",
    "    The INSERT and SELECT are also designed to deliver the analytics team request on that query.\n",
    "'''\n",
    "\n",
    "query = \"CREATE TABLE IF NOT EXISTS song_fans \"\n",
    "query = query + \"(song text, userid int, firstname text, lastname text, PRIMARY KEY (song, userid))\"\n",
    "try:\n",
    "    session.execute(query)\n",
    "except Exception as e:\n",
    "    print(e)                    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "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 song_fans (song, userid, firstname, lastname)\"\n",
    "        query = query + \" VALUES (%s, %s, %s, %s)\"\n",
    "        row = read_line(line)\n",
    "        session.execute(query, (row[\"song\"], row[\"userid\"], row[\"firstname\"], row[\"lastname\"]))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Description of the model\n",
    "In the song_fans table the Primary Key has two fields: song name and user id. Song name is used as the partition key and user id as clustering keys. Partitioning is done by song name and within that partition, rows are ordered by the user id. Since we only want to provide the users that have listened to specific songs, we did not add more fields from events. In particular, we are not interested to know how many times the song was played in total nor how many times a specific user played that song. Song name and user id suffice our requirements here as they together identify uniquely what is important for the query."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>firstname</th>\n",
       "      <th>lastname</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>Jacqueline</td>\n",
       "      <td>Lynch</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>Tegan</td>\n",
       "      <td>Levine</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>Sara</td>\n",
       "      <td>Johnson</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "    firstname lastname\n",
       "0  Jacqueline    Lynch\n",
       "1       Tegan   Levine\n",
       "2        Sara  Johnson"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "query = \"select firstname, lastname from song_fans WHERE song = 'All Hands Against His Own'\"\n",
    "run_query(session, query)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Drop the tables before closing out the sessions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "query = \"drop table session_songs\"\n",
    "try:\n",
    "    rows = session.execute(query)\n",
    "except Exception as e:\n",
    "    print(e)\n",
    "\n",
    "query = \"drop table users_plays\"\n",
    "try:\n",
    "    rows = session.execute(query)\n",
    "except Exception as e:\n",
    "    print(e)\n",
    "    \n",
    "query = \"drop table song_fans\"\n",
    "try:\n",
    "    rows = session.execute(query)\n",
    "except Exception as e:\n",
    "    print(e)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Close the session and cluster connection¶"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "session.shutdown()\n",
    "cluster.shutdown()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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
}