Project_1B_ Project_Template.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "source": [
    "# Part I. ETL Pipeline for Pre-Processing the Files"
   ],
   "metadata": {
    "editable": true
   }
  },
  {
   "cell_type": "markdown",
   "source": [
    "## PLEASE RUN THE FOLLOWING CODE FOR PRE-PROCESSING THE FILES"
   ],
   "metadata": {
    "editable": true
   }
  },
  {
   "cell_type": "markdown",
   "source": [
    "#### Import Python packages "
   ],
   "metadata": {
    "editable": true
   }
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "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\n",
    "# to visualize progress\n",
    "from tqdm import tqdm"
   ],
   "outputs": [],
   "metadata": {
    "editable": true
   }
  },
  {
   "cell_type": "markdown",
   "source": [
    "#### Creating list of filepaths to process original event csv data files"
   ],
   "metadata": {
    "editable": true
   }
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "source": [
    "print('working directory is', os.getcwd())\n",
    "print(\"\")\n",
    "\n",
    "filepath = os.getcwd() + '/event_data'\n",
    "file_path_list = glob.glob(os.path.join(filepath,'*.csv'))\n",
    "print(file_path_list)"
   ],
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "text": [
      "working directory is /home/workspace\n",
      "\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"
     ]
    }
   ],
   "metadata": {
    "editable": true
   }
  },
  {
   "cell_type": "markdown",
   "source": [
    "#### Processing the files to create the data file csv that will be used for Apache Casssandra tables"
   ],
   "metadata": {
    "editable": true
   }
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "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",
    "            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"
   ],
   "outputs": [],
   "metadata": {
    "editable": true
   }
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "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))"
   ],
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "text": [
      "6821\n"
     ]
    }
   ],
   "metadata": {
    "editable": true
   }
  },
  {
   "cell_type": "markdown",
   "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\">"
   ],
   "metadata": {
    "editable": true
   }
  },
  {
   "cell_type": "markdown",
   "source": [
    "#### Creating a Cluster"
   ],
   "metadata": {
    "editable": true
   }
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "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(['127.0.0.1'])\n",
    "\n",
    "# To establish connection and begin executing queries, need a session\n",
    "session = cluster.connect()"
   ],
   "outputs": [],
   "metadata": {
    "editable": true
   }
  },
  {
   "cell_type": "markdown",
   "source": [
    "#### Create Keyspace"
   ],
   "metadata": {
    "editable": true
   }
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "source": [
    "try:\n",
    "    session.execute(\"\"\"\n",
    "    CREATE KEYSPACE IF NOT EXISTS udacity \n",
    "    WITH REPLICATION = \n",
    "    { 'class' : 'SimpleStrategy', 'replication_factor' : 1 }\"\"\"\n",
    ")\n",
    "\n",
    "except Exception as e:\n",
    "    print(e)"
   ],
   "outputs": [],
   "metadata": {
    "editable": true
   }
  },
  {
   "cell_type": "markdown",
   "source": [
    "#### Set Keyspace"
   ],
   "metadata": {
    "editable": true
   }
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "source": [
    "try:\n",
    "    session.set_keyspace('udacity')\n",
    "except Exception as e:\n",
    "    print(e)"
   ],
   "outputs": [],
   "metadata": {
    "editable": true
   }
  },
  {
   "cell_type": "markdown",
   "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"
   ],
   "metadata": {
    "editable": true
   }
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "query = \"CREATE TABLE IF NOT EXISTS song_table \"\n",
    "query = query + \"(artist text, song text, length float, sessionId int, \\\n",
    "itemInSession int, PRIMARY KEY (sessionId, itemInSession));\"\n",
    "session.execute(query)"
   ],
   "outputs": [
    {
     "output_type": "execute_result",
     "data": {
      "text/plain": [
       "<cassandra.cluster.ResultSet at 0x7ffbc9b17048>"
      ]
     },
     "metadata": {},
     "execution_count": 8
    }
   ],
   "metadata": {
    "editable": true
   }
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "source": [
    "# line_index: column_name\n",
    "# 0 artist\n",
    "# 1 firstName\n",
    "# 2 gender\n",
    "# 3 itemInSession\n",
    "# 4 lastName\n",
    "# 5 length\n",
    "# 6 level\n",
    "# 7 location\n",
    "# 8 sessionId\n",
    "# 9 song\n",
    "# 10 userId"
   ],
   "outputs": [],
   "metadata": {
    "editable": true
   }
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "source": [
    "# Insert rows into song_table\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 tqdm(csvreader):\n",
    "        query = \"INSERT INTO song_table (artist, song, length, sessionId, itemInSession) \"\n",
    "        query = query + \"VALUES (%s, %s, %s, %s, %s);\"\n",
    "        session.execute(query, (line[0], line[9], float(line[5]), int(line[8]), int(line[3])))"
   ],
   "outputs": [
    {
     "output_type": "stream",
     "name": "stderr",
     "text": [
      "6820it [00:11, 597.27it/s]\n"
     ]
    }
   ],
   "metadata": {
    "editable": true
   }
  },
  {
   "cell_type": "markdown",
   "source": [
    "#### Do a SELECT to verify that the data have been inserted into each table"
   ],
   "metadata": {
    "editable": true
   }
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "source": [
    "query = \"SELECT artist, song, length FROM song_table WHERE sessionId = 338 AND itemInSession = 4;\"\n",
    "res = session.execute(query)\n",
    "for r in res:\n",
    "    print(r)"
   ],
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "text": [
      "Row(artist='Faithless', song='Music Matters (Mark Knight Dub)', length=495.30731201171875)\n"
     ]
    }
   ],
   "metadata": {
    "editable": true,
    "scrolled": true
   }
  },
  {
   "cell_type": "markdown",
   "source": [
    "### 2. Give me only the following: name of artist, song (sorted by itemInSession) and user (first and last name) for userid = 10, sessionid = 182"
   ],
   "metadata": {
    "editable": true
   }
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "source": [
    "query = \"CREATE TABLE IF NOT EXISTS artist_table \"\n",
    "query = query + \"(artist text, song text, firstName text, lastName text,itemInSession int, \\\n",
    "userId int, sessionId int, PRIMARY KEY ((userId, sessionId), itemInSession));\"\n",
    "session.execute(query)"
   ],
   "outputs": [
    {
     "output_type": "execute_result",
     "data": {
      "text/plain": [
       "<cassandra.cluster.ResultSet at 0x7ffbc9b13240>"
      ]
     },
     "metadata": {},
     "execution_count": 12
    }
   ],
   "metadata": {
    "editable": true
   }
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "source": [
    "# Insert rows into artist_table\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 tqdm(csvreader):\n",
    "        query = \"INSERT INTO artist_table (artist, song, firstName, lastName, itemInSession, userId, sessionId) \"\n",
    "        query = query + \"VALUES (%s, %s, %s, %s, %s, %s, %s);\"\n",
    "        session.execute(query, (line[0], line[9], line[1], line[4], int(line[3]), int(line[10]), int(line[8])))"
   ],
   "outputs": [
    {
     "output_type": "stream",
     "name": "stderr",
     "text": [
      "6820it [00:10, 662.18it/s]\n"
     ]
    }
   ],
   "metadata": {
    "editable": true
   }
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "source": [
    "query = \"SELECT artist, song, firstName, lastName FROM artist_table \\\n",
    "WHERE userId = 10 AND sessionId = 182 ORDER BY itemInSession;\"\n",
    "res = session.execute(query)\n",
    "for r in res:\n",
    "    print(r)"
   ],
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "text": [
      "Row(artist='Down To The Bone', song=\"Keep On Keepin' On\", firstname='Sylvie', lastname='Cruz')\n",
      "Row(artist='Three Drives', song='Greece 2000', firstname='Sylvie', lastname='Cruz')\n",
      "Row(artist='Sebastien Tellier', song='Kilometer', firstname='Sylvie', lastname='Cruz')\n",
      "Row(artist='Lonnie Gordon', song='Catch You Baby (Steve Pitron & Max Sanna Radio Edit)', firstname='Sylvie', lastname='Cruz')\n"
     ]
    }
   ],
   "metadata": {
    "editable": true
   }
  },
  {
   "cell_type": "markdown",
   "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'"
   ],
   "metadata": {
    "editable": true
   }
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "source": [
    "query = \"CREATE TABLE IF NOT EXISTS user_table \"\n",
    "query = query + \"(firstName text, lastName text, song text, userId int, PRIMARY KEY (song, userId));\"\n",
    "session.execute(query)"
   ],
   "outputs": [
    {
     "output_type": "execute_result",
     "data": {
      "text/plain": [
       "<cassandra.cluster.ResultSet at 0x7ffbc9b16278>"
      ]
     },
     "metadata": {},
     "execution_count": 15
    }
   ],
   "metadata": {
    "editable": true
   }
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "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 tqdm(csvreader):\n",
    "        query = \"INSERT INTO user_table (firstName, lastName, song, userId) \"\n",
    "        query = query + \"VALUES (%s, %s, %s, %s);\"\n",
    "        session.execute( query, (line[1], line[4], line[9], int(line[10])) )"
   ],
   "outputs": [
    {
     "output_type": "stream",
     "name": "stderr",
     "text": [
      "6820it [00:09, 685.59it/s]\n"
     ]
    }
   ],
   "metadata": {
    "editable": true
   }
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "source": [
    "query = \"SELECT firstName, lastName FROM user_table WHERE song = 'All Hands Against His Own';\"\n",
    "row = session.execute(query)\n",
    "for r in row:\n",
    "    print(r)"
   ],
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "text": [
      "Row(firstname='Jacqueline', lastname='Lynch')\n",
      "Row(firstname='Tegan', lastname='Levine')\n",
      "Row(firstname='Sara', lastname='Johnson')\n"
     ]
    }
   ],
   "metadata": {
    "editable": true
   }
  },
  {
   "cell_type": "markdown",
   "source": [
    "### Drop the tables before closing out the sessions"
   ],
   "metadata": {
    "editable": true
   }
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "source": [
    "query1 = \"DROP TABLE IF EXISTS song_table;\"\n",
    "query2 = \"DROP TABLE IF EXISTS artist_table;\"\n",
    "query3 = \"DROP TABLE IF EXISTS user_table;\"\n",
    "session.execute(query1)\n",
    "session.execute(query2)\n",
    "session.execute(query3)"
   ],
   "outputs": [
    {
     "output_type": "execute_result",
     "data": {
      "text/plain": [
       "<cassandra.cluster.ResultSet at 0x7ffbc9b16358>"
      ]
     },
     "metadata": {},
     "execution_count": 18
    }
   ],
   "metadata": {
    "editable": true
   }
  },
  {
   "cell_type": "markdown",
   "source": [
    "### Close the session and cluster connection¶"
   ],
   "metadata": {
    "editable": true
   }
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "source": [
    "session.shutdown()\n",
    "cluster.shutdown()"
   ],
   "outputs": [],
   "metadata": {
    "editable": true
   }
  }
 ],
 "metadata": {
  "kernelspec": {
   "name": "python3",
   "display_name": "Python 3.8.2 64-bit"
  },
  "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.8.2"
  },
  "interpreter": {
   "hash": "31f2aee4e71d21fbe5cf8b01ff0e069b9275f58929596ceb00d14d90e3e16cd6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}