<jupyter_start><jupyter_text>Part I. ETL Pipeline for Pre-Processing the Files  Import Python packages<jupyter_code># Import Python packages 
import pandas as pd
from tabulate import tabulate
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>print(os.getcwd())
filepath = os.getcwd() + '/event_data'
for root, dirs, files in os.walk(filepath):
    file_path_list = glob.glob(os.path.join(root,'*'))
    # print(file_path_list)<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>full_data_rows_list = [] 

for f in file_path_list:

    with open(f, 'r', encoding = 'utf8', newline='') as csvfile: 
        csvreader = csv.reader(csvfile) 
        next(csvreader)    
        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>267<jupyter_text>Part II. Complete the Apache Cassandra coding portion of your project.  Now you are ready to work with the CSV file titled event_datafile_new.csv, 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:  Begin Apache Cassandra code.  Creating a Cluster<jupyter_code>from cassandra.cluster import Cluster
try: 
    cluster = Cluster(['127.0.0.1'])
    session = cluster.connect()
except Exception as e:
    print(e)<jupyter_output><empty_output><jupyter_text>Create Keyspace<jupyter_code>try:
    session.execute("""
    CREATE KEYSPACE IF NOT EXISTS eventdata 
    WITH REPLICATION = 
    { 'class' : 'SimpleStrategy', 'replication_factor' : 1 }""")

except Exception as e:
    print(e)<jupyter_output><empty_output><jupyter_text>Set Keyspace<jupyter_code>try:
    session.set_keyspace('eventdata')
except Exception as e:
    print(e)<jupyter_output><empty_output><jupyter_text>Sample Queries and Create, Insert, Retrieve Operation  1. Give me the artist, song title and song's length in the music app history that was heard during  sessionId = 172, and itemInSession  = 4`SELECT artist, title, song FROM session_event WHERE sessionId=172 and itemInSession=4`Since we use sessionId as the main identifier alongside itemInSession as the composite, Partition Key should be the sessionId follows by itemInSession as the Clustering Columns.<jupyter_code>query = "CREATE TABLE IF NOT EXISTS song_by_session_item "
query = query + "(sessionId int, itemInSession int, artist text, song text, length float, PRIMARY KEY (sessionId, itemInSession))"
try:
    session.execute(query)
except Exception as e:
    print(e)
file = 'event_datafile_new.csv'

with open(file, encoding = 'utf8') as f:
    csvreader = csv.reader(f)
    next(csvreader)
    for line in csvreader:
        #print(line)
        query = "INSERT INTO song_by_session_item (sessionId, itemInSession, artist, song, length)"
        query = query + "VALUES (%s, %s, %s, %s, %s)"
        session.execute(query, (int(line[8]), int(line[3]), line[0], line[9], float(line[5])))<jupyter_output><empty_output><jupyter_text>Do a SELECT to verify that the data have been inserted<jupyter_code>query = "select artist, song, length from song_by_session_item WHERE sessionId=172 AND itemInSession=4"
try:
    rows = session.execute(query)
except Exception as e:
    print(e)

df = []
for row in rows:
    df.append([row.artist, row.song, row.length])

column_labels = ['artist', 'song', 'length']
result = pd.DataFrame(df, columns=column_labels)
print(tabulate(result, headers='keys', tablefmt='psql'))<jupyter_output>+----+--------------+---------------------+----------+
|    | artist       | song                |   length |
|----+--------------+---------------------+----------|
|  0 | <NAME> | World Keeps Turning |   179.67 |
+----+--------------+---------------------+----------+<jupyter_text>2. Give me only the following: name of artist, song (sorted by itemInSession) and user (first and last name) for userid = 10, sessionid = 182`SELECT artist, song, firstName, lastName FROM song_by_user_session WHERE userId=10 AND sessionId=182 SORT BY itemInSession`userId is likely the main identifier as the Partition Key due to many duplicate sessionId. This way, we use sessionId as the Cluster Columns with itemInSession to sort naturally.<jupyter_code>query = "CREATE TABLE IF NOT EXISTS song_by_user_session "
query = query + "(userId int, sessionId int, itemInSession int, artist text, song text, firstName text, lastName text, PRIMARY KEY (userId, sessionId, itemInSession))"
try:
    session.execute(query)
except Exception as e:
    print(e)  
file = 'event_datafile_new.csv'

with open(file, encoding = 'utf8') as f:
    csvreader = csv.reader(f)
    next(csvreader)
    for line in csvreader:
        query = "INSERT INTO song_by_user_session (userId, sessionId, itemInSession, artist, song, firstName, lastName)"
        query = query + "VALUES (%s, %s, %s, %s, %s, %s, %s)"
        session.execute(query, (int(line[10]), int(line[8]), int(line[3]), line[0], line[9], line[1], line[4]))<jupyter_output><empty_output><jupyter_text>Do a SELECT to verify that the data have been inserted<jupyter_code>query = "select artist, song, firstName, lastName from song_by_user_session WHERE userId=10 AND sessionId=182"
try:
    rows = session.execute(query)
except Exception as e:
    print(e)

df = []
for row in rows:
    df.append([row.artist, row.song, row.firstname, row.lastname])

column_labels = ['artist', 'song', 'firstname', 'lastname']
result = pd.DataFrame(df, columns=column_labels)
print(tabulate(result, headers='keys', tablefmt='psql'))<jupyter_output>+----+-------------------+------------------------------------------------------+-------------+------------+
|    | artist            | song                                                 | firstname   | lastname   |
|----+-------------------+------------------------------------------------------+-------------+------------|
|  0 | Down To The Bone  | Keep On Keepin' On                                   | Sylvie      | Cruz       |
|  1 | Three Drives      | Greece 2000                                          | Sylvie      | Cruz       |
|  2 | <NAME> | Kilometer                                            | Sylvie      | Cruz       |
|  3 | <NAME>     | Catch You Baby (<NAME> & Max Sanna Radio Edit) | Sylvie      | Cruz       |
+----+-------------------+------------------------------------------------------+-------------+------------+<jupyter_text>3. Give me every user name (first and last) in my music app history who listened to the song 'Come As You Are'`SELECT firstName, lastName FROM user_by_song WHERE song='Come As You Are'`We only use song as the Partition Key since we only need to filter the song. userId is set as the Cluster Columns for identifying the name efficiently.<jupyter_code>query = "CREATE TABLE IF NOT EXISTS user_by_song "
query = query + "(song text, userId int, firstName text, lastName text, PRIMARY KEY (song, userId))"
try:
    session.execute(query)
except Exception as e:
    print(e)
file = 'event_datafile_new.csv'

with open(file, encoding = 'utf8') as f:
    csvreader = csv.reader(f)
    next(csvreader)
    for line in csvreader:
        query = "INSERT INTO user_by_song (song, userId, firstName, lastName)"
        query = query + "VALUES (%s, %s, %s, %s)"
        session.execute(query, (line[9], int(line[10]), line[1], line[4]))<jupyter_output><empty_output><jupyter_text>Do a SELECT to verify that the data have been inserted<jupyter_code>query = "select firstName, lastName from user_by_song WHERE song='Come As You Are'"
try:
    rows = session.execute(query)
except Exception as e:
    print(e)

df = []
for row in rows:
    df.append([row.firstname, row.lastname])

column_labels = ['firstname', 'lastname']
result = pd.DataFrame(df, columns=column_labels)
print(tabulate(result, headers='keys', tablefmt='psql'))<jupyter_output>+----+-------------+------------+
|    | firstname   | lastname   |
|----+-------------+------------|
|  0 | Lily        | Koch       |
+----+-------------+------------+<jupyter_text>Drop the tables before closing out the sessions<jupyter_code>query = "drop table song_by_session_item"
try:
    rows = session.execute(query)
except Exception as e:
    print(e)
query = "drop table song_by_user_session"
try:
    rows = session.execute(query)
except Exception as e:
    print(e)
query = "drop table user_by_song"
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.159536990437846, 'fraction_numerical': 0.009562154001006542, 'mean_word_length': 3.1434528773978316, 'pattern_counts': {'":': 0, '<': 56, '<?xml version=': 0, '>': 56, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 1, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '14', 'dir': 'jupyter-structured-clean-dedup', 'id': '417422', 'license': 'MIT', 'n_tokens_mistral': 3078, 'n_tokens_neox': 2953, 'n_words': 943, 'path': 'CassandraModelling/Project_1B_ Project_Template.ipynb', 'repo_name': 'nigelfranciscus/DEND', 'size': '9965'}