<jupyter_start><jupyter_text>ETL ProcessesUse this notebook to develop the ETL process for each of your tables before completing the `etl.py` file to load the whole datasets.  Standard Libraries<jupyter_code>import os
import glob
import psycopg2
import pandas as pd
import json
from io import StringIO
import logging
import datetime<jupyter_output><empty_output><jupyter_text>Custom Libraries<jupyter_code>from postgre import Postgre<jupyter_output><empty_output><jupyter_text>Setting Up Logging<jupyter_code>logger = logging.getLogger()
logger.setLevel(logging.INFO)
logger.addHandler(logging.StreamHandler())<jupyter_output><empty_output><jupyter_text>Utility FunctionsThe functions below serve as utility throughout the code to perform simple operations<jupyter_code>def get_json_data(filepath):
    for root, dirs, files in os.walk(filepath):
        files = glob.glob(os.path.join(root,'*.json'))
        for file in files :
            with open(os.path.abspath(file), 'r') as f:
                try:
                    yield json.load(f)
                except:
                    with open(os.path.abspath(file), 'r') as f:
                        for line in f.readlines():
                            try:
                                yield json.loads(line)
                            except:
                                yield None

def df_rows_to_list(df:pd.DataFrame, number_of_rows:int=None):
    list_of_row_values = []
    number_of_rows = number_of_rows if number_of_rows else df.shape[0]
    for index in range(0, number_of_rows):
        list_of_row_values.append(list(df.values[index]))
    
    return list_of_row_values
def list_to_string_io(list_of_entries:list):
    """
    Return file like object of the type StringIO from a given list of list of strings.
    Argument:
        - list_of_entries {list} - list of list of strings to transform to StringIO
    Example:
        [
             ['AR8IEZO1187B99055E', 'SOINLJW12A8C13314C', 'City Slickers', 2008, 149.86404],
             ['AR558FS1187FB45658', 'SOGDBUF12A8C140FAA', 'Intro', 2003, 75.67628]
        ]
        
    Return:
        {StringIO} - file type object with values in input list concatenated.
    
    Example:
        'AR8IEZO1187B99055E\tSOINLJW12A8C13314C\tCity Slickers\t2008\t149.86404\n
         AR558FS1187FB45658\tSOGDBUF12A8C140FAA\tIntro\t2003\t75.67628'
    """
    return StringIO('\n'.join(['\t'.join([str(entry) for entry in set_of_entries]) for set_of_entries in list_of_entries]))<jupyter_output><empty_output><jupyter_text>Establish Connection to Postgre and Create Sparkify Database<jupyter_code>sparkifydb = 'sparkifydb'
Postgre('studentdb').create_database(sparkifydb)
postgre = Postgre(sparkifydb)<jupyter_output>Connection Established to: studentdb
Database sparkifydb dropped: True
Database sparkifydb created: True
Connection Established to: sparkifydb<jupyter_text>Process `song_data`<jupyter_code>songs_filepath = "./data/song_data"
songs_data =[json_data for json_data in get_json_data(songs_filepath) if json_data]
songs_df = pd.DataFrame.from_dict(songs_data)
songs_df.sample(5)<jupyter_output><empty_output><jupyter_text>Establish Tables Info<jupyter_code>songs_table_name = "songs"
artists_table_name = "artists"
time_table_name = "time"
users_table_name = "users"
songsplay_table_name = 'songplays'
table_info ={
    songs_table_name: {'artist_id': 'CHAR(18) NOT NULL',
                       'song_id': 'CHAR(18) NOT NULL',
                       'title': 'VARCHAR(100)',
                       'year': 'INTEGER',
                       'duration':'REAL'
                      },
    artists_table_name: {'artist_id': 'CHAR(18) NOT NULL',
                         'artist_name': 'VARCHAR(100)',
                         'artist_location': 'VARCHAR(100)',
                         'artist_latitude': 'REAL',
                         'artist_longitude':'REAL'
                      },
    time_table_name: {'day': 'INTEGER',
                      'hour': 'INTEGER',
                      'month': 'INTEGER',
                      'timestamp': 'TIMESTAMP(6)',
                      'week': 'INTEGER',
                      'weekday': 'INTEGER',
                      'year': 'INTEGER',       
                    },
    users_table_name: {'userId': 'INTEGER',
                      'firstName': 'VARCHAR(55)',
                      'lastName': 'VARCHAR(55)',
                      'gender': 'CHAR(1)',
                      'level': 'CHAR(10)'   
                        },
    songsplay_table_name:{
                "ts":"BIGINT",
                 "userId":"INTEGER",
                 "level":"CHAR(10)",
                 "song_id":"CHAR(18)",
                 "artist_id":"CHAR(18)",
                 "sessionId":"INTEGER",
                 "location":"VARCHAR(255)",
                 "userAgent":"VARCHAR(255)",
        }
    }<jupyter_output><empty_output><jupyter_text>1: `songs` Table Extract Data for Songs Table- Select columns for song ID, title, artist ID, year, and duration- Use `df.values` to select just the values from the dataframe- Index to select the first (only) record in the dataframe- Convert the array to a list and set it to `song_data`  Insert Record into Song TableImplement the `song_table_insert` query in `sql_queries.py` and run the cell below to insert a record for this song into the `songs` table. Remember to run `create_tables.py` before running the cell below to ensure you've created/resetted the `songs` table in the sparkify database  Select Single Row from DF and Transform it to String IO<jupyter_code>column_names = list(table_info.get(songs_table_name).keys())
song_selected_df = songs_df[column_names]
song_data = df_rows_to_list(song_selected_df, number_of_rows=1)
file_object = list_to_string_io(song_data)
song_data<jupyter_output><empty_output><jupyter_text>Create Song Table and Insert Single Record<jupyter_code>postgre.create_table(table_name=songs_table_name, columns_dict=table_info.get(songs_table_name))
postgre.copy_to_table(file_object, table_name=songs_table_name, columns=column_names)<jupyter_output>Table songs dropped: True
Table songs created: True
Inserted 1 Records to Table songs<jupyter_text>Verify Record got Inserted<jupyter_code>results = postgre._execute_query("select * from songs;", results=True)
results<jupyter_output><empty_output><jupyter_text>2: `artists` Table Extract Data for Artists Table- Select columns for artist ID, name, location, latitude, and longitude- Use `df.values` to select just the values from the dataframe- Index to select the first (only) record in the dataframe- Convert the array to a list and set it to `artist_data`  Insert Record into Artist TableImplement the `artist_table_insert` query in `sql_queries.py` and run the cell below to insert a record for this song's artist into the `artists` table. Remember to run `create_tables.py` before running the cell below to ensure you've created/resetted the `artists` table in the sparkify database.<jupyter_code>column_names = list(table_info.get(artists_table_name).keys())
artist_df = songs_df[column_names]
artist_data = df_rows_to_list(artist_df, number_of_rows=1)
file_object = list_to_string_io(artist_data)
artist_data
postgre.create_table(table_name=artists_table_name, columns_dict=table_info.get(artists_table_name))
postgre.copy_to_table(file_object, table_name=artists_table_name, columns=column_names)
results = postgre._execute_query("select * from artists;", results=True)
results<jupyter_output><empty_output><jupyter_text>Process `log_data`In this part, you'll perform ETL on the second dataset, `log_data`, to create the `time` and `users` dimensional tables, as well as the `songplays` fact table.Let's perform ETL on a single log file and load a single record into each table.- Use the `get_files` function provided above to get a list of all log JSON files in `data/log_data`- Select the first log file in this list- Read the log file and view the data<jupyter_code>logs_filepath = "./data/log_data"
logs_data =[json_data for json_data in get_json_data(logs_filepath) if json_data]
logs_df = pd.DataFrame.from_dict(logs_data)
logs_df.sample(5)<jupyter_output><empty_output><jupyter_text>3: `time` Table Extract Data for Time Table- Filter records by `NextSong` action- Convert the `ts` timestamp column to datetime  - Hint: the current timestamp is in milliseconds- Extract the timestamp, hour, day, week of year, month, year, and weekday from the `ts` column and set `time_data` to a list containing these values in order  - Hint: use pandas' [`dt` attribute](https://pandas.pydata.org/pandas-docs/stable/reference/api/pandas.Series.dt.html) to access easily datetimelike properties.- Specify labels for these columns and set to `column_labels`- Create a dataframe, `time_df,` containing the time data for this file by combining `column_labels` and `time_data` into a dictionary and converting this into a dataframe<jupyter_code>time_df = logs_df[logs_df['page'] == 'NextSong'].reset_index(drop=True)
time_df.head()
def convert_timestamp(timestamp_miliseconds_list:list):
    series = []
    for record in timestamp_miliseconds_list:
         series.append(datetime.datetime.fromtimestamp(record/1000.0))
    return series

time_df = time_df.apply(lambda x: convert_timestamp(x) if x.name == 'ts' else x)
time_data = (time_df.ts, time_df.ts.dt.hour, time_df.ts.dt.day, time_df.ts.dt.week, time_df.ts.dt.month, time_df.ts.dt.year, time_df.ts.dt.weekday)
column_labels = ('timestamp', 'hour', 'day', 'week', 'month', 'year', 'weekday')
time_dict = []
for index in range(0, len(time_data[0])):
    time_dict.append({key:value for key, value in zip(column_labels, (entry[index] for entry in time_data))})
time_df = pd.DataFrame(time_dict)
time_df.head()
column_names = list(table_info.get(time_table_name).keys())
file_object = StringIO()
time_df.to_csv(file_object, columns=column_names, sep='\t', header=False, index=False)
file_object.seek(0)
postgre.create_table(table_name=time_table_name, columns_dict=table_info.get(time_table_name))
postgre.copy_to_table(file_object, table_name=time_table_name, columns=column_names)
results = postgre._execute_query("select * from time;", results=True)
results[0]<jupyter_output>Table time dropped: True
Table time created: True
Inserted 6820 Records to Table time<jupyter_text>4: `users` Table Extract Data for Users Table- Select columns for user ID, first name, last name, gender and level and set to `user_df`  Insert Records into Users TableImplement the `user_table_insert` query in `sql_queries.py` and run the cell below to insert records for the users in this log file into the `users` table. Remember to run `create_tables.py` before running the cell below to ensure you've created/resetted the `users` table in the sparkify database.<jupyter_code>column_names = list(table_info.get(users_table_name).keys())
user_df = logs_df[column_names]
file_object = StringIO()
user_df.to_csv(file_object, columns=column_names, sep='\t', header=False, index=False)
file_object.seek(0)
postgre.create_table(table_name=users_table_name, columns_dict=table_info.get(users_table_name))
postgre.copy_to_table(file_object, table_name=users_table_name, columns=column_names)
results = postgre._execute_query("select * from users;", results=True)
results[0]<jupyter_output>Table users dropped: True
Table users created: True
Inserted 8056 Records to Table users<jupyter_text>Insert full data to tables<jupyter_code>tables_dataframes ={
    songs_table_name: song_selected_df,
    artists_table_name: artist_df,
    time_table_name: time_df,
    users_table_name: user_df
    }
for table_name, table_df in tables_dataframes.items():
    postgre.create_table(table_name=table_name, columns_dict=table_info.get(table_name))
    postgre.copy_dataframe_to_table(table_df, table_name)<jupyter_output>Table songs dropped: True
Table songs created: True
Inserted 72 Records to Table songs
Table artists dropped: True
Table artists created: True
Inserted 72 Records to Table artists
Table time dropped: True
Table time created: True
Inserted 6820 Records to Table time
Table users dropped: True
Table users created: True
Inserted 8056 Records to Table users<jupyter_text>5: `songplays` Table Extract Data and Songplays TableThis one is a little more complicated since information from the songs table, artists table, and original log file are all needed for the `songplays` table. Since the log file does not specify an ID for either the song or the artist, you'll need to get the song ID and artist ID by querying the songs and artists tables to find matches based on song title, artist name, and song duration time.- Implement the `song_select` query in `sql_queries.py` to find the song ID and artist ID based on the title, artist name, and duration of a song.- Select the timestamp, user ID, level, song ID, artist ID, session ID, location, and user agent and set to `songplay_data` Insert Records into Songplays Table- Implement the `songplay_table_insert` query and run the cell below to insert records for the songplay actions in this log file into the `songplays` table. Remember to run `create_tables.py` before running the cell below to ensure you've created/resetted the `songplays` table in the sparkify database.<jupyter_code>def get_song_artist_id(postgre:Postgre, song_title:str, artist_name, song_duration:int):
    results = [None, None]
    if all((song_title, artist_name, song_duration)):
        song_select ="""SELECT DISTINCT a.song_id, a.artist_id
                    FROM {SONGS_TABLE_NAME} a
                    FULL JOIN {ARTISTS_TABLE_NAME} b ON a.artist_id = b.artist_id
                    WHERE a.title='{SONG_TITLE}'
                    -- AND a.duration={SONG_DURATION}
                    -- AND b.artist_name='{ARTIST_NAME}'
                    """

        results = postgre._execute_query(song_select.format(
                                                    SONGS_TABLE_NAME=songs_table_name,
                                                    ARTISTS_TABLE_NAME=artists_table_name,
                                                    SONG_TITLE=song_title.replace("'","''"),
                                                    SONG_DURATION=song_duration,
                                                    ARTIST_NAME=artist_name.replace("'","''") ), 
                                         results=True)
        if results:
            results = list(results[0].values())

    return results
logs_df[['song_id', 'artist_id']] = logs_df.apply(lambda row: pd.Series(get_song_artist_id(postgre, row['song'], row['artist'], row['length'])), axis=1)
column_names = list(table_info.get(songsplay_table_name).keys())
songplay_df = logs_df[column_names]
postgre.create_table(table_name=songsplay_table_name, columns_dict=table_info.get(songsplay_table_name))
postgre.copy_dataframe_to_table(songplay_df, songsplay_table_name)<jupyter_output>Table songplays dropped: True
Table songplays created: True
Inserted 8056 Records to Table songplays<jupyter_text>Close Connection to Sparkify Database<jupyter_code>postgre.conn.close()<jupyter_output><empty_output><jupyter_text>Implement `etl.py`Use what you've completed in this notebook to implement `etl.py`.<jupyter_code>from etl import main
main()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.09228857348420433, 'fraction_numerical': 0.013277519785466676, 'mean_word_length': 3.7976153122058363, 'pattern_counts': {'":': 8, '<': 67, '<?xml version=': 0, '>': 67, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '18', 'dir': 'jupyter-structured-clean-dedup', 'id': '369698', 'license': 'MIT', 'n_tokens_mistral': 4857, 'n_tokens_neox': 4618, 'n_words': 1287, 'path': 'postgres/etl.ipynb', 'repo_name': 'alexanderluiscampino/sparkify', 'size': '15289'}