<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
import cassandra
import re
import os
import glob
import numpy as np
import json
import csv
from decimal import Decimal<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,'*'))
    #print(file_path_list)<jupyter_output>/home/workspace<jupyter_text>An event file looks like.![Sample event log!](./images/original_event_log_sample.jpg "2018-11-01-events.csv")  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) 
            
# uncomment the code below if you would like to get total number of rows 
#print(len(full_data_rows_list))
# uncomment the code below if you would like to check to see what the list of event data rows will look like
#print(full_data_rows_list)

# 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. Complete the Apache Cassandra coding portion of this project.  Now event_datafile_new.csv file is ready, 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:  Let's connect to Cassandra Cluser and have a Keyspace ready  Creating a Cluster<jupyter_code># This should make a connection to a Cassandra instance your local machine (127.0.0.1)
# If your Cassandra is on remote machine, please put remote IP

from cassandra.cluster import Cluster
try:
    cluster = Cluster(['127.0.0.1'])
# To establish connection and begin executing queries, need a session
    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 song_play 
    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('song_play')
except Exception as e:
    print(e)<jupyter_output><empty_output><jupyter_text>Query requirement from Analyst team 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'  Create tables Table 1: song_play_by_sessionFor 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```pythonquery1 = "SELECT artist, song, length FROM song_play_by_session WHERE sessionId=338 AND itemInSession=4"``` This query is straight forward which requires **combination PK (sessionID, itemInSession)**  - column 1: sessionId  **<-- Partition Key**  - column 2: itemInSession **<-- Clustering Column 1**  - column 3: artist  - column 4: song (song title)  - column 5: length (length of song in seconds)  - PRIMARY KEY: sessionId, itemInSession<jupyter_code>create_table = "CREATE TABLE IF NOT EXISTS song_play_by_session \
                (sessionId int, itemInSession int, artist text, song text, length double \
                , PRIMARY KEY (sessionId, itemInSession))"
try:
    session.execute(create_table)
except Exception as e:
    print(e)<jupyter_output><empty_output><jupyter_text>Table 2: song_play_by_userFor 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```pythonquery2 = "SELECT artist, song, first_name, last_name FROM song_play_by_user WHERE userId=10 AND sessionId=182"``` This query is a little bit tricky because it requires **sorted by itemInSession** which must be a part of PK. Therefore, the **PK becomes combination of ((userId, sessionId), itemInSession)**, where itemInSession is not used in WHERE clause but implicitly affects the order of query output.  - column 1: userId  **<-- Partition Key**  - column 2: sessionId  **<-- Clustering Column 1**  - column 3: itemInSession  **<-- Clustering Column 2, Have the effect of sorting query output in ascending order**  - column 4: artist  - column 5: song  - column 6: first_name  - column 7: last_name  - PRIMARY KEY: userId, sessionId, itemInSession<jupyter_code>create_table = "CREATE TABLE IF NOT EXISTS song_play_by_user \
                (userId int, sessionId int, itemInSession int, artist text, song text, first_name text, last_name text \
                , PRIMARY KEY ((userId, sessionId), itemInSession))"
try:
    session.execute(create_table)
except Exception as e:
    print(e)<jupyter_output><empty_output><jupyter_text>Table 3: user_by_song_playedFor 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'```pythonquery3 = "SELECT first_name, last_name FROM user_by_song_played WHERE song='All Hands Against His Own'"``` This query requires output of user's first and last name. Since there can be **different users with the same first & last name**. We better use userId as a part of PK. Therefore, the **PK becomes combination of (song, userId)**, where userId helps separating different users with the same first & last name  - column 1: song  **<-- Partition Key**  - column 2: userId  **<-- Clustering Column 1**  - column 3: first_name  - column 4: last_name    - PRIMARY KEY: song, userId<jupyter_code>create_table = "CREATE TABLE IF NOT EXISTS user_by_song_played \
                (song text, userId int, first_name text, last_name text \
                , PRIMARY KEY (song, userId))"
try:
    session.execute(create_table)
except Exception as e:
    print(e)<jupyter_output><empty_output><jupyter_text>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  Expected output of Query 1 for data in `event_datafile_new.csv` file![Verify query output!](./images/query1.jpg "Query 1 output")<jupyter_code>query1 = "SELECT artist, song, length FROM song_play_by_session WHERE sessionId=338 AND itemInSession=4"
file = 'event_datafile_new.csv'

with open(file, encoding = 'utf8') as f:
    csvreader = csv.reader(f)
    next(csvreader) # skip header
    # for each line of input data, insert into tables defined above
    for line in csvreader:
        # insert1 inserts records to song_play_by_session table
        insert1 = "INSERT INTO song_play_by_session (sessionId, itemInSession, artist, song, length)"
        insert1 = insert1 + " VALUES (%s, %s, %s, %s, %s)"
        try:
            session.execute(insert1, (int(line[8]), int(line[3]), line[0], line[9], Decimal(line[5])))
        except Exception as e:
            print(e)<jupyter_output><empty_output><jupyter_text>Do a SELECT to verify that the data have been inserted into each table<jupyter_code>column_labels = ['artist', 'song', 'length']
try:
    rows = session.execute(query1)
except Exception as e:
    print(e)
    
matches = []
for row in rows:
    #print(row.artist, row.song, row.length)
    matches.append([row.artist, row.song, row.length])

df1 = pd.DataFrame(matches, columns=column_labels)    
print(df1)<jupyter_output>artist                             song    length
0  Faithless  Music Matters (<NAME>)  495.3073<jupyter_text>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  Expected output of Query 2 for data in `event_datafile_new.csv` file![Verify query output!](./images/query2.jpg "Query 2 output")<jupyter_code>query2 = "SELECT artist, song, first_name, last_name FROM song_play_by_user WHERE userId=10 AND sessionId=182"
file = 'event_datafile_new.csv'

with open(file, encoding = 'utf8') as f:
    csvreader = csv.reader(f)
    next(csvreader) # skip header
    # for each line of input data, insert into tables defined above
    for line in csvreader:
        # insert2 inserts records to song_play_by_user table
        insert2 = "INSERT INTO song_play_by_user (userId, sessionId, itemInSession, artist, song, first_name, last_name)"
        insert2 = insert2 + " VALUES (%s, %s, %s, %s, %s, %s, %s)"
        try:
            session.execute(insert2, (int(line[10]), int(line[8]), int(line[3]), line[0], line[9], line[1], line[4]))
        except Exception as e:
            print(e)     
### query2 = "SELECT artist, song, first_name, last_name FROM song_play_by_user WHERE userId=10 AND sessionId=182"
column_labels = ['artist', 'song', 'first_name', 'last_name']
try:
    rows = session.execute(query2)
except Exception as e:
    print(e)
    
matches = []
for row in rows:
    #print(row.artist, row.song, row.first_name, row.last_name)
    matches.append([row.artist, row.song, row.first_name, row.last_name])

df2 = pd.DataFrame(matches, columns=column_labels)    
print(df2)<jupyter_output>artist                                               song  \
0   Down To The Bone                                 Keep On Keepin' On   
1       Three Drives                                        Greece 2000   
2  <NAME>                                          Kilometer   
3      <NAME>  Catch You Baby (<NAME> & Max Sanna Radio...   

  first_name last_name  
0     Sylvie      Cruz  
1     Sylvie      Cruz  
2     Sylvie      Cruz  
3     Sylvie      Cruz<jupyter_text>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'  Expected output of Query 3 for data in `event_datafile_new.csv` file![Verify query output!](./images/query3.jpg "Query 3 output")<jupyter_code>query3 = "SELECT first_name, last_name FROM user_by_song_played WHERE song='All Hands Against His Own'"
file = 'event_datafile_new.csv'

with open(file, encoding = 'utf8') as f:
    csvreader = csv.reader(f)
    next(csvreader) # skip header
    # for each line of input data, insert into tables defined above
    for line in csvreader:
        # insert3 inserts records to user_by_song_played table
        insert3 = "INSERT INTO user_by_song_played (song, userId, first_name, last_name)"
        insert3 = insert3 + " VALUES (%s, %s, %s, %s)"
        try:
            session.execute(insert3, (line[9], int(line[10]), line[1], line[4]))
        except Exception as e:
            print(e)
column_labels = ['first_name', 'last_name']
try:
    rows = session.execute(query3)
except Exception as e:
    print(e)

matches = []
for row in rows:
    #print(row.first_name, row.last_name)
    matches.append([row.first_name, row.last_name])
    
df3 = pd.DataFrame(matches, columns=column_labels)
print(df3)<jupyter_output>first_name last_name
0  Jacqueline     Lynch
1       Tegan    Levine
2        Sara   Johnson<jupyter_text>Drop the tables before closing out the sessions<jupyter_code>## Drop song_play_by_session table
drop1 = "DROP TABLE IF EXISTS song_play_by_session"
try:
    rows = session.execute(drop1)
except Exception as e:
    print(e)
## Drop song_play_by_user table
drop2 = "DROP TABLE IF EXISTS song_play_by_user"
try:
    rows = session.execute(drop2)
except Exception as e:
    print(e)
## Drop user_by_song_played table
drop3 = "DROP TABLE IF EXISTS user_by_song_played"
try:
    rows = session.execute(drop3)
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.09296937223951922, 'fraction_numerical': 0.01563970747954529, 'mean_word_length': 3.8091922005571033, 'pattern_counts': {'":': 0, '<': 67, '<?xml version=': 0, '>': 60, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 7, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '15', 'dir': 'jupyter-structured-clean-dedup', 'id': '550627', 'license': 'Apache-2.0', 'n_tokens_mistral': 4387, 'n_tokens_neox': 4160, 'n_words': 1514, 'path': 'Project_1B_ Project.ipynb', 'repo_name': 'iyiiyi/cassandra_etl', 'size': '13844'}