data_modeling_with_apache_cassandra.ipynb

{
 "cells": [
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Project: Data Modeling with Apache Cassandra\n",
    "A startup called Sparkify wants to analyze the data they've been collecting on songs and user activity on their new music streaming app. The analysis team is particularly interested in understanding what songs users are listening to. Currently, there is no easy way to query the data to generate the results, since the data reside in a directory of CSV files on user activity on the app.\n",
    "\n",
    "They'd like a data engineer to create an Apache Cassandra database which can create queries on song play data to answer the questions. You'll be able to test your database by running queries given to you by the analytics team from Sparkify to create the results."
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Part I. ETL Pipeline for pre-processing the CSV files"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#pip install cassandra-driver"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Import Python packages "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "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",
    "\n",
    "# set Jupyter setting to display all float values with 2 decimals\n",
    "pd.set_option('display.float_format', lambda x: '%.2f' % x)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Creating list of filepaths to process csv data files"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "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)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Merging all of the csv files for the Apache Cassandra tables"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "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], \\\n",
    "                         row[7], row[8], row[12], row[13], row[16]))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "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))"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Exploratory Data Analysis"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create dataframe from event_datafile_new.csv\n",
    "df = pd.read_csv('event_datafile_new.csv')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# show sample of data\n",
    "df.sample(n=3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# show any null values and data types\n",
    "print(df.info())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# show summary statistics for both numeric\n",
    "print(df.describe(include='number'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# show summary statistics for object columns\n",
    "print(df.describe(include='object'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# show unique values for each column\n",
    "for col in df.columns:\n",
    "    print(col, df[col].nunique())"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Part II. Create Apache Cassandra databases\n",
    "\n",
    "The csv file titled <font color=red>event_datafile_new.csv</font> should be located within the 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 an example of how the denormalized data should appear in the <font color=red>**event_datafile_new.csv**</font> after the CSV files are merged and processed:<br>\n",
    "\n",
    "<img src=\"image_event_datafile_new.jpg\">"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Creating a Cluster"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create a cluster\n",
    "from cassandra.cluster import Cluster\n",
    "# establish connection and create a session\n",
    "try: \n",
    "    cluster = Cluster(['127.0.0.1']) #If you have a locally installed Apache Cassandra instance\n",
    "    session = cluster.connect()\n",
    "except Exception as e:\n",
    "    print(e)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Create Keyspace"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create keyspace\n",
    "try:\n",
    "    session.execute(\"\"\"\n",
    "    CREATE KEYSPACE IF NOT EXISTS project \n",
    "    WITH REPLICATION = \n",
    "    { 'class' : 'SimpleStrategy', 'replication_factor' : 1 }\"\"\"\n",
    ")\n",
    "\n",
    "except Exception as e:\n",
    "    print(e)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Set Keyspace"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# connrect to keyspace\n",
    "try:\n",
    "    session.set_keyspace('project')\n",
    "except Exception as e:\n",
    "    print(e)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create queries to ask the following three questions of the data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# set variable name for csv file to be inserted into tables\n",
    "file = 'event_datafile_new.csv'"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### Query 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",
    "`SELECT artist, song, length FROM session_library WHERE sessionid = 338 AND itemInSession = 4;`\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create the table for query 1\n",
    "query = \"CREATE TABLE IF NOT EXISTS session_library \"\n",
    "query = query + \"(sessionId int, itemInSession int, \\\n",
    "                artist text, song text, length float, \\\n",
    "                PRIMARY KEY (sessionId, itemInSession))\";\n",
    "try:\n",
    "    session.execute(query)\n",
    "except Exception as e:\n",
    "    print(e)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": false
   },
   "outputs": [],
   "source": [
    "with open(file, encoding='utf8') as f:\n",
    "    # use csv.DictReader to read the CSV file as a dictionary\n",
    "    csvreader = csv.DictReader(f)\n",
    "    \n",
    "    for row in csvreader:\n",
    "        # assign the INSERT statements into the `query` variable\n",
    "        query = \"\"\"INSERT INTO session_library (sessionId, itemInSession, artist, song, length)\n",
    "        VALUES (%s, %s, %s, %s, %s)\"\"\"\n",
    "        # assign which column element should be assigned for each column in the INSERT statement\n",
    "        session.execute(query, (int(row['sessionId']), int(row['itemInSession']),\n",
    "                                row['artist'], row['song'], float(row['length'])))"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Rationale: \n",
    "- *sessionid* is used as the partition key. This will help evenely distribute data across nodes and the requested query filters data based on the sessionid.\n",
    "- *itemInSession* is chosen as the clustering column because the requested query filters on this column. Additionally, using it in the primary key helps to avoid the ALLOW FILTERING clause."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "## add in the SELECT statement to verify the data was entered into the table\n",
    "query = \"select artist, song, length from session_library WHERE sessionID = 338 AND itemInSession = 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, row.song, row.length)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### Query 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",
    "`SELECT artist, song, firstName, lastName FROM user_session_library WHERE userId = 10 AND sessionId = 182 ORDER BY itemInSession;`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create table for query 2\n",
    "query = \"CREATE TABLE IF NOT EXISTS user_session_library \"\n",
    "query = query + \"(userId int, sessionId int, itemInSession int, artist text, \\\n",
    "        song text, firstName text, lastName text, \\\n",
    "        PRIMARY KEY ((userId, sessionId), itemInSession))\";\n",
    "try:\n",
    "    session.execute(query)\n",
    "except Exception as e:\n",
    "    print(e)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": false
   },
   "outputs": [],
   "source": [
    "with open(file, encoding='utf8') as f:\n",
    "    csvreader = csv.DictReader(f)\n",
    "\n",
    "    for row in csvreader:\n",
    "        # assign the INSERT statements into the `query` variable\n",
    "        query = \"\"\"INSERT INTO user_session_library (userId, sessionId, itemInSession,\n",
    "                                          artist, song, firstName, lastName)\n",
    "                   VALUES (%s, %s, %s, %s, %s, %s, %s)\"\"\"\n",
    "        # assign which column element should be assigned for each column in the INSERT statement\n",
    "        session.execute(query, (int(row['userId']), int(row['sessionId']), \\\n",
    "                                int(row['itemInSession']), row['artist'], \\\n",
    "                                row['song'], row['firstName'], \\\n",
    "                                row['lastName']))"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Rationale: \n",
    "- A compound partition key consisting of *userId* and *sessionId* is used to distribute the data across nodes. The query filters data based on both userId and sessionId, making them suitable candidates for the partition key.\n",
    "- *itemInSession* is chosen as the clustering column because the requested query sorts on this column. Additionally, this allows Cassandra store data within each partition in the desired order, making for a more efficient query."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "## add in the SELECT statement to verify the data was entered into the table\n",
    "query = \"select artist, song, firstName, lastName from user_session_library WHERE userId = 10 AND sessionId = 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, row.song, row.firstname, row.lastname)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### 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",
    "\n",
    "`SELECT firstName, lastName FROM song_listeners WHERE song = 'All Hands Against His Own';`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create table for query 3\n",
    "query = \"CREATE TABLE IF NOT EXISTS song_listeners \"\n",
    "query = query + \"(song text, userId int, firstName text, lastName text, \\\n",
    "    PRIMARY KEY (song, userId))\";\n",
    "try:\n",
    "    session.execute(query)\n",
    "except Exception as e:\n",
    "    print(e)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": false
   },
   "outputs": [],
   "source": [
    "with open(file, encoding='utf8') as f:\n",
    "    csvreader = csv.DictReader(f)\n",
    "\n",
    "    for row in csvreader:\n",
    "        # assign the INSERT statements into the `query` variable\n",
    "        query = \"\"\"INSERT INTO song_listeners (song, userId, firstName, lastName)\n",
    "                   VALUES (%s, %s, %s, %s)\"\"\"\n",
    "        # assign which column element should be assigned for each column in the INSERT statement\n",
    "        session.execute(query, (row['song'], int(row['userId']), \\\n",
    "                                row['firstName'], row['lastName']))\n"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Rationale: \n",
    "- *song* is used as the partition key, as the query filters data based on the song's title. Using song as the partition key allows Cassandra to efficiently locate the partition containing the data for a specific song.\n",
    "- *userId* is chosen as the clustering column to ensure that each row within a partition is unique. Since Sparkify is interested in retrieving the firstName and lastName of users who listened to a specific song, we need to consider that the same user might listen to the same song multiple times. Using *userId* as the clustering column helps avoid overwriting data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# add the SELECT statement to verify the data was entered into the table\n",
    "query = \"SELECT firstName, lastName FROM song_listeners WHERE song = 'All Hands Against His Own';\"\n",
    "try:\n",
    "    rows = session.execute(query)\n",
    "except Exception as e:\n",
    "    print(e)\n",
    "\n",
    "for row in rows:\n",
    "    print(row.firstname, row.lastname)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Drop the tables before closing out the sessions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## drop the table before closing out the sessions\n",
    "table_names = ['session_library', 'user_session_library', 'song_listeners']\n",
    "\n",
    "for table_name in table_names:\n",
    "    query = f\"DROP TABLE {table_name}\"\n",
    "    try:\n",
    "        session.execute(query)\n",
    "        print(f\"Table {table_name} dropped.\")\n",
    "    except Exception as e:\n",
    "        print(f\"Error dropping table {table_name}: {e}\")\n"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Close the session and cluster connection¶"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "session.shutdown()\n",
    "cluster.shutdown()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.10.9"
  },
  "vscode": {
   "interpreter": {
    "hash": "6ed82f31c4f11492b8cc0aa5134042239548daa150e944f93ef890f5c5bb4ab2"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}