Cassandra_ETL.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": 28,
   "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": 29,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "8056\n"
     ]
    }
   ],
   "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_new 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": 2,
   "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. Completing the Apache Cassandra coding portion of the project. \n",
    "\n",
    "## Now we're 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": 2,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# This should make a connection to a Cassandra instance on 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": 3,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "try:\n",
    "    session.execute(\"\"\"\n",
    "    CREATE KEYSPACE IF NOT EXISTS project_1b \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": 4,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "try:\n",
    "    session.set_keyspace('project_1b')\n",
    "except Exception as e:\n",
    "    print(e)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "### Now we need to create tables to run the following queries by modeling 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": [
    "## Query 1:\n",
    "\n",
    "### How should we model this data? What should be our Primary Key and Partition Key? \n",
    "\n",
    "#### Since we're looking for data based off the session_id let's set the partition key as the `sessionId`. Then the clustering column can be `item_in_session`, as we're also filtering based on that. Since every item in a specific `session_id` should only occur once this will set our rows to be unique."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "### Create Tables"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "query = \"CREATE TABLE IF NOT EXISTS song_details_library \"\n",
    "query = query + \"(session_id int, item_in_session int, artist text, song text, length float, PRIMARY KEY (session_id, item_in_session))\"\n",
    "try:\n",
    "    session.execute(query)\n",
    "except Exception as e:\n",
    "    print(e)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "### Insert the data into the table\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "editable": true
   },
   "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_details_library (session_id, item_in_session, 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 has been inserted into each table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "editable": true,
    "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>artist_name</th>\n",
       "      <th>song_title</th>\n",
       "      <th>song_length</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\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": [
       "  artist_name                       song_title  song_length\n",
       "0   Faithless  Music Matters (Mark Knight Dub)   495.307312"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "query = \"select artist, song, length from song_details_library WHERE session_id=338 and item_in_session=4\"\n",
    "try:\n",
    "    rows = session.execute(query)\n",
    "except Exception as e:\n",
    "    print(e)\n",
    "\n",
    "song_details_df = pd.DataFrame([rows[0]], columns=['artist_name', 'song_title', 'song_length'])\n",
    "song_details_df.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "## Query 2:\n",
    "\n",
    "### How should we model this data? What should be our Primary Key and Partition Key? \n",
    "\n",
    "#### Since we're looking for data based off the user_id let's set a composite partition key of `user_id` and `session_id`. We're adding `session_id` to the partition key for performance, so that sessions from the same user get stored on the same node. Then since we need to sort songs by `item_in_session` lets set that as the clustering column."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "### Create Tables"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "query = \"CREATE TABLE IF NOT EXISTS user_song_library \"\n",
    "query = query + \"(user_id int, session_id int, item_in_session int, artist text, song text, user_lname text, \\\n",
    "                    user_fname text, PRIMARY KEY ((user_id, session_id), item_in_session))\"\n",
    "try:\n",
    "    session.execute(query)\n",
    "except Exception as e:\n",
    "    print(e)        "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "### Insert the data into the table\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "editable": true
   },
   "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 user_song_library (user_id, session_id, item_in_session, artist, song, user_lname, user_fname)\"\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[4], line[1]))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "\n",
    "### Do a SELECT to verify that the data has been inserted into each table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "editable": 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>artist_name</th>\n",
       "      <th>song_title</th>\n",
       "      <th>user_fname</th>\n",
       "      <th>user_lname</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>Down To The Bone</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>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>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>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_name                                         song_title  \\\n",
       "0   Down To The Bone                                 Keep On Keepin' On   \n",
       "1       Three Drives                                        Greece 2000   \n",
       "2  Sebastien Tellier                                          Kilometer   \n",
       "3      Lonnie Gordon  Catch You Baby (Steve Pitron & Max Sanna Radio...   \n",
       "\n",
       "  user_fname user_lname  \n",
       "0     Sylvie       Cruz  \n",
       "1     Sylvie       Cruz  \n",
       "2     Sylvie       Cruz  \n",
       "3     Sylvie       Cruz  "
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "query = \"select artist, song, user_fname, user_lname from user_song_library WHERE user_id=10 and session_id=182\"\n",
    "try:\n",
    "    rows = session.execute(query)\n",
    "except Exception as e:\n",
    "    print(e)\n",
    "\n",
    "rows_data = [list(x) for x in rows]\n",
    "    \n",
    "user_song_df = pd.DataFrame(rows_data, columns=['artist_name', 'song_title', 'user_fname', 'user_lname'])\n",
    "user_song_df.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "## Query 3:\n",
    "\n",
    "### How should we model this data? What should be our Primary Key and Partition Key? \n",
    "\n",
    "#### Since we're looking for data based off the song let's set the partition key as `song`. Then to ensure uniqueness let's add `user_id` because we want to get the unique users who listened to the song \"All Hands Against His Own\". We shouldn't use `user_lname` and `user_fname` as clustering columns because there could be multiple users with the same name.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "### Create Tables"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "query = \"CREATE TABLE IF NOT EXISTS users_for_song \"\n",
    "query = query + \"(song text, user_id int, user_lname text, user_fname text, PRIMARY KEY (song, user_id))\"\n",
    "try:\n",
    "    session.execute(query)\n",
    "except Exception as e:\n",
    "    print(e)        \n",
    "                    "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "### Insert the data into the table\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "editable": true
   },
   "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_for_song (song, user_id, user_lname, user_fname)\"\n",
    "        query = query + \"VALUES (%s, %s, %s, %s)\"\n",
    "        session.execute(query, (line[9], int(line[10]), line[4], line[1]))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "#### Do a SELECT to verify that the data has been inserted into each table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "editable": 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>user_fname</th>\n",
       "      <th>user_lname</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": [
       "   user_fname user_lname\n",
       "0  Jacqueline      Lynch\n",
       "1       Tegan     Levine\n",
       "2        Sara    Johnson"
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "query = \"select user_fname, user_lname from users_for_song WHERE song='All Hands Against His Own'\"\n",
    "try:\n",
    "    rows = session.execute(query)\n",
    "except Exception as e:\n",
    "    print(e)\n",
    "\n",
    "rows_data = [list(x) for x in rows]\n",
    "\n",
    "users_for_song_df = pd.DataFrame(rows_data, columns=['user_fname', 'user_lname'])\n",
    "users_for_song_df.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "### Drop the tables before closing out the sessions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "query = \"drop table song_details_library\"\n",
    "try:\n",
    "    session.execute(query)\n",
    "except Exception as e:\n",
    "    print(e)\n",
    "    \n",
    "query = \"drop table user_song_library\"\n",
    "try:\n",
    "    session.execute(query)\n",
    "except Exception as e:\n",
    "    print(e)\n",
    "\n",
    "query = \"drop table users_for_song\"\n",
    "try:\n",
    "    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": 11,
   "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
}