Project_1B.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "# Project: Data Modeling with Cassandra\n",
    "\n",
    "## Context\n",
    "\n",
    "A startup called Sparkify wants to analyze the data they've been collecting on songs and user activity on their new music streaming app. The analysis team is particularly interested in understanding what songs users are listening to. Currently, there is no easy way to query the data to generate the results, since the data reside in a directory of CSV files on user activity on the app.\n",
    "\n",
    "They'd like a data engineer to create an Apache Cassandra database which can create queries on song play data to answer the questions, and wish to bring you on the project. Your role is to create a database for this analysis. You'll be able to test your database by running queries given to you by the analytics team from Sparkify to create the results."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "# Part I. ETL Pipeline for Pre-Processing the Files"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "In this first part, we'll focus on ingesting the CSV files that reside inside the `event_data` folder, and create a single \"compiled\" file that we'll use to load the data into the Cassandra database."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "#### Import Python packages "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "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": {
    "editable": true
   },
   "source": [
    "#### Creating list of filepaths to process original event csv data files"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "/home/workspace\n"
     ]
    }
   ],
   "source": [
    "print(os.getcwd())\n",
    "\n",
    "filepath = os.getcwd() + '/event_data'\n",
    "\n",
    "for root, dirs, files in os.walk(filepath):\n",
    "    # ignore notebook temp files\n",
    "    if ('.ipynb_checkpoints' in root):\n",
    "        continue\n",
    "    \n",
    "    file_path_list = glob.glob(os.path.join(root,'*.csv'))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "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": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Processing file /home/workspace/event_data/2018-11-27-events.csv\n",
      "Processing file /home/workspace/event_data/2018-11-04-events.csv\n",
      "Processing file /home/workspace/event_data/2018-11-07-events.csv\n",
      "Processing file /home/workspace/event_data/2018-11-09-events.csv\n",
      "Processing file /home/workspace/event_data/2018-11-19-events.csv\n",
      "Processing file /home/workspace/event_data/2018-11-05-events.csv\n",
      "Processing file /home/workspace/event_data/2018-11-22-events.csv\n",
      "Processing file /home/workspace/event_data/2018-11-16-events.csv\n",
      "Processing file /home/workspace/event_data/2018-11-26-events.csv\n",
      "Processing file /home/workspace/event_data/2018-11-24-events.csv\n",
      "Processing file /home/workspace/event_data/2018-11-29-events.csv\n",
      "Processing file /home/workspace/event_data/2018-11-15-events.csv\n",
      "Processing file /home/workspace/event_data/2018-11-20-events.csv\n",
      "Processing file /home/workspace/event_data/2018-11-06-events.csv\n",
      "Processing file /home/workspace/event_data/2018-11-18-events.csv\n",
      "Processing file /home/workspace/event_data/2018-11-21-events.csv\n",
      "Processing file /home/workspace/event_data/2018-11-10-events.csv\n",
      "Processing file /home/workspace/event_data/2018-11-23-events.csv\n",
      "Processing file /home/workspace/event_data/2018-11-02-events.csv\n",
      "Processing file /home/workspace/event_data/2018-11-28-events.csv\n",
      "Processing file /home/workspace/event_data/2018-11-03-events.csv\n",
      "Processing file /home/workspace/event_data/2018-11-13-events.csv\n",
      "Processing file /home/workspace/event_data/2018-11-30-events.csv\n",
      "Processing file /home/workspace/event_data/2018-11-12-events.csv\n",
      "Processing file /home/workspace/event_data/2018-11-01-events.csv\n",
      "Processing file /home/workspace/event_data/2018-11-14-events.csv\n",
      "Processing file /home/workspace/event_data/2018-11-25-events.csv\n",
      "Processing file /home/workspace/event_data/2018-11-08-events.csv\n",
      "Processing file /home/workspace/event_data/2018-11-17-events.csv\n",
      "Processing file /home/workspace/event_data/2018-11-11-events.csv\n"
     ]
    }
   ],
   "source": [
    "full_data_rows_list = [] \n",
    "\n",
    "for f in file_path_list:\n",
    "    print(f\"Processing file {f}\")\n",
    "    \n",
    "    with open(f, 'r', encoding = 'utf8', newline='') as csvfile: \n",
    "        csvreader = csv.reader(csvfile) \n",
    "        next(csvreader)\n",
    "        \n",
    "        for line in csvreader:\n",
    "            full_data_rows_list.append(line) \n",
    "\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]))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "There are 6821 rows in the resulting CSV\n"
     ]
    }
   ],
   "source": [
    "with open('event_datafile_new.csv', 'r', encoding = 'utf8') as f:\n",
    "    print(\"There are %d rows in the resulting CSV\" % sum(1 for line in f))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "# Part II. Data Modeling\n",
    "\n",
    "For the next part, we will work with the CSV file titled <font color=red>event_datafile_new.csv</font>, located within the Workspace directory, that was generated in Part I.  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": {
    "editable": true
   },
   "source": [
    "#### Creating a Cluster"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "from cassandra.cluster import Cluster\n",
    "\n",
    "try:\n",
    "    cluster = Cluster(['127.0.0.1'])\n",
    "    session = cluster.connect()\n",
    "except Exception as e:\n",
    "    print(e)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "#### Create Keyspace"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "try:\n",
    "    session.execute(\"\"\"\n",
    "        CREATE KEYSPACE IF NOT EXISTS sparkify\n",
    "        WITH REPLICATION = {'class': 'SimpleStrategy', 'replication_factor': 1}\n",
    "    \"\"\")\n",
    "except Exception as e:\n",
    "    print(e)\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "#### Set Keyspace"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "try:\n",
    "    session.set_keyspace('sparkify')\n",
    "except Exception as e:\n",
    "    print(e)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "## Questions made by the Sparkify team\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "### Question 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",
    "\n",
    "It is required by the team to return the name of the artist, the title and length of the song associated to the itemInSession \"4\", during the sessionId \"338\". To answer the question, we have to approach data from a session's perspective.\n",
    "\n",
    "First we should understand the data involved in the requirement:\n",
    "1. `artist` (varchar): Name of the artist\n",
    "2. `song` (varchar): Title of the song\n",
    "3. `length` (decimal): Length of the song in seconds, including fractions of a second\n",
    "4. `sessionId` (int): Id of the session during which the song was played\n",
    "5. `itemInSession` (int): Position of the song in the session's history of songs played\n",
    "\n",
    "_Please remember that we are refering to the column headers in the CSV files._\n",
    "\n",
    "Since we are talking about the songs **played** during a **session**, we can name the table `session_plays`.\n",
    "\n",
    "Then, we must define the attributes (columns) that will be using to identify each row and partition the data. In this case, the search will be done by `sessionId` and `itemInSession`, so these will conform the primary key for the table.\n",
    "\n",
    "\n",
    "If we translate this requirement to a SQL instruction, we have the following query:\n",
    "\n",
    "`SELECT artist, song, length FROM session_plays WHERE sessionid = 338 and iteminsession = 4`\n",
    "\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "query = \"CREATE TABLE IF NOT EXISTS session_plays \\\n",
    "            (sessionid int, iteminsession int, artist varchar, song varchar, length decimal, \\\n",
    "            PRIMARY KEY ((sessionid, iteminsession)))\"\n",
    "try:\n",
    "    session.execute(query)\n",
    "except Exception as e:\n",
    "    print(e)\n",
    "                    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "file = 'event_datafile_new.csv'\n",
    "\n",
    "with open(file, encoding = 'utf8') as f:\n",
    "    csvreader = csv.DictReader(f)\n",
    "\n",
    "    for line in csvreader:\n",
    "        query = \"INSERT INTO session_plays (sessionid, iteminsession, artist, song, length)\"\n",
    "        query = query + \"VALUES (%s, %s, %s, %s, %s)\"\n",
    "        session.execute(query, (int(line[\"sessionId\"]), int(line[\"itemInSession\"]), line[\"artist\"], line[\"song\"], float(line[\"length\"])))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "editable": true,
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Artist: Faithless\n",
      "Song: Music Matters (Mark Knight Dub)\n",
      "Length: 495.307300 seconds (8.26 minutes)\n"
     ]
    }
   ],
   "source": [
    "query = \"SELECT artist, song, length FROM session_plays WHERE sessionid = 338 and iteminsession = 4\"\n",
    "\n",
    "try:\n",
    "    rows = session.execute(query)\n",
    "    for row in rows:\n",
    "        print (\"Artist: %s\\nSong: %s\\nLength: %f seconds (%.2f minutes)\" % (row.artist, row.song, row.length, row.length/60))\n",
    "except Exception as e:\n",
    "    print(e)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "#### Answer to Question 1:\n",
    "\n",
    "The song that was played during session id \"338\", associated with itemInSession \"4\", was \"Music Matters (Mark Knight Dub)\" by Faithless with a duration of 8.26 minutes."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "### Question 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",
    "\n",
    "We are asked to return a list of artist names and song titles ordered by `itemInSession`, along with the user's full name, for a specific user's session. In this case, we are going to focus on the **user's activity during a session**, thus, assigning the name `user_activity` to the table that holds the data to answer the question.\n",
    "\n",
    "For this requirement we'll use the following columns:\n",
    "1. userid (int): User's id\n",
    "2. sessionid (int): Session's id\n",
    "3. itemInSession (int): Position of the song in the session's history of songs played\n",
    "4. artist (varchar): Artist's name\n",
    "5. song (varchar): Song's title\n",
    "6. firstName (varchar): Firstname of the user associated with the session\n",
    "7. lastName (varchar): Lastname of the user associated with the session\n",
    "\n",
    "To partition the data and identify rows, we'll be using `userId` and `sessionId` as a compound primary key, since those are the attributes used to filter the data. To order the data in the partition, we'll be using `itemInSession` as clustering key.\n",
    "\n",
    "The SQL query that we'll be using for this is the following:\n",
    "\n",
    "```\n",
    "SELECT artist, song, firstname, lastname\n",
    "FROM user_activity WHERE userid = 10 and sessionid = 182\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "query = \"CREATE TABLE IF NOT EXISTS user_activity \\\n",
    "            (userid int, sessionid int, iteminsession int, artist varchar, song varchar, firstname varchar, lastname varchar, \\\n",
    "            PRIMARY KEY ((userid, sessionid), iteminsession))\"\n",
    "try:\n",
    "    session.execute(query)\n",
    "except Exception as e:\n",
    "    print(e)           "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "file = 'event_datafile_new.csv'\n",
    "\n",
    "with open(file, encoding = 'utf8') as f:\n",
    "    csvreader = csv.DictReader(f)\n",
    "\n",
    "    for line in csvreader:\n",
    "        query = \"INSERT INTO user_activity (userid, sessionid, iteminsession, artist, song, firstname, lastname)\"\n",
    "        query = query + \"VALUES (%s, %s, %s, %s, %s, %s, %s)\"\n",
    "        session.execute(query, (int(line[\"userId\"]), int(line[\"sessionId\"]), int(line[\"itemInSession\"]), line[\"artist\"], line[\"song\"], line[\"firstName\"], line[\"lastName\"]))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Artist: Down To The Bone | Song: Keep On Keepin' On | User: Sylvie Cruz\n",
      "Artist: Three Drives | Song: Greece 2000 | User: Sylvie Cruz\n",
      "Artist: Sebastien Tellier | Song: Kilometer | User: Sylvie Cruz\n",
      "Artist: Lonnie Gordon | Song: Catch You Baby (Steve Pitron & Max Sanna Radio Edit) | User: Sylvie Cruz\n"
     ]
    }
   ],
   "source": [
    "query = \"SELECT artist, song, firstname, lastname \\\n",
    "            FROM user_activity WHERE userid = 10 and sessionid = 182\"\n",
    "\n",
    "try:\n",
    "    rows = session.execute(query)\n",
    "    for row in rows:\n",
    "        print (\"Artist: %s | Song: %s | User: %s %s\" % (row.artist, row.song, row.firstname, row.lastname,))\n",
    "except Exception as e:\n",
    "    print(e)          "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "#### Answer to Question 2:\n",
    "\n",
    "The user Sylvie Cruz (id: 10) played the following songs during session id \"182\":\n",
    "- \"Keep On Keepin' On\" by Down To The Bone\n",
    "- \"Greece 2000\" by Three Drives\n",
    "- \"Kilometer\" by Sebastien Tellier\n",
    "- \"Catch You Baby (Steve Pitron & Max Sanna Radio Edit)\" by Lonnie Gordon"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "### Question 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",
    "\n",
    "Finally, we were required to list all the users that listened to the song \"All Hands Against His Own\". This will required that we model after the relation between songs and users that listened to them. For this, the table will be called `song_plays` and we'll be using the following columns:\n",
    "\n",
    "1. song (varchar): Song title\n",
    "2. userId (int): User's id\n",
    "3. firstName (varchar): User's firstname\n",
    "4. lastName (varchar): User's lastname\n",
    "\n",
    "Since the we'll be searching using the title of the song, the primary key for the table will be `song`. Additionally, we will include `userId` as clustering key to identify each user that listened to the song (otherwise every song would have just 1 user due to Cassandra's insert/update behaviour on primary keys).\n",
    "\n",
    "Translating this requirement into a SQL query results in the following: \n",
    "\n",
    "`SELECT firstname, lastname FROM song_plays WHERE song = 'All Hands Against His Own'`\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "query = \"CREATE TABLE IF NOT EXISTS song_plays \\\n",
    "            (song varchar, userid int, firstname varchar, lastname varchar, \\\n",
    "            PRIMARY KEY((song), userid))\"\n",
    "\n",
    "try:\n",
    "    session.execute(query)\n",
    "except Exception as e:\n",
    "    print(e)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "file = 'event_datafile_new.csv'\n",
    "\n",
    "with open(file, encoding = 'utf8') as f:\n",
    "    csvreader = csv.DictReader(f)\n",
    "\n",
    "    for line in csvreader:\n",
    "        query = \"INSERT INTO song_plays (song, userid, firstname, lastname)\"\n",
    "        query = query + \"VALUES (%s, %s, %s, %s)\"\n",
    "        session.execute(query, (line[\"song\"], int(line[\"userId\"]), line[\"firstName\"], line[\"lastName\"]))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "User: Jacqueline Lynch\n",
      "User: Tegan Levine\n",
      "User: Sara Johnson\n"
     ]
    }
   ],
   "source": [
    "query = \"SELECT firstname, lastname FROM song_plays WHERE song = 'All Hands Against His Own'\"\n",
    "\n",
    "try:\n",
    "    rows = session.execute(query)\n",
    "    for row in rows:\n",
    "        print (\"User: %s %s\" % (row.firstname, row.lastname,))\n",
    "except Exception as e:\n",
    "    print(e)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "#### Answer to Question 3\n",
    "\n",
    "The song \"All Hands Against His Own\" was listened by Jacqueline Lynch, Tegan Levine and Sara Johnson"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "### Drop the tables before closing out the sessions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "query = \"DROP TABLE session_plays\"\n",
    "try:\n",
    "    rows = session.execute(query)\n",
    "except Exception as e:\n",
    "    print(e)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "query = \"DROP TABLE user_activity\"\n",
    "try:\n",
    "    rows = session.execute(query)\n",
    "except Exception as e:\n",
    "    print(e)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "query = \"DROP TABLE song_plays\"\n",
    "try:\n",
    "    rows = session.execute(query)\n",
    "except Exception as e:\n",
    "    print(e)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "### Close the session and cluster connection¶"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "session.shutdown()\n",
    "cluster.shutdown()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "editable": true
   },
   "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": 4
}