<jupyter_start><jupyter_text>Part I. ETL Pipeline for Pre-Processing the Files  PLEASE RUN THE FOLLOWING CODE FOR PRE-PROCESSING THE FILES  Import Python packages<jupyter_code># Import Python packages 
import pandas as pd
import cassandra
import re
import os
import glob
import numpy as np
import json
import csv<jupyter_output><empty_output><jupyter_text>Creating list of filepaths to process original event csv data files<jupyter_code># checking your current working directory
print(os.getcwd())

# Get your current folder and subfolder event data
filepath = os.getcwd() + '/event_data'

# Create a for loop to create a list of files and collect each filepath
for root, dirs, files in os.walk(filepath):
    
# join the file path and roots with the subdirectories using glob
    file_path_list = glob.glob(os.path.join(root,'*'))<jupyter_output>/home/workspace<jupyter_text>Processing the files to create the data file csv that will be used for Apache Casssandra tables<jupyter_code># initiating an empty list of rows that will be generated from each file
full_data_rows_list = [] 
    
# for every filepath in the file path list 
for f in file_path_list:

# reading csv file 
    with open(f, 'r', encoding = 'utf8', newline='') as csvfile: 
        # creating a csv reader object 
        csvreader = csv.reader(csvfile) 
        next(csvreader)
        
 # extracting each data row one by one and append it        
        for line in csvreader:
            full_data_rows_list.append(line) 
            
# creating a smaller event data csv file called event_datafile_full csv that will be used to insert data into the \
# Apache Cassandra tables
csv.register_dialect('myDialect', quoting=csv.QUOTE_ALL, skipinitialspace=True)

with open('event_datafile_new.csv', 'w', encoding = 'utf8', newline='') as f:
    writer = csv.writer(f, dialect='myDialect')
    writer.writerow(['artist','firstName','gender','itemInSession','lastName','length',\
                'level','location','sessionId','song','userId'])
    for row in full_data_rows_list:
        if (row[0] == ''):
            continue
        writer.writerow((row[0], row[2], row[3], row[4], row[5], row[6], row[7], row[8], row[12], row[13], row[16]))

# check the number of rows in your csv file
with open('event_datafile_new.csv', 'r', encoding = 'utf8') as f:
    print(sum(1 for line in f))<jupyter_output>6821<jupyter_text>Part II. Apache Cassandra code.  Our CSV file titled event_datafile_new.csv is located within the 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:  Creating a Cluster<jupyter_code># This should make a connection to a Cassandra instance your local machine 
# (127.0.0.1)

from cassandra.cluster import Cluster
cluster = Cluster(['127.0.0.1'])

# To establish connection and begin executing queries, need a session
session = cluster.connect()<jupyter_output><empty_output><jupyter_text>Create Keyspace<jupyter_code># Creating a Keyspace named template
try:
    session.execute("""
    CREATE KEYSPACE IF NOT EXISTS template 
    WITH REPLICATION = 
    { 'class' : 'SimpleStrategy', 'replication_factor' : 1 }"""
)

except Exception as e:
    print(e)<jupyter_output><empty_output><jupyter_text>Set Keyspace<jupyter_code># Set KEYSPACE to the keyspace specified above
try:
    session.set_keyspace('template')
except Exception as e:
    print(e)<jupyter_output><empty_output><jupyter_text>CREATING OUR SPARKIFY QUERIES  Queries to ask the following three questions of the data 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 2. Give me only the following: name of artist, song (sorted by itemInSession) and user (first and last name) for userid = 10, sessionid = 182     3. Give me every user name (first and last) in my music app history who listened to the song 'All Hands Against His Own'  Processing our event CSV file for our all of our queries<jupyter_code>## A function to process our INSERT STATEMENTS to any table we choose that has been previously created in Apacha Cassandra 
## within the session and keyspace initiated above.

def insert_with_csv(file, table, query_columns, index_columns, ints, floats):
    """
    It take a file as input plus some query arguments to execute insert statements to an Apache Cassandra table specified.
    
    file : csv file to read per row.
    table : Apache Cassandra table created in keyspace.
    query_columns : columns of table in order that wil be used to insert values.
    index_columns : index of columns in csv file header that will be used to insert values for each row read in the csv read.
    ints: index of values in the insert values that need to be converted to type int.
    floats: index of values in the insert values that need to be converted to type float.
    
    """
    ## Code to open file and read csv
    with open(file, encoding = 'utf8') as f:
        csvreader = csv.reader(f)
        next(csvreader) # skip header
        ## creates a string with the number of placeholders based on the number of columns in the table
        values = "%s, " * (len(index_columns) - 1) + "%s" 
        for line in csvreader:
            ## Assigns the INSERT statements into the `query` variable
            query = "INSERT INTO " + table + " " + query_columns
            query = query + " VALUES " + "(" + values + ")"
            ## Assigns which column element should be assigned for each column in the INSERT statement.
            ## For e.g., to INSERT artist_name and user first_name, you would change the index_columns argumetn to [0,1]
            row = [line[i] for i in index_columns]
            ## Uses the ints argument to cast to int the values in row that needed to be inserted as an int type
            row = [row[i] if i not in ints else int(row[i]) for i in range(len(row))]
            ## Uses the floats argument to cast to float the values in row needed to be inserted as a float type
            row = [row[i] if i not in floats else float(row[i]) for i in range(len(row))]
            ## Executes the query
            session.execute(query, row)
## 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
query = "CREATE TABLE IF NOT EXISTS songs_per_session"
## PRIMARY KEY is set below to focus our query on each session and its children sessionId.
query = query + "(artist varchar, song varchar, length float, sessionId int, itemInSession int, PRIMARY KEY (sessionId, itemInSession))"
try:
    session.execute(query)
except Exception as e:
    print(e)
## To use insert_with_csv function we define the required arguments that need to be specified 
## to do insert statements to our Apache Cassandra table.

## The file that will be read and each row used to populate the table defined next
file = 'event_datafile_new.csv'
## Apache Cassandra table created
table = 'songs_per_session'
## The columns that will be used to form the INSERT INTO query
query_columns = '(artist, song, length, sessionId, itemInSession)'
## The index_columns will be used to obtain only the columns that are needed to form the query 
index_columns = [0, 9, 5, 8, 3]
## With 'ints' and 'floats' the function knows which values to convert to int/float before submitting the query 
ints = [3, 4] 
floats = [2]

## We call insert_with_csv that doesn't return an object but execute INSERT INTO queries into the "songs_per_session" table.
insert_with_csv(file, table, query_columns, index_columns, ints, floats)<jupyter_output><empty_output><jupyter_text>Do a SELECT to verify that the data have been inserted into each table<jupyter_code>## States and executes the 1st query. 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

query = "SELECT artist, song, length FROM songs_per_session WHERE sessionId = 338 AND itemInSession =4"
try:
    rows = session.execute(query)
except Exception as e:
    print(e)

## Uses a dataframe to save the query results and display it with column names.
df = pd.DataFrame([row for row in rows], columns = ['artist', 'song', 'length'])
print(df)<jupyter_output>artist                             song      length
0  Faithless  Music Matters (<NAME>)  495.307312<jupyter_text>New queries can be processed with a CREATE statement and our processing function to speed up results.<jupyter_code>## 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

query = "CREATE TABLE IF NOT EXISTS songs_per_user_per_session"
## PRIMARY KEY is set below in order to sort our results with our first clustering column itemInSession.Hence the need for a composite primary key.
query = query + "(artist varchar, song varchar, first_name text, last_name text, user_id int, sessionId int, itemInSession int, PRIMARY KEY ((user_id, sessionId), itemInSession))"
try:
    session.execute(query)
except Exception as e:
    print(e)   

## Prepares arguments that will be used in insert_with_csv function to populate 'songs_user_session' table.
table = 'songs_per_user_per_session'
query_columns = '(artist, song, first_name, last_name, user_id, sessionId, itemInSession)'
index_columns = [0, 9, 1, 4, 10, 8, 3]
ints = [4, 5, 6]
floats = []    

## We call insert_with_csv to do INSERT INTO queries from each row on the csv file to the table 'songs_per_user_per_session'.
insert_with_csv(file, table, query_columns, index_columns, ints, floats)

## Continues stating the SELECT statement for our query 2.
query = "SELECT artist, song, first_name, last_name FROM songs_per_user_per_session WHERE sessionId = 182 AND user_id =10"
try:
    rows = session.execute(query)
except Exception as e:
    print(e)
df = pd.DataFrame([row for row in rows], columns = ['artist', 'song', 'first_name', 'last_name'])
print(df)
## 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'

query = "CREATE TABLE IF NOT EXISTS songs_per_user"
## PRIMARY KEY makes sure a song can exist for two or more users. A song name can be repeated this is why the artist column is not needed in the PK \
## since the query only focuses on the song name.
query = query + "(first_name text, last_name text, song varchar, user_id int, PRIMARY KEY (song, user_id))"
try:
    session.execute(query)
except Exception as e:
    print(e)                     

table = 'songs_per_user'
query_columns = '(first_name, last_name, song, user_id)'
index_columns = [1, 4, 9, 10]
ints = [3]
floats = []  

insert_with_csv(file, table, query_columns, index_columns, ints, floats)
## SELECT statement for the last query to 'songs_per_user'. 
query = "SELECT first_name, last_name FROM songs_per_user WHERE song = 'All Hands Against His Own'"
try:
    rows = session.execute(query)
except Exception as e:
    print(e)
df = pd.DataFrame([row for row in rows], columns = ['first', 'last'])
print(df)<jupyter_output>first     last
0  Jacqueline    Lynch
1       Tegan   Levine
2        Sara  Johnson<jupyter_text>Drop the tables before closing out the sessions<jupyter_code>## Drop tables before closing out the sessions
query = "DROP TABLE IF EXISTS songs_per_session"
try:
    rows = session.execute(query)
except Exception as e:
    print(e)

query = "DROP TABLE IF EXISTS songs_per_user_per_session"
try:
    rows = session.execute(query)
except Exception as e:
    print(e)

query = "DROP TABLE IF EXISTS songs_per_user"
try:
    rows = session.execute(query)
except Exception as e:
    print(e)<jupyter_output><empty_output><jupyter_text>Close the session and cluster connection¶<jupyter_code>session.shutdown()
cluster.shutdown()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.08103674540682415, 'fraction_numerical': 0.009678477690288715, 'mean_word_length': 4.011508425811755, 'pattern_counts': {'":': 0, '<': 42, '<?xml version=': 0, '>': 42, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '11', 'dir': 'jupyter-structured-clean-dedup', 'id': '336918', 'license': 'MIT', 'n_tokens_mistral': 3744, 'n_tokens_neox': 3526, 'n_words': 1537, 'path': 'cassandra etl/Project_1B_ Project_Template.ipynb', 'repo_name': 'Mgmaplus/DataEngineer_Nano', 'size': '12201'}