<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 pandas as pd
import cassandra
import re
import os
import glob
import numpy as np
import json
import csv
def pandas_factory(colnames, rows):
    return pd.DataFrame(rows, columns=colnames)<jupyter_output><empty_output><jupyter_text>Creating list of filepaths to process original event csv data files<jupyter_code># checking current working directory
print(os.getcwd())

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,'*'))
    #print(file_path_list)<jupyter_output>/Users/lamo/Desktop/Cources/UdacityaDataEngineering/4. Non-Relational Databases/project/DataModelingWithApacheCassandra<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 f in file_path_list:

    with open(f, 'r', encoding = 'utf8', newline='') as csvfile: 
        csvreader = csv.reader(csvfile) 
        next(csvreader)
        
        for line in csvreader:
            #print(line)
            full_data_rows_list.append(line) 
            
print(len(full_data_rows_list))
#print(full_data_rows_list)

# creating a smaller event data csv file called event_datafile_full csv
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>The 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 writing your Apache Cassandra code in the cells below  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 UdacitySparkify 
    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('udacitysparkify')
except Exception as e:
    print(e)<jupyter_output><empty_output><jupyter_text>Create 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'<jupyter_code>## 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 music_sessions "
query = query + "(sessionId int, item_in_session int, artist_name text, song text, song_length float, PRIMARY KEY (sessionId, item_in_session))"
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 music_sessions (sessionId, item_in_session, artist_name, song, 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>In the first query we are interested in retrieving the artist, song and song length from music logs, we chose the (session id and item in session) as a unique primary key<jupyter_code>#QUERY 1
session.row_factory = pandas_factory

query = "SELECT artist_name, song, song_length FROM music_sessions where sessionId=338 and item_in_session=4"

q1 = session.execute(query)._current_rows 
q1

## 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 user_sessions "
query = query + "(userId int ,sessionId int, item_in_session int, artist_name text, song text,\
user_first_name text, user_last_name text, PRIMARY KEY ((userId, sessionId), item_in_session))"
try:
    session.execute(query)
except Exception as e:
    print(e)                    
with open(file, encoding = 'utf8') as f:
    csvreader = csv.reader(f)
    next(csvreader) 
    for line in csvreader:
        query = "INSERT INTO user_sessions (artist_name, song, item_in_session, user_first_name, user_last_name, userId, sessionId)"
        query = query + "VALUES (%s, %s, %s, %s,%s,%s,%s)"
        session.execute(query, (line[0], line[9], int(line[3]), line[1], line[4], int(line[10]), int(line[8])))<jupyter_output><empty_output><jupyter_text>In the second query we are interested in retrieving the artist song and user first and last name, we want our data sorted by item in session field, therefore we chose the (userid and session id) as unique primary key and (item in session) as a clustering column to sort by it<jupyter_code>#QUERY 2
session.row_factory = pandas_factory

query = "SELECT artist_name, song, user_first_name, user_last_name FROM user_sessions\
    where userId=10 and sessionId=182"

    
q2 = session.execute(query)._current_rows 
q2
##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 user_by_song "
query = query + "(song text, userId int, user_first_name text, user_last_name text, PRIMARY KEY(song, userId))"
try:
    session.execute(query)
except Exception as e:
    print(e) 

                    
with open(file, encoding = 'utf8') as f:
    csvreader = csv.reader(f)
    next(csvreader) 
    for line in csvreader:
        query = "INSERT INTO user_by_song (userId, user_first_name, user_last_name, song)"
        query = query + "VALUES (%s, %s, %s, %s)"
        session.execute(query, (int(line[10]), line[1], line[4], line[9]))<jupyter_output><empty_output><jupyter_text>In the third query we are interested in getting songs listened by each user, retrieving user first and last name as well as the song, to make every primary key unique we partition by song and user id<jupyter_code>#QUERY 3
session.row_factory = pandas_factory

query = "SELECT user_first_name, user_last_name, song FROM user_by_song\
    where song='All Hands Against His Own'"

q3 = session.execute(query)._current_rows 
q3<jupyter_output><empty_output><jupyter_text>Drop Tables<jupyter_code>query = "drop table music_sessions"
try:
    rows = session.execute(query)
except Exception as e:
    print(e)
query = "drop table user_sessions"
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.10006101281269067, 'fraction_numerical': 0.011470408785845028, 'mean_word_length': 4.602187286397813, 'pattern_counts': {'":': 0, '<': 47, '<?xml version=': 0, '>': 47, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 1, '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': '12', 'dir': 'jupyter-structured-clean-dedup', 'id': '662592', 'license': 'FTL', 'n_tokens_mistral': 2712, 'n_tokens_neox': 2574, 'n_words': 863, 'path': '.ipynb_checkpoints/Project_1B_ Project_Template-checkpoint.ipynb', 'repo_name': 'Lamasheg/DataModelingWithApacheCassandra', 'size': '8195'}