<jupyter_start><jupyter_text>Data modeling with Apache Cassandra  Part I. ETL Pipeline for pre-processing files  Setup<jupyter_code>from pathlib import Path
import pandas as pd
import cassandra
from helpers import *
from queries import *
%load_ext autoreload
%autoreload 2<jupyter_output><empty_output><jupyter_text>Create a list of filepaths to process original event data csv files<jupyter_code>root_path = Path.cwd()
data_path = root_path.joinpath("event_data")
file_path_list = [e for e in data_path.rglob("*.csv")]
print(f"Found {len(file_path_list)} CSV files in: {data_path}")<jupyter_output>Found 30 CSV files in: /home/workspace/event_data<jupyter_text>Process individual files to create a single file that will be used for Apache Casssandra tables Let's take a look at one individual csv file first<jupyter_code>first_file = file_path_list[0]
!head $first_file
first_df = pd.read_csv(first_file)
first_df.shape
first_df.head()<jupyter_output><empty_output><jupyter_text>Load all csv files into one dataframe and save it as CSV<jupyter_code>try:
    df = pd.read_csv(f"{root_path}/event_datafile_new.csv")
    print("Loaded file from disk")
except FileNotFoundError:
    columns = ["artist", "firstName", "gender", "itemInSession", "lastName", "length", "level", "location", "sessionId", "song", "userId"]
    df = load_all_records(file_path_list, columns)
finally:
    print(f"Shape: {df.shape}")
df.head()
df.dtypes<jupyter_output><empty_output><jupyter_text>Set correct dtypes for ts (timestamp) and userId (int)<jupyter_code># Although userId should be of type int, because of a limitation of pandas < 0.24 (Series of type int can not hold NaNs) we leave it as int
# df["userId"] = df["userId"].fillna(0).astype("int64")
df["ts"] = df["ts"].astype("datetime64[ms]")
df.head()
if not root_path.joinpath("event_datafile_new.csv").exists():
    df.to_csv(f"{root_path}/event_datafile_new.csv", index=False)
else:
    print("File event_datafile_new.csv does already exist")<jupyter_output>File event_datafile_new.csv does already exist<jupyter_text>Part II. Data Modelling with Apache CassandraNow we are ready to work with the CSV file titled **event_datafile_new.csv**, located within the W\workspace directory.  The event_datafile_new.csv contains the following columns: - artist - firstName of user- gender of user- item number in session- last name of user- length of the song- level (paid or free song)- location of the user- sessionId- song title- userIdThe image below is a screenshot of what the denormalized data should appear like in the **event_datafile_new.csv** after the code above is run:  Create Cluster<jupyter_code>from cassandra.cluster import Cluster

cluster = Cluster()
session = cluster.connect()<jupyter_output><empty_output><jupyter_text>Create Keyspace<jupyter_code># http://cassandra.apache.org/doc/latest/cql/ddl.html#create-keyspace
session.execute(cluster_create_keyspace)<jupyter_output><empty_output><jupyter_text>Set Keyspace<jupyter_code>session.set_keyspace("sparkify")<jupyter_output><empty_output><jupyter_text>Create queries to find answers to the following three questions1. Give me the artist, song title and song's length in the music app history that was heard during  sessionId = 338, and itemInSession  = 42. Give me only the following: name of artist, song (sorted by itemInSession) and user (first and last name) for userid = 10, sessionid = 1823. Give me every user name (first and last) in my music app history who listened to the song 'All Hands Against His Own'  Query 1<jupyter_code>session.execute(q1_drop_table)
session.execute(q1_create_table)<jupyter_output><empty_output><jupyter_text>Insert data<jupyter_code>table = "song_playlist_session"
cols = ["sessionId", "itemInSession", "artist", "song", "length"]
q1_cql = f"INSERT INTO {table} ({cols[0]}, {cols[1]}, {cols[2]}, {cols[3]}, {cols[4]}) VALUES (?, ?, ?, ?, ?)"
print(q1_cql)
batch_insert(cql=q1_cql, cols=cols, data=df, size=500, session=session)<jupyter_output>Starting batch insert for 8056 rows in 17 batches of size 500
Inserted 474 rows of data
Inserted 474 rows of data
Inserted 474 rows of data
Inserted 474 rows of data
Inserted 474 rows of data
Inserted 474 rows of data
Inserted 474 rows of data
Inserted 474 rows of data
Inserted 474 rows of data
Inserted 474 rows of data
Inserted 474 rows of data
Inserted 474 rows of data
Inserted 474 rows of data
Inserted 474 rows of data
Inserted 474 rows of data
Inserted 473 rows of data
Inserted 473 rows of data
Batch insert finished<jupyter_text>Perform sanity check<jupyter_code>query(f"SELECT * FROM {table} LIMIT 5", session, print_result=False)<jupyter_output><empty_output><jupyter_text>Execute query_Give me the artist, song title and song's length in the music app history that was heard during sessionId = 338, and itemInSession = 4_<jupyter_code>q1 = query(q1_query, session)<jupyter_output>[Row(artist='Faithless', song='Music Matters (<NAME>)', length=495.30731201171875)]<jupyter_text>Query 2<jupyter_code>session.execute(q2_drop_table)
session.execute(q2_create_table)<jupyter_output><empty_output><jupyter_text>Insert data<jupyter_code>table = "song_playlist_user"
cols = ["userId", "sessionId", "itemInSession", "artist", "song", "firstName", "lastName"]
q2_cql = f"INSERT INTO {table} ({cols[0]}, {cols[1]}, {cols[2]}, {cols[3]}, {cols[4]}, {cols[5]}, {cols[6]}) VALUES (?, ?, ?, ?, ?, ?, ?)"
print(q2_cql)
batch_insert(cql=q2_cql, cols=cols, data=df, size=250, session=session)<jupyter_output>Starting batch insert for 8056 rows in 33 batches of size 250
Inserted 245 rows of data
Inserted 245 rows of data
Inserted 245 rows of data
Inserted 245 rows of data
Inserted 244 rows of data
Inserted 244 rows of data
Inserted 244 rows of data
Inserted 244 rows of data
Inserted 244 rows of data
Inserted 244 rows of data
Inserted 244 rows of data
Inserted 244 rows of data
Inserted 244 rows of data
Inserted 244 rows of data
Inserted 244 rows of data
Inserted 244 rows of data
Inserted 244 rows of data
Inserted 244 rows of data
Inserted 244 rows of data
Inserted 244 rows of data
Inserted 244 rows of data
Inserted 244 rows of data
Inserted 244 rows of data
Inserted 244 rows of data
Inserted 244 rows of data
Inserted 244 rows of data
Inserted 244 rows of data
Inserted 244 rows of data
Inserted 244 rows of data
Inserted 244 rows of data
Inserted 244 rows of data
Inserted 244 rows of data
Inserted 244 rows of data
Batch insert finished<jupyter_text>Perform sanity check<jupyter_code>query(f"SELECT * FROM {table} LIMIT 5", session, print_result=False)<jupyter_output><empty_output><jupyter_text>Execute query_Give me only the following: name of artist, song (sorted by itemInSession) and user (first and last name) for userid = 10, sessionid = 182_<jupyter_code># explicitly select itemInSession to show that sorting works
q2 = query(q2_query, session)<jupyter_output>[Row(artist='<NAME>', song='Catch You Baby (Steve Pitron & Max Sanna Radio Edit)', firstname='Sylvie', lastname='Cruz', iteminsession=3),
 Row(artist='<NAME>', song='Kilometer', firstname='Sylvie', lastname='Cruz', iteminsession=2),
 Row(artist='Three Drives', song='Greece 2000', firstname='Sylvie', lastname='Cruz', iteminsession=1),
 Row(artist='Down To The Bone', song="Keep On Keepin' On", firstname='Sylvie', lastname='Cruz', iteminsession=0)]<jupyter_text>Query 3<jupyter_code>session.execute(q3_drop_table)
session.execute(q3_create_table)<jupyter_output><empty_output><jupyter_text>Insert data<jupyter_code>table = "song_user_name"
cols = ["song", "userId", "firstName", "lastName"]
q3_cql = f"INSERT INTO {table} ({cols[0]}, {cols[1]}, {cols[2]}, {cols[3]}) VALUES (?, ?, ?, ?)"
print(q3_cql)
batch_insert(cql=q3_cql, cols=cols, data=df, size=500, session=session)<jupyter_output>Starting batch insert for 8056 rows in 17 batches of size 500
Inserted 474 rows of data
Inserted 474 rows of data
Inserted 474 rows of data
Inserted 474 rows of data
Inserted 474 rows of data
Inserted 474 rows of data
Inserted 474 rows of data
Inserted 474 rows of data
Inserted 474 rows of data
Inserted 474 rows of data
Inserted 474 rows of data
Inserted 474 rows of data
Inserted 474 rows of data
Inserted 474 rows of data
Inserted 474 rows of data
Inserted 473 rows of data
Inserted 473 rows of data
Batch insert finished<jupyter_text>Perform sanity check<jupyter_code>query(f"SELECT * FROM {table} LIMIT 5", session, print_result=False)<jupyter_output><empty_output><jupyter_text>Execute query_Give me every user name (first and last) in my music app history who listened to the song 'All Hands Against His Own'_<jupyter_code>q3 = query(q3_query, session)<jupyter_output>[Row(firstname='Jacqueline', lastname='Lynch'),
 Row(firstname='Tegan', lastname='Levine'),
 Row(firstname='Sara', lastname='Johnson')]<jupyter_text>Drop tables before closing out session<jupyter_code>[session.execute(q) for q in [q1_drop_table, q2_drop_table, q3_drop_table]]<jupyter_output><empty_output><jupyter_text>Close session and cluster connection¶<jupyter_code>session.shutdown()
cluster.shutdown()<jupyter_output><empty_output>
StarcoderdataJupyterStructuredCleanDedup