Project_1B_ Project_Template.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "# Part I. ETL Pipeline for Pre-Processing the Files"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "## PLEASE RUN THE FOLLOWING CODE FOR PRE-PROCESSING THE FILES"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "#### Import Python packages "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "editable": true
   },
   "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": {
    "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",
      "['/home/workspace/event_data/2018-11-27-events.csv', '/home/workspace/event_data/2018-11-04-events.csv', '/home/workspace/event_data/2018-11-07-events.csv', '/home/workspace/event_data/2018-11-09-events.csv', '/home/workspace/event_data/2018-11-19-events.csv', '/home/workspace/event_data/2018-11-05-events.csv', '/home/workspace/event_data/2018-11-22-events.csv', '/home/workspace/event_data/2018-11-16-events.csv', '/home/workspace/event_data/2018-11-26-events.csv', '/home/workspace/event_data/2018-11-24-events.csv', '/home/workspace/event_data/2018-11-29-events.csv', '/home/workspace/event_data/2018-11-15-events.csv', '/home/workspace/event_data/2018-11-20-events.csv', '/home/workspace/event_data/2018-11-06-events.csv', '/home/workspace/event_data/2018-11-18-events.csv', '/home/workspace/event_data/2018-11-21-events.csv', '/home/workspace/event_data/2018-11-10-events.csv', '/home/workspace/event_data/2018-11-23-events.csv', '/home/workspace/event_data/2018-11-02-events.csv', '/home/workspace/event_data/2018-11-28-events.csv', '/home/workspace/event_data/2018-11-03-events.csv', '/home/workspace/event_data/2018-11-13-events.csv', '/home/workspace/event_data/2018-11-30-events.csv', '/home/workspace/event_data/2018-11-12-events.csv', '/home/workspace/event_data/2018-11-01-events.csv', '/home/workspace/event_data/2018-11-14-events.csv', '/home/workspace/event_data/2018-11-25-events.csv', '/home/workspace/event_data/2018-11-08-events.csv', '/home/workspace/event_data/2018-11-17-events.csv', '/home/workspace/event_data/2018-11-11-events.csv']\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": {
    "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": [],
   "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",
    "# 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": {
    "editable": true
   },
   "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": {
    "editable": true
   },
   "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 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": [
    "# 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": "code",
   "execution_count": 6,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<cassandra.cluster.Session object at 0x7fd045c0e358>\n"
     ]
    }
   ],
   "source": [
    "print(session)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "#### Create Keyspace"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<cassandra.cluster.ResultSet at 0x7fd045bf05f8>"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "session.execute(\"\"\"\n",
    "    CREATE KEYSPACE IF NOT EXISTS sparkify \n",
    "    WITH REPLICATION = \n",
    "    { 'class' : 'SimpleStrategy', 'replication_factor' : 1 }\n",
    "\"\"\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "#### Set Keyspace"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "session.set_keyspace('sparkify')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "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": {
    "editable": true
   },
   "source": [
    "### song_plays_by_session\n",
    "\n",
    "Desired Query:\n",
    "\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",
    "Our table will need artist, song, length, because that is the information that will be requested. We also need sessionId, itemInSession because that is how the results will be filtered.\n",
    "\n",
    "Partition Key:\n",
    "\n",
    "- Should be sessionId so that our data is distributed by session rather than distributed by position in session (itemInSession).\n",
    "Primary Key:\n",
    "\n",
    "- sessionId by itself will not be unique, but if we add itemInSession as a clustering column, we can ensure uniqueness."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<cassandra.cluster.ResultSet at 0x7fd0337b4e10>"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "session.execute(\"\"\"\n",
    "    CREATE TABLE IF NOT EXISTS song_plays_by_session (\n",
    "        sessionId int,\n",
    "        itemInSession int,\n",
    "        artist text,\n",
    "        song text,\n",
    "        length float,\n",
    "        PRIMARY KEY (sessionId, itemInSession)\n",
    "    )\n",
    "\"\"\")                    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# We have provided part of the code to set up the CSV file. Please complete the Apache Cassandra code below#\n",
    "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_plays_by_session (sessionId, itemInSession, artist, song, length)\"\n",
    "        query = query + \" VALUES (%s, %s, %s, %s, %s)\"\n",
    "        session.execute(query, (int(line[8]), int(line[3]), line[0], line[9], float(line[5])))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "#### Do a SELECT to verify that the data have been inserted into each table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "editable": true,
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "['Faithless', 'Music Matters (Mark Knight Dub)', 495.30731201171875]\n"
     ]
    }
   ],
   "source": [
    "rows = session.execute(\"\"\"\n",
    "    SELECT artist, song, length\n",
    "    FROM song_plays_by_session\n",
    "    WHERE sessionId=338 AND itemInSession=4\n",
    "\"\"\")\n",
    "\n",
    "for row in rows:\n",
    "    print(list(row))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "### song_plays_by_user_session\n",
    "Desired Query:\n",
    "\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",
    "Our table will need artist, song, firstName, lastName, because that is the information that will be requested. We also need userId, sessionId, itemInSession because that is how the results will be filtered and ordered.\n",
    "\n",
    "Partition Key:\n",
    "\n",
    "- Should be userId and sessionId so that our data is distributed by user which is the first filter that is applied in the query. Sessions belonging to the same user might be in different nodes therefore we should use both userId and sessionId as partition keys so sessions from the same user are stored together\n",
    "\n",
    "Primary Key:\n",
    "\n",
    "- In this case we are filtering on two keys and then ordering on a third key, so our PK should be the partition key as well as two clustering columns, sessionId, itemInSession."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<cassandra.cluster.ResultSet at 0x7fd025f17828>"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "session.execute(\"\"\"\n",
    "    CREATE TABLE IF NOT EXISTS song_plays_by_user_session (\n",
    "        userId int,\n",
    "        sessionId int,\n",
    "        itemInSession int,\n",
    "        artist text,\n",
    "        song text,\n",
    "        firstName text,\n",
    "        lastName text,\n",
    "        PRIMARY KEY ((userId, sessionId), itemInSession)\n",
    "    )\n",
    "\"\"\")                "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "file = 'event_datafile_new.csv'\n",
    "\n",
    "# Insert expected data into song_plays_by_user_session\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_plays_by_user_session (userId, sessionId, itemInSession, artist, song, firstName, lastName)\"\n",
    "        query = query + \" VALUES (%s, %s, %s, %s, %s, %s, %s)\"\n",
    "        session.execute(query, (int(line[10]), int(line[8]), int(line[3]), line[0], line[9], line[1], line[4]))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[0, 'Down To The Bone', \"Keep On Keepin' On\", 'Sylvie', 'Cruz']\n",
      "[1, 'Three Drives', 'Greece 2000', 'Sylvie', 'Cruz']\n",
      "[2, 'Sebastien Tellier', 'Kilometer', 'Sylvie', 'Cruz']\n",
      "[3, 'Lonnie Gordon', 'Catch You Baby (Steve Pitron & Max Sanna Radio Edit)', 'Sylvie', 'Cruz']\n"
     ]
    }
   ],
   "source": [
    "rows = session.execute(\"\"\"\n",
    "    SELECT itemInSession, artist, song, firstName, lastName\n",
    "    FROM song_plays_by_user_session \n",
    "    WHERE userId=10 AND sessionId=182\n",
    "\"\"\")\n",
    "\n",
    "for row in rows:\n",
    "    # Convert named tuple into a list so we only see the values\n",
    "    print(list(row))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "### users_by_song_listens\n",
    "Desired Query:\n",
    "\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",
    "Our table will need firstName, lastName, because that is the information that will be requested. We also need song, userId for our Primary Key.\n",
    "\n",
    "Partition Key:\n",
    "\n",
    "- Should be song so that our data is distributed by song which is the filter that is applied in the query.\n",
    "Primary Key:\n",
    "\n",
    "- I've chosen to add userId to the PK as a clustering column for two reasons:\n",
    " - I want to ensure a unique PK and (song, name) didn't seem sufficient\n",
    " - As a clustering column, it will ensure my results are ordered by id which is nice\n",
    "- If i instead wanted alphabetical results, I could use PRIMARY KEY (song, firstName, userId), which would still ensure uniqueness, but would order by firstName instead."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<cassandra.cluster.ResultSet at 0x7fd045bece10>"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "session.execute(\"\"\"\n",
    "    CREATE TABLE IF NOT EXISTS users_by_song_listens (\n",
    "        song text,\n",
    "        userId int,\n",
    "        firstName text,\n",
    "        lastName text,\n",
    "        PRIMARY KEY (song, userId)\n",
    "    )\n",
    "\"\"\")       "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "file = 'event_datafile_new.csv'\n",
    "\n",
    "# Insert expected data into users_by_song_listens\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_by_song_listens (song, userId, firstName, lastName)\"\n",
    "        query = query + \" VALUES (%s, %s, %s, %s)\"\n",
    "        session.execute(query, (line[9], int(line[10]), line[1], line[4]))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[29, 'Jacqueline', 'Lynch']\n",
      "[80, 'Tegan', 'Levine']\n",
      "[95, 'Sara', 'Johnson']\n"
     ]
    }
   ],
   "source": [
    "rows = session.execute(\"\"\"\n",
    "    SELECT userId, firstName, lastName\n",
    "    FROM users_by_song_listens\n",
    "    WHERE song='All Hands Against His Own'\n",
    "\"\"\")\n",
    "\n",
    "for row in rows:\n",
    "    # Convert named tuple into a list so we only see the values\n",
    "    print(list(row))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "### Drop the tables before closing out the sessions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<cassandra.cluster.ResultSet at 0x7fd01c2bcb38>"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "session.execute(\"drop table song_plays_by_session\")\n",
    "session.execute(\"drop table song_plays_by_user_session\")\n",
    "session.execute(\"drop table users_by_song_listens\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "### Close the session and cluster connection¶"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "session.shutdown()\n",
    "cluster.shutdown()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": []
  },
  {
   "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
}