<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
from decimal import Decimal
import shutil<jupyter_output><empty_output><jupyter_text>Creating list of filepaths to process original event csv data files<jupyter_code># Get current folder and subfolder event data
filepath = os.getcwd() + '/event_data'
file_path_list=[]

#DELETE FOLDER IF EXIST TO SELECT ONLY NEEDED DATA
try:
    shutil.rmtree(filepath+'/.ipynb_checkpoints')
except Exception as e :
    print(e)
    
#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,'*events.csv')))
    #print(file_path_list)
    #print(len(file_path_list))<jupyter_output>[Errno 2] No such file or directory: '/home/workspace/event_data/.ipynb_checkpoints'<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:
            #print(line)
            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 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. 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:  Creating a Cluster<jupyter_code>#connecting to local machine casendra
from cassandra.cluster import Cluster
cluster = Cluster()

#  Creating a session
session = cluster.connect()<jupyter_output><empty_output><jupyter_text>Create Keyspace<jupyter_code>try:
    session.execute("""
    CREATE KEYSPACE IF NOT EXISTS SPARKIFY
    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('sparkify')
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'  ------------------------------------------------------------------------------------------------------------------------------------------------------------  Query 1:  SELECT the artist, song title and song's length in the music app history that was heard during sessionId = 338, and itemInSession = 4so our data will by partitioned by sessionId and clustered by itemInSession<jupyter_code>try:
    session.execute('DROP TABLE IF EXISTS song_hist') 
except Exception as e :
    print(e)
try:
    query='CREATE TABLE song_hist'
    query+="""(sess_id int,
    sess_item int ,
    user_id int,
    first_name varchar,
    last_name varchar ,
    gender varchar ,
    location text,
    level varchar,
    artist varchar,
    song varchar,
    length double,
    primary key (sess_id,sess_item)
    )"""
    session.execute(query) 
except Exception as e :
    print(e)                  
#reading the smaller file and inserting data in table
file = 'event_datafile_new.csv'

with open(file, encoding = 'utf8') as f:
    csvreader = csv.reader(f)
    next(csvreader) # skip header
    for line in csvreader:

        query = """INSERT INTO song_hist (sess_id,
        user_id  ,
        sess_item ,
        first_name ,
        last_name ,
        gender  ,
        location,
        level,
        artist,
        song ,
        length)"""
        query = query + "VALUES(%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)"
        session.execute(query, (int(line[8]), int(line[10]), int(line[3]), line[1], line[4], line[2], line[7], line[6], line[0], line[9], Decimal(line[5])))
    
#Query1: select the artist, song title and song's length in the music app history that was heard during sessionId = 338, and itemInSession = 4 
    
try:
        query ='SELECT artist,song ,length FROM song_hist where sess_id=338 and sess_item=4'
        rows =session.execute(query) 
except Exception as e :
        print(e)
    
for row in rows:
        print (row.artist,row.song,row.length)<jupyter_output>Faithless Music Matters (<NAME>) 495.3073<jupyter_text>Query 2 : Select only the following: name of artist, song (sorted by itemInSession) and user (first and last name) for userid = 10, sessionid = 182So our table will be partitioned by sessionId & userId and clustered by items in Session for uniqueness<jupyter_code>try:
    session.execute('DROP TABLE IF EXISTS song_of_sess') 
except Exception as e :
    print(e)
try:
    query='CREATE TABLE song_of_sess'
    query+="""(
    sess_id int,
    user_id int,
    sess_item int ,
    first_name varchar,
    last_name varchar ,
    gender varchar ,
    location text,
    level varchar,
    artist varchar,
    song varchar,
    length double,
    primary key ((sess_id,user_id),sess_item)
    )"""
    session.execute(query) 
except Exception as e :
    print(e)  
                    
#reading the smaller file and inserting data in table
file = 'event_datafile_new.csv'

with open(file, encoding = 'utf8') as f:
    csvreader = csv.reader(f)
    next(csvreader) # skip header
    for line in csvreader:

        query = """INSERT INTO song_of_sess (sess_id,
        user_id  ,
        sess_item ,
        first_name ,
        last_name ,
        gender  ,
        location,
        level,
        artist,
        song ,
        length)"""
        query = query + "VALUES(%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)"
        session.execute(query, (int(line[8]), int(line[10]), int(line[3]), line[1], line[4], line[2], line[7], line[6], line[0], line[9], Decimal(line[5])))
  
#Query2:Select only the following: name of artist, song (sorted by itemInSession) and user (first and last name) for userid = 10, sessionid = 182
    
try:
    query='SELECT artist,sess_item,song,first_name,last_name FROM song_of_sess where  sess_id=182 and user_id=10 '
    rows=session.execute(query) 
except Exception as e :
    print(e)
    
for row in rows:
    print (row.artist,row.sess_item,row.song,row.first_name,row.last_name)<jupyter_output>Down To The Bone 0 Keep On Keepin' On <NAME>
Three Drives 1 Greece 2000 <NAME>
<NAME> 2 Kilometer <NAME>
<NAME> 3 Catch You Baby (<NAME> & Max Sanna Radio Edit) <NAME><jupyter_text>Query 3: SELECT every user name (first and last) in my music app history who listened to the song 'All Hands Against His Own'So our table will be partitioned by song and for uniqueness we will add userId  to clustering columns to identify row<jupyter_code>try:
    session.execute('DROP TABLE IF EXISTS user_of_song ') 
except Exception as e :
    print(e)
try:
    query='CREATE TABLE user_of_song '
    query+="""(song varchar,
    user_id int,
    first_name varchar,
    last_name varchar ,
    location text,
    level varchar,
    gender varchar ,
    artist varchar,
    length double,
    sess_id int,
    sess_item int ,
    primary key (song,user_id)
    )"""
    session.execute(query) 
except Exception as e :
    print(e)  

                    
#reading the smaller file and inserting data in table

file = 'event_datafile_new.csv'

with open(file, encoding = 'utf8') as f:
    csvreader = csv.reader(f)
    next(csvreader) # skip header
    for line in csvreader:

        query = """INSERT INTO user_of_song  (sess_id,
        user_id  ,
        sess_item ,
        first_name ,
        last_name ,
        gender  ,
        location,
        level,
        artist,
        song ,
        length)"""
        query = query + "VALUES(%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)"
        session.execute(query, (int(line[8]), int(line[10]), int(line[3]), line[1], line[4], line[2], line[7], line[6], line[0], line[9], Decimal(line[5])))
#Query3:Select  every user name (first and last) in my music app history who listened to the song 'All Hands Against His Own'
    
try:
    query="SELECT first_name,last_name FROM user_of_song  where song='All Hands Against His Own'"
    rows=session.execute(query) 
except Exception as e :
    print(e)
    
for row in rows:
    print (row.first_name,row.last_name)<jupyter_output><NAME>
<NAME>
<NAME><jupyter_text>Dropping the tables<jupyter_code>##Dropping tables
try:
    session.execute('DROP TABLE IF EXISTS song_hist') 
except Exception as e :
    print(e)

try:
    session.execute('DROP TABLE IF EXISTS song_of_sess') 
except Exception as e :
    print(e)
    
try:
    session.execute('DROP TABLE IF EXISTS user_of_song') 
except Exception as e :
    print(e)<jupyter_output><empty_output><jupyter_text>Closing the session and cluster connection<jupyter_code>session.shutdown()
cluster.shutdown()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.11262296589133033, 'fraction_numerical': 0.011400202261653029, 'mean_word_length': 3.323529411764706, 'pattern_counts': {'":': 0, '<': 51, '<?xml version=': 0, '>': 51, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, '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': '255413', 'license': 'MIT', 'n_tokens_mistral': 3600, 'n_tokens_neox': 3371, 'n_words': 1125, 'path': 'Data Modeling with Cassandra/Data Modeling with Cassandra.ipynb', 'repo_name': 'Abuelseaoud/Data-Engineering-Nanodegree', 'size': '10952'}