gabspereira/workspace
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
57e6e5a6fdfceed853528b9908a09dcc06377d31
workspace/data-engineering/02-cassandra-modeling/Project_1B_ Project_Template.ipynb
@gabriel.pereira 6796398924 refactor: restructure monorepo for clean portfolio layout 
- Move timesfm-forecast into apps/ directory
- Flatten Udacity portfolio projects from deep URL-encoded paths
  into data-engineering/01-XX numbered directories
- Remove old My-Data-Engineering-Portifolio/ parent directory
- Rewrite root README.md: professional overview with badges,
  project table, and repo structure diagram
- Create data-engineering/README.md with per-project descriptions
- Add README.md for 02-cassandra-modeling (was missing)
- Add README.md for 05-airflow-pipelines (was missing)
- Normalize capstone readme.md -> README.md
- Update .gitignore: add *.cfg, *.env, *.zip, *.sas7bdat,
  Jupyter checkpoints, IDE dirs; remove uv.lock exclusion
- Add dwh.cfg.example and dl.cfg.example credential templates
- Untrack real credential files (dwh.cfg, dl.cfg)

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
2026-03-26 16:48:50 -03:00

481 lines
12 KiB
Plaintext
Raw Blame History

{
"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": 1,
"metadata": {},
"outputs": [],
"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"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"#### Creating list of filepaths to process original event csv data files"
]
},
{
"cell_type": "code",
"execution_count": 2,
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"/home/workspace\n"
]
}
],
"source": [
"print(os.getcwd())\n",
"filepath = os.getcwd() + '/event_data'\n",
"for root, dirs, files in os.walk(filepath):\n",
" file_path_list = glob.glob(os.path.join(root,'*'))"
]
},
{
"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": 3,
"metadata": {},
"outputs": [],
"source": [
"full_data_rows_list = [] \n",
"for f in file_path_list:\n",
" with open(f, 'r', encoding = 'utf8', newline='') as csvfile: \n",
" csvreader = csv.reader(csvfile) \n",
" next(csvreader)\n",
" \n",
" for line in csvreader:\n",
" full_data_rows_list.append(line) \n",
" \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",
" print(sum(1 for line in f))"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"#### Creating a Cluster"
]
},
{
"cell_type": "code",
"execution_count": 5,
"metadata": {},
"outputs": [],
"source": [
"from cassandra.cluster import Cluster\n",
"cluster = Cluster()\n",
"session = cluster.connect()"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"#### Create Keyspace"
]
},
{
"cell_type": "code",
"execution_count": 6,
"metadata": {},
"outputs": [],
"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)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"#### Set Keyspace"
]
},
{
"cell_type": "code",
"execution_count": 7,
"metadata": {},
"outputs": [],
"source": [
"try:\n",
" session.set_keyspace('udacity')\n",
"except Exception as e:\n",
" print(e)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"#### Create table 'songs_by_user_session'. Partition keys: 'sessionId' and 'itemInSession'. The query will use these column as a filter."
]
},
{
"cell_type": "code",
"execution_count": 8,
"metadata": {},
"outputs": [],
"source": [
"query = \"CREATE TABLE IF NOT EXISTS songs_by_user_session\"\n",
"query = query + \"(sessionId int, itemSession int, artist_name text, song_title text, length double, PRIMARY KEY (sessionId, itemSession))\"\n",
"try:\n",
" session.execute(query)\n",
"except Exception as e:\n",
" print(e) "
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"#### Insert data into 'songs_by_user_session' table"
]
},
{
"cell_type": "code",
"execution_count": 9,
"metadata": {
"scrolled": false
},
"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 songs_by_user_session (sessionId, itemSession, artist_name, song_title, 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": {},
"source": [
"#### Verify that the data have been inserted into 'songs_by_user_session'"
]
},
{
"cell_type": "code",
"execution_count": 10,
"metadata": {
"scrolled": true
},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"Faithless Music Matters (Mark Knight Dub) 495.3073\n"
]
}
],
"source": [
"query = \"select artist_name, song_title, length from songs_by_user_session where sessionId=338 and itemSession=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.length)\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"#### Create table 'songs_by_user_and_session'. Partition keys: 'userId' and 'sessionId'. The query will use these column as a filter. Column: 'itemSession' will be used as a cluster."
]
},
{
"cell_type": "code",
"execution_count": 11,
"metadata": {},
"outputs": [],
"source": [
"query = \"CREATE TABLE IF NOT EXISTS songs_by_user_and_session\"\n",
"query = query + \"(userId int, sessionId int, artist_name text, song_title text, itemSession int, first_name text, last_name text, PRIMARY KEY ((userId, sessionId), itemSession))\"\n",
"try:\n",
" session.execute(query)\n",
"except Exception as e:\n",
" print(e) \n",
" "
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"#### Insert data into into 'songs_by_user_and_session'"
]
},
{
"cell_type": "code",
"execution_count": 12,
"metadata": {},
"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 songs_by_user_and_session (userId, sessionId, artist_name, song_title, itemSession, first_name, last_name)\"\n",
" query = query + \"VALUES (%s, %s, %s, %s, %s, %s, %s)\"\n",
" session.execute(query, (int(line[10]), int(line[8]), line[0], line[9], int(line[3]), line[1], line[4]))"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"#### Verify that the data have been inserted into 'songs_by_user_and_session'"
]
},
{
"cell_type": "code",
"execution_count": 13,
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"Error from server: code=2200 [Invalid query] message=\"Undefined column name usedid\"\n"
]
}
],
"source": [
"query = \"select artist_name, song_title, first_name, last_name from songs_by_user_and_session where usedId=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_name, row.song_title, row.first_name, row.last_name)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"#### Create table 'user_by_song'. Partition keys: 'song' and 'userId'. The 'song' column will be used as a filter, and 'userId' will compose a unique key since use IDs are already unique identifiers."
]
},
{
"cell_type": "code",
"execution_count": 14,
"metadata": {},
"outputs": [],
"source": [
"query = \"CREATE TABLE IF NOT EXISTS user_by_song\"\n",
"query = query + \"(song text, userId int, first_name text, last_name text, PRIMARY KEY (song, userId))\"\n",
"try:\n",
" session.execute(query)\n",
"except Exception as e:\n",
" print(e) "
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"#### Insert data into into 'user_by_song'"
]
},
{
"cell_type": "code",
"execution_count": 15,
"metadata": {},
"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_by_song (song, userId,first_name, last_name)\"\n",
" query = query + \"VALUES (%s, %s, %s, %s)\"\n",
" session.execute(query, (line[9], int(line[10]),line[1], line[4]))"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"#### Verify that the data have been inserted into 'user_by_song'"
]
},
{
"cell_type": "code",
"execution_count": 16,
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"Jacqueline Lynch\n",
"Tegan Levine\n",
"Sara Johnson\n"
]
}
],
"source": [
"query = \"select first_name, last_name from user_by_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",
"for row in rows:\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": 17,
"metadata": {},
"outputs": [],
"source": [
"query = \"drop table if exists songs_by_user_session\"\n",
"try:\n",
" rows = session.execute(query)\n",
"except Exception as e:\n",
" print(e)\n",
" \n",
"query = \"drop table if exists songs_by_user_and_session\"\n",
"try:\n",
" rows = session.execute(query)\n",
"except Exception as e:\n",
" print(e)\n",
" \n",
"query = \"drop table if exists user_by_song\"\n",
"try:\n",
" rows = session.execute(query)\n",
"except Exception as e:\n",
" print(e)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### Close the session and cluster connection¶"
]
},
{
"cell_type": "code",
"execution_count": 18,
"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.3"
}
},
"nbformat": 4,
"nbformat_minor": 2
}
Reference in New Issue View Git Blame Copy Permalink