Project 2 Data Modeling With Apache Cassandra/Project 2.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"
     ]
    }
   ],
   "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": 8,
   "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",
    "# 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": 9,
   "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": [
    "## Begin writing your Apache Cassandra code in the cells below"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "#### Creating a Cluster"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "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": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "#### Create Keyspace"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Creating a Keyspace \n",
    "\n",
    "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)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "#### Set Keyspace"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Setting the keyspace \n",
    "\n",
    "try:\n",
    "    session.set_keyspace('udacity')\n",
    "except Exception as e:\n",
    "    print(e)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "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": {
    "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": [
    "#### Creating the first table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "## Query 1:  Print artist, song title and song's length in the music app history that was heard during \\\n",
    "## sessionId = 338, and itemInSession = 4\n",
    "\n",
    "# A table specific to the query has to be created \n",
    "# Here the PRIMARY KEY is chosen to be ((session_id), item_in_session) being session_id the partition key\n",
    "# We are querying on these two attributes and session_id will be the better choice as PARTITION KEY as it will be hashed to find the cluster\n",
    "# Since we need to query on item_in_session, it is chosen to be a clustering column and is added in the PRIMARY KEY \n",
    "\n",
    "drop = \"DROP TABLE IF EXISTS music_library \"\n",
    "create = \"CREATE TABLE IF NOT EXISTS music_library \"\n",
    "create = create + \"(session_id int, item_in_session int, artist_name varchar, song_title varchar, song_duration float  \\\n",
    "                    ,PRIMARY KEY ((session_id), item_in_session))                                                      ;\"\n",
    "\n",
    "try:\n",
    "    session.execute(drop)\n",
    "    session.execute(create)\n",
    "except Exception as e:\n",
    "    print(e)\n",
    "\n",
    "                    "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "#### Inserting the data into first table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "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",
    "## Inserting values into the table\n",
    "        query = \"INSERT INTO music_library (session_id, item_in_session, artist_name, song_title, song_duration)\"\n",
    "        query = query + \" VALUES (%s, %s, %s, %s, %s)\"\n",
    "        ## Assigning the column elements with each column in the INSERT statement.\n",
    "        ## For e.g., to INSERT artist_name and user first_name, you would change the code below to `line[0], line[1]`\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 first table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {
    "editable": true,
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Faithless Music Matters (Mark Knight Dub) 495.30731201171875\n"
     ]
    }
   ],
   "source": [
    "## Executing the select query to verify if the data was entered into the table\n",
    "\n",
    "query = \"SELECT artist_name, song_title, song_duration \\\n",
    "         FROM music_library                            \\\n",
    "         WHERE SESSION_ID = 338                        \\\n",
    "             AND ITEM_IN_SESSION = 4                   ;\"               \n",
    "try:\n",
    "    rows = session.execute(query)\n",
    "except Exception as e:\n",
    "    print(e)\n",
    "    \n",
    "for row in rows:\n",
    "    print (row.artist_name, row.song_title, row.song_duration)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "#### Creating the second table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "## Select name of artist, song (sorted by itemInSession) and user (first and last name)\\\n",
    "## for userid = 10, sessionid = 182\n",
    "\n",
    "# A table specific to the query has to be created \n",
    "# Here the PRIMARY KEY is chosen to be ((user_id, session_id), item_in_session) being (user_id, sessionId) the partition key\n",
    "# We are querying on these two attributes and we want the results to be sorted in order of item_in_session\n",
    "# If only user_id was used as PARTITION KEY, then there exists a high chance of data aggregating in a single node (depends on data)\n",
    "# Hence using (user_id, session_id) as PARTITION KEY, it is more likely to be distributed across nodes as session_id changes periodically\n",
    "# Since the output needs to be sorted in item_in_session, it is chosen to be a clustering column and is added in the PRIMARY KEY  \n",
    "\n",
    "drop = \"DROP TABLE IF EXISTS artist_library \"\n",
    "create = \"CREATE TABLE IF NOT EXISTS artist_library \"\n",
    "create = create + \"(user_id int, session_id int, item_in_session int, artist_name varchar, song_title text   \\\n",
    "                    ,first_name varchar, last_name varchar                                                   \\\n",
    "                    ,PRIMARY KEY ((user_id, session_id), item_in_session))                                   ;\"\n",
    "\n",
    "try:\n",
    "    session.execute(drop)\n",
    "    session.execute(create)\n",
    "except Exception as e:\n",
    "    print(e)                  \n",
    "\n",
    "\n",
    "                    "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "#### Inserting the data in the second table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "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",
    "## Inserting values into the table\n",
    "        query = \"INSERT INTO artist_library (user_id, session_id, item_in_session, artist_name    \\\n",
    "                                             ,song_title, first_name, last_name)                  \"\n",
    "        query = query + \" VALUES (%s, %s, %s, %s, %s, %s, %s)\"\n",
    "        ## Assigning the column elements with each column in the INSERT statement.\n",
    "        ## For e.g., to INSERT artist_name and user first_name, you would change the code below to `line[0], line[1]`\n",
    "        session.execute(query, (int(line[10]), int(line[8]), int(line[3]), line[0], line[9], line[1], line[4]))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "#### Do a SELECT to verify that the data have been inserted into second table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "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"
     ]
    }
   ],
   "source": [
    "## Executing the select query to verify if the data was entered into the table\n",
    "\n",
    "query = \"SELECT artist_name, song_title, first_name, last_name \\\n",
    "         FROM artist_library                                   \\\n",
    "         WHERE user_id = 10                                    \\\n",
    "             AND session_id = 182                              ;\"               \n",
    "try:\n",
    "    rows = session.execute(query)\n",
    "except Exception as e:\n",
    "    print(e)\n",
    "    \n",
    "for row in rows:\n",
    "    print (row.artist_name, row.song_title, row.first_name, row.last_name)\n",
    "                    "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "### Creating the third table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 66,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "## Select every user name (first and last) in my music app history who listened to the song 'All Hands Against His Own'\n",
    "\n",
    "# A table specific to the query has to be created \n",
    "# Here the PRIMARY KEY is chosen to be ((song_title), user_id) being song_title the partition key\n",
    "# We are querying on the song_title and hence it should be the PARTITION KEY\n",
    "# If only song_title was as PRIMARY KEY, since Cassandra does not allow duplicate entries, the entries will be overwritten\n",
    "# Hence using (song_title, user_id) as PRIMARY KEY, we can overcome this issue without violating the query constraints\n",
    "\n",
    "\n",
    "drop = \"DROP TABLE IF EXISTS play_history \"\n",
    "create = \"CREATE TABLE IF NOT EXISTS play_history \"\n",
    "create = create + \"(song_title text, user_id int, first_name varchar, last_name varchar    \\\n",
    "                  ,PRIMARY KEY ((song_title), user_id))                                    ;\"\n",
    "\n",
    "try:\n",
    "    session.execute(drop)\n",
    "    session.execute(create)\n",
    "except Exception as exep:\n",
    "    print(exep)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "#### Inserting the data in the third table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 67,
   "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",
    "    \n",
    "    for line in csvreader:\n",
    "        # Insert into the table\n",
    "        query = \"INSERT INTO play_history (song_title, user_id, first_name, last_name) \"\n",
    "        query = query + \"VALUES (%s, %s, %s, %s)\"\n",
    "        \n",
    "        # Properly assign the values from CSV to the table attributes\n",
    "        try:\n",
    "            session.execute(query, (line[9], int(line[10]), line[1], line[4]))\n",
    "        except Exception as exep:\n",
    "            print(exep)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "#### Do a SELECT to verify that the data have been inserted into third table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 69,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Jacqueline Lynch\n",
      "Tegan Levine\n",
      "Sara Johnson\n"
     ]
    }
   ],
   "source": [
    "## Executing the select query to verify if the data was entered into the table\n",
    "\n",
    "select_query = \"SELECT first_name, last_name                   \\\n",
    "                FROM play_history                              \\\n",
    "                WHERE song_title='All Hands Against His Own'   ;\"\n",
    "\n",
    "try:\n",
    "    rows = session.execute(select_query)\n",
    "except Exception as exep:\n",
    "    print(exep)\n",
    "\n",
    "# Print the results\n",
    "for row in rows:\n",
    "    print(row.first_name, row.last_name)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "### Drop the tables before closing out the sessions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "## Dropping all the tables"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 70,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "try:\n",
    "    session.execute(\"DROP TABLE IF EXISTS music_library\")\n",
    "except Execption as exep:\n",
    "    print(exep)\n",
    "    \n",
    "try:\n",
    "    session.execute(\"DROP TABLE IF EXISTS artist_library\")\n",
    "except Execption as exep:\n",
    "    print(exep)\n",
    "    \n",
    "try:\n",
    "    session.execute(\"DROP TABLE IF EXISTS play_history\")\n",
    "except Execption as exep:\n",
    "    print(exep)  "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "### Close the session and cluster connection¶"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 71,
   "metadata": {
    "editable": true
   },
   "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": 4
}