Project_1B_Cassandra.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Part I. ETL Pipeline for Pre-Processing the Files"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Import Python packages "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "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\n",
    "from decimal import Decimal "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Creating list of filepaths to process original event csv data files"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "/Users/StriderKeni/Documents/Data_Science/Data_Engineering/Projects/cassandra-ETL\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": {},
   "source": [
    "#### Processing the files to create the data file csv that will be used for Apache Casssandra tables"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "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_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": {},
   "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",
    "    # the number below include header\n",
    "    print(sum(1 for line in f))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Part II. Modeling NoSQL Database\n",
    "\n",
    "## event_datafile_new.csv 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 the file <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": {},
   "source": [
    "#### Creating a Cluster"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Creating a cluster using local machine (127.0.0.1)\n",
    "from cassandra.cluster import Cluster\n",
    "try:\n",
    "    cluster = Cluster(['127.0.0.1'])\n",
    "    # Creating a session variable to  establish connection and begin executing queries\n",
    "    session = cluster.connect()\n",
    "except Exception as e:\n",
    "    print(e)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Create Keyspace"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Keyspace for NoSQL database\n",
    "try:\n",
    "    session.execute(\"\"\"\n",
    "    CREATE KEYSPACE IF NOT EXISTS sparkifydb\n",
    "    WITH REPLICATION =\n",
    "    {'class': 'SimpleStrategy', 'replication_factor': 1}\n",
    "    \"\"\")\n",
    "except Exception as e:\n",
    "    print(e)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Set Keyspace"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Set KEYSPACE to the keyspace specified above\n",
    "## This allow to start using and executing queries to the keyspace\n",
    "try:\n",
    "    session.set_keyspace('sparkifydb')\n",
    "except Exception as e:\n",
    "    print(e)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Queries based on needs of the Analytic Team\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": {},
   "source": [
    "### Creating Data Model based on Queries Above"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "errors={'127.0.0.1': 'Client request timeout. See Session.execute[_async](timeout)'}, last_host=127.0.0.1\n",
      "Table songs_user_session successfully dropped\n",
      "Table usernames_library successfully dropped\n"
     ]
    }
   ],
   "source": [
    "# Drop table statement to be sure that is not created before. \n",
    "table_names = ['music_playlist_session', 'songs_user_session', 'usernames_library']\n",
    "\n",
    "for table_name in table_names:\n",
    "    try:\n",
    "        drop_query = 'DROP TABLE IF EXISTS {}'.format(table_name)\n",
    "        session.execute(drop_query)\n",
    "        print(\"Table {} successfully dropped\".format(table_name))\n",
    "    except Exception as e:\n",
    "        print(e)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Function to insert data into tables"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "def insert_data(query, session, values, dtypes, file):\n",
    "    \n",
    "    '''\n",
    "    Pipeline to insert data to Apache Cassandra table from CSV File.\n",
    "    Args:\n",
    "        query (str): INSERT statement query\n",
    "        session (object): Apache Cassandra Session\n",
    "        values (list): list of indices to pass to INSERT statement\n",
    "        dtype (list): list of values to convert `values`\n",
    "        file (str): CSV file path\n",
    "    '''\n",
    "    # Validate that len values is equal to dtypes, if not, it's not possible to run the query.\n",
    "    assert(len(values) == len(dtypes))\n",
    "    # open csv file\n",
    "    with open(file, encoding = 'utf8') as f:\n",
    "        csvreader = csv.reader(f)\n",
    "        next(csvreader) # skip header\n",
    "        for line in csvreader:\n",
    "            session.execute(query, tuple([dtype(line[x])  for x, dtype in zip(values, dtypes)]))\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## First 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"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [],
   "source": [
    "# CREATE TABLE statement for first table based on first query.\n",
    "\n",
    "query_1 = \"CREATE TABLE IF NOT EXISTS music_playlist_session\"\n",
    "query_1 = query_1 + \"(session_id int, item_session int, user_id int, first_name text, last_name text, artist text, song text, length decimal, PRIMARY KEY(session_id, item_session))\"\n",
    "\n",
    "'''\n",
    "\n",
    "We need the artist, song and length of every song. Filtered by session_id and item_session.\n",
    "Based on those filters, we are going to create a composite primary key using session_id and item_session\n",
    "(session_id, item_session). There's no need to order the output data\n",
    "that's why we are not going to use any clustering key.\n",
    "\n",
    "'''\n",
    "\n",
    "try:\n",
    "    session.execute(query_1)\n",
    "except Exception as e:\n",
    "    print(e)                    "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### INSERT code for table `music_playlist_session`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "# file name\n",
    "file = 'event_datafile_new.csv'\n",
    "#query\n",
    "query = \"INSERT INTO music_playlist_session (session_id, item_session, user_id, first_name, last_name, artist, song, length)\"\n",
    "query = query + \"VALUES (%s, %s, %s, %s, %s, %s, %s, %s)\"\n",
    "# values and data types\n",
    "values = [8,3,10,1,4,0,9,5]\n",
    "dtypes = [int, int, int, str,str,str,str,Decimal]\n",
    "\n",
    "insert_data(query, session, values, dtypes, file)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### SELECT to verify that the data have been inserted into each table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "6820 rows inserted\n"
     ]
    }
   ],
   "source": [
    "## Select statement to verify if the data was inserted into the table\n",
    "try:\n",
    "    rows = session.execute(\"SELECT * FROM music_playlist_session\")\n",
    "except Exception as e:\n",
    "    print(e)\n",
    "    \n",
    "print(\"{} rows inserted\".format(len(list(rows))))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Faithless Music Matters (Mark Knight Dub) 495.3073\n"
     ]
    }
   ],
   "source": [
    "# Select First Query\n",
    "# Query 1:  Give me the artist, song title and song's length in the music app history that was heard during \\\n",
    "## sessionId = 338, and itemInSession = 4\n",
    "try:\n",
    "    results_1 = session.execute(\"SELECT artist, song, length FROM music_playlist_session \\\n",
    "    WHERE session_id = 338 and item_session = 4\")\n",
    "except Exception as e:\n",
    "    print(e)\n",
    "    \n",
    "for row in results_1:\n",
    "    print(row.artist, row.song, row.length)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Second 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"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [],
   "source": [
    "# CREATE STATEMENT for second table based on second query.\n",
    "\n",
    "query_2 = \"CREATE TABLE IF NOT EXISTS songs_user_session\"\n",
    "query_2 = query_2 + \"(user_id int, session_id int, item_session int, first_name text, last_name text, artist text, song text, PRIMARY KEY((user_id, session_id), item_session))\"\n",
    "\n",
    "'''\n",
    "\n",
    "The Analytic Team needs the data order by iteminsession, that means that our clustering key needs to be itemInSession.\n",
    "As well, if we looks the second query, we need to filter for user_id and session_id, so we're going to make both variables as partition key in that particular order.\n",
    "The entire primary key statement is ((user_id, session_id), item_session)\n",
    "With that, we are going to succesfully achieve the results that the Analytic Team needs.\n",
    "\n",
    "'''\n",
    " \n",
    "\n",
    "try:\n",
    "    session.execute(query_2)\n",
    "except Exception as e:\n",
    "    print(e)         "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### INSERT code for table `songs_user_session`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "# query\n",
    "query = \"INSERT INTO songs_user_session (user_id, session_id, item_session, first_name, last_name, artist, song)\"\n",
    "query = query + \"VALUES (%s, %s, %s, %s, %s, %s, %s)\"\n",
    "\n",
    "# values and dtypes\n",
    "\n",
    "values = [10,8,3,1,4,0,9]\n",
    "dtypes = [int, int, int, str, str, str, str]\n",
    "\n",
    "insert_data(query, session, values, dtypes, file)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "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": [
    "# SELECT for second query\n",
    "## Query 2: Give me only the following: name of artist, song (sorted by itemInSession) and user (first and last name)\\\n",
    "## for userid = 10, sessionid = 182\n",
    "\n",
    "try:\n",
    "    results_2 = session.execute(\"SELECT artist, song, first_name, last_name \\\n",
    "    FROM songs_user_session WHERE user_id = 10 and session_id = 182\")\n",
    "except Exception as e:\n",
    "    print(e)\n",
    "    \n",
    "for row in results_2:\n",
    "    print(row.artist, row.song, row.first_name, row.last_name)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Third 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'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "# CREATE STATEMENT for third table based on third query.\n",
    "query_3 = \"CREATE TABLE usernames_library\"\n",
    "query_3 = query_3 + \"(song text, user_id int, first_name text, last_name text, PRIMARY KEY(song, user_id))\"\n",
    "\n",
    "# We are going to use SONG and USER_ID as Primary Key and Clustering Key, this will allow the database to update the values\n",
    "## which may be ok based on the statement of the query, because the team is asking about a list of users (name and last name)\n",
    "### and not about how many times that song was listened.\n",
    "\n",
    "try:\n",
    "    session.execute(query_3)\n",
    "except Exception as e:\n",
    "    print(e)        "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### INSERT code for table `usernames_library`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "# query\n",
    "query = \"INSERT INTO usernames_library (song, user_id, first_name, last_name)\"\n",
    "query = query + \"VALUES (%s, %s, %s, %s)\"\n",
    "# values and dtypes\n",
    "values = [9,10,1,4]\n",
    "dtypes = [str, int, str, str]\n",
    "\n",
    "# insert into table `usernames_library`\n",
    "insert_data(query, session, values, dtypes, file)\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Jacqueline Lynch\n",
      "Tegan Levine\n",
      "Sara Johnson\n"
     ]
    }
   ],
   "source": [
    "# SELECT for third query\n",
    "## Query 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",
    "try:\n",
    "    results_3 = session.execute(\"SELECT first_name, last_name FROM usernames_library \\\n",
    "    WHERE song = 'All Hands Against His Own'\")\n",
    "except Exception as e:\n",
    "    print(e)\n",
    "    \n",
    "for row in results_3:\n",
    "    print(row.first_name, row.last_name)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Drop the tables before closing out the sessions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Table music_playlist_session successfully dropped\n",
      "Table songs_user_session successfully dropped\n",
      "Table usernames_library successfully dropped\n"
     ]
    }
   ],
   "source": [
    "## Drop table statements\n",
    "table_names = ['music_playlist_session', 'songs_user_session', 'usernames_library']\n",
    "\n",
    "for table_name in table_names:\n",
    "    try:\n",
    "        drop_query = 'DROP TABLE IF EXISTS {}'.format(table_name)\n",
    "        session.execute(drop_query)\n",
    "        print(\"Table {} successfully dropped\".format(table_name))\n",
    "    except Exception as e:\n",
    "        print(e)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Close the session and cluster connection¶"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "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.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}