import os import glob import psycopg2 import pandas as pd from sql_queries import * def process_song_file(cur, filepath): """ Description: This function can be used to read the file in the filepath (data/song_data) to get the song records and used to populate the song and artist dim tables. Arguments: cur: cursor reference. filepath: complete file path for the file to load. Returns: None """ # open song file df = pd.read_json(filepath, typ='series') # insert song record song_data = df[['song_id','title','artist_id','year','duration']].values cur.execute(song_table_insert, song_data) # insert artist record artist_data = df[['artist_id', 'artist_name', 'artist_location', 'artist_latitude', 'artist_longitude']] cur.execute(artist_table_insert, artist_data) def process_log_file(cur, filepath): """ Description: This function can be used to read the file in the filepath (data/log_data) to get the log records and used to populate the time, user dim tables. Also build the songplay fact table. Arguments: cur: cursor reference. filepath: complete file path for the file to load. Returns: None """ # open log file df = pd.read_json(filepath, lines=True) # filter by NextSong action df = df[df['page'].str.contains('NextSong')] # convert timestamp column to datetime t = pd.to_datetime(df['ts'], unit='ms') # insert time data records time_data = (t, t.dt.hour, t.dt.day, t.dt.week, t.dt.month, t.dt.year, t.dt.weekday) column_labels = ('timestamp', 'hour', 'day', 'week', 'month', 'year', 'weekday') time_df = pd.DataFrame(dict(zip(column_labels,time_data))) for i, row in time_df.iterrows(): cur.execute(time_table_insert, list(row)) # load user table user_df = df[['userId', 'firstName', 'lastName', 'gender', 'level']] # insert user records for i, row in user_df.iterrows(): cur.execute(user_table_insert, row) # insert songplay records for index, row in df.iterrows(): # get songid and artistid from song and artist tables cur.execute(song_select, (row.song, row.artist, row.length)) results = cur.fetchone() if results: songid, artistid = results else: songid, artistid = None, None # insert songplay record # ref.: https://stackoverflow.com/questions/35312981/using-pandas-to-datetime-with-timestamps start_time = pd.to_datetime(row.ts, unit='ms').strftime('%Y-%m-%d %I:%M:%S') songplay_data = (index, start_time, row.userId, row.level, str(songid), str(artistid), row.sessionId, row.location, row.userAgent) cur.execute(songplay_table_insert, songplay_data) def process_data(cur, conn, filepath, func): """ Process function to load data from songs and event log files into Postgres database. Arguments: cur: cursor reference. conn: connection credential for database access. filepath: complete file path for the file to load. func: function to call Returns: None """ # get all files matching extension from directory all_files = [] for root, dirs, files in os.walk(filepath): files = glob.glob(os.path.join(root,'*.json')) for f in files : all_files.append(os.path.abspath(f)) # get total number of files found num_files = len(all_files) print('{} files found in {}'.format(num_files, filepath)) # iterate over files and process for i, datafile in enumerate(all_files, 1): func(cur, datafile) conn.commit() print('{}/{} files processed.'.format(i, num_files)) def main(): """ Main function for loading songs and log data into Postgres database """ conn = psycopg2.connect("host=127.0.0.1 dbname=sparkifydb user=student password=student") cur = conn.cursor() process_data(cur, conn, filepath='data/song_data', func=process_song_file) process_data(cur, conn, filepath='data/log_data', func=process_log_file) conn.close() if __name__ == "__main__": main()