<jupyter_start><jupyter_text>Sparkify ProjectEste projeto foi proposto pelo Nanodegree Data Engineer da Udacity. Introdução**Modelagem de Dados com Postgres e ETL pipeline**Uma startup chamada Sparkify quer analisar os dados que eles tem coletados das músicas e atividades dos usuários no seu novo aplicativo de streaming de música. O time de análises está interessando em entender quais músicas os usuários estão ouvindo. Atualmente, eles não tem uma forma fácil de consultar esses dados, que estão num diretório em formato JSON, um com os dados dos logs das atividades dos usuarios no aplicativo e outro com os metadados das musicas na biblioteca do aplicativo.Eles gostariam que um engenheiro de dados criasse um banco de dados Postgres com as tabelas desenvolvidas para otimizar as consultas das análises das músicas reproduzidas. Seu papel é criar um database schema e um ETL pipeline para estas análises. Descrição do ProjetoNeste projeto, iremos aplicar o que aprendemos em modelagem de dados com Postgres e construir um pipeline para extrair, transformar e carregar os dados no banco de dados usando Python. Para completar o projeto, precisaremos definir as tabelas fatos e dimensão, usando um star schema para facilitar as consultas do time de análise de dados, além de escrever um pipeline de dados para transferir os dados dos arquivos locais em formato JSON para estas tabelas criadas no Postgres, usando Python e SQL Consiste em extrair os dados de duas fontes:- song_data: Exemplo de cada registro`{"num_songs": 1, "artist_id": "ARMJAGH1187FB546F3", "artist_latitude": 35.14968, "artist_longitude": -90.04892, "artist_location": "Memphis, TN", "artist_name": "The Box Tops", "song_id": "SOCIWDW12A8C13D406", "title": "Soul Deep", "duration": 148.03546, "year": 1969}`- log_data: Exemplo de cada registro`{"artist":"Des'ree","auth":"Logged In","firstName":"Kaylee","gender":"F","itemInSession":1,"lastName":"Summers","length":246.30812,"level":"free","location":"Phoenix-Mesa-Scottsdale, AZ","method":"PUT","page":"NextSong","registration":1540344794796.0,"sessionId":139,"song":"You Gotta Be","status":200,"ts":1541106106796,"userAgent":"\"Mozilla\/5.0 (Windows NT 6.1; WOW64) AppleWebKit\/537.36 (KHTML, like Gecko) Chrome\/35.0.1916.153 Safari\/537.36\"","userId":"8"}`  Print da tabela dos logs de eventos dos usuários:![Screenshot of log_data](images/log-data.png)  Definindo Schema e tabelas de relacionamentosComo o intuito é facilitar as consultas para a área de análise de dados executar e obter dados para futuras análises, vamos modelar seguindo o modelo Star Schema, um processamento OLAP (On-line Analytical Processing), que é desenhado para performar melhor e tem as seguintes características:- Aplicação: No nível estratégico, auxilia na análise empresarial e tomada de decisões;- Funcionalidade: Gera análises e relatórios gerenciais com leitura otimizada;- Estrutura de dados: Poucos detalhes, pois tem alto nível de sumarização;- Armazenamento dos dados: Utiliza-se da Data Warehouse para otimizar o desempenho da grande quantidade de dados;- Usuários: Destinados aos gestores e time analítico;- Frequência de utilização: Baixa, conforme programação da empresa;- Volatilidade: Dados não sofrem alterações, pois os usuários apenas realizarão sua leitura.O OLAP (On-line Analytical Processing) é voltado para a tomada de decisões, proporciona uma visão dos dados orientado à análise, além de uma navegação rápida e flexível. O OLAP recebe dados do OLTP (On-line Transactional Processing) para que possa realizar as análises. Essa carga de dados acontece conforme a necessidade da empresa. Sendo um sistema para tomada de decisões, não realiza transações (INSERT, UPDATE, DELETE) pois sua finalidade são consultas. Possui dados atuais e históricos e não há necessidade de backups regularmente, sendo que ele possui informações do OLTP. Caso algo aconteça com a base OLAP basta fazer uma carga novamente. TabelasVamos agora começar a definir nossas tabelas fato e tabelas dimensão, lembrando que uma tabela fato é um evento, uma venda, uma transação ocorrida, um fato transacional que ocorreu no nosso sistema, no caso, quando um usuário acessa o aplicativo sparkify e clica numa música para ouvir, esses dados da musica selecionada, do usuário, localização, página em que ele estava e etc, fazem parte do evento "tocar música", esse é um bom candidato para nossa tabela fato. As tabelas fatos geralmente são de dados numéricos e não categóricos. Resumindo:A tabela fato é a principal tabela do Data Warehouse, ela vai conectar nas dimensões. Nessa tabela são armazenadas duas coisas: as métricas, que são os fatos propriamente ditos, e as foreign keys, chaves que servem para ligar os dados das dimensões com a fato. Ou seja, a tabela fato é composta pelas métricas, que são tudo aquilo que a empresa quer medir, junto com as foreign keys, chaves que ligam às dimensões que descrevem essas métricas. As métricas são utilizadas para medir, quantificar algo, são sempre números provenientes de transações da empresa. Tudo que a empresa quer mensurar é métrica, geralmente sendo o que o usuário quer medir.Mas o que é uma Foreign Key? É uma chave estrangeira que serve para relacionar os dados entre as tabelas fato e dimensão.Já as tabelas Dimensão, são as categorias de cada entidade envolvida na tabela fato, que podem ser usadas para trazer mais informações sobre os dados e facilitar as análises. Os dados do usuário que reproduziu uma música é um exemplo para uma tabela dimensão, o nome, sobrenome, gênero, nível na aplicação são os dados dessa dimensão. Os dados da música selecionadas já fazem parte de outra dimensão, como o nome da música, o ano de lançamento, tempo de duração e etc.A Dimensão possui característica descritiva dentro do DW. Ela qualifica as informações provenientes da tabela Fato. Através dela é possível analisar os dados sob múltiplas perspectivas. Fact Table- songplays: registros nos dados log de eventos associados com músicas reproduzidas    - songplay_id,     - start_time,     - user_id,     - level,     - song_id,     - artist_id,     - session_id,     - location,     - user_agent Dimension Tables- users: usuários do aplicativo    - user_id,     - first_name,     - last_name,     - gender,     - level        - songs: músicas no banco de dados    - song_id,     - title,     - artist_id,     - year,     - duration        - artists: Artistas das músicas no banco de dados    - artist_id,     - name,     - location,     - latitude,     - longitude- time: Timestamps dos registros das músicas reproduzidas quebrados em unidades específicas para facilitar futuras análises    - start_time,     - hour,     - day,     - week,     - month,     - year,     - weekday  Passo a passoPrimeiro vamos escrever as queries para dropar e criar as tabelas e a query para selecionar as músicas que coincidem com o nome do artista e da música para inserir esses dados na tabela fato songplays, depois vamos criar o banco de dados e as tabelas, vou explicando linha a linha para fácil entendimento<jupyter_code># DROPA AS TABELAS

drop_table_songplays = "DROP TABLE IF EXISTS songplays"
drop_table_users = "DROP TABLE IF EXISTS users"
drop_table_songs = "DROP TABLE IF EXISTS songs"
drop_table_artists = "DROP TABLE IF EXISTS artists"
drop_table_time = "DROP TABLE IF EXISTS time"

# CRIA TABELAS

create_table_songplays = ("""
    CREATE TABLE IF NOT EXISTS songplays
    (songplay_id int PRIMARY KEY, 
    start_time date, 
    user_id text NOT NULL,
    level text, 
    song_id text,
    artist_id text,
    session_id int,
    location text,
    user_agent text)
""")

create_table_users = ("""
  CREATE TABLE IF NOT EXISTS users
  (user_id text PRIMARY KEY,
  first_name text NOT NULL,
  last_name text NOT NULL,
  gender text,
  level text)
""")

create_table_songs = ("""
  CREATE TABLE IF NOT EXISTS songs
  (song_id text PRIMARY KEY,
  title text NOT NULL,
  artist_id text NOT NULL,
  year int,
  duration float NOT NULL)
""")

create_table_time = ("""
  CREATE TABLE IF NOT EXISTS time
    (start_time date PRIMARY KEY,
     hour int, 
     day int, 
     week int, 
     month int, 
     year int, 
     weekday text)
""")

create_table_artists = ("""
  CREATE TABLE IF NOT EXISTS artists
  (artist_id text PRIMARY KEY,
  name text NOT NULL,
  location text,
  latitude float,
  longitude float)
""")
# INSERE REGISTROS

insert_song = ("""
  INSERT INTO songs
  (song_id, title, artist_id, year, duration)
  VALUES (%s, %s, %s, %s, %s)
  ON CONFLICT (song_id) DO NOTHING
""")

insert_artist = ("""
  INSERT INTO artists
  (artist_id, name, location, latitude, longitude)
  VALUES (%s, %s, %s, %s, %s)
  ON CONFLICT (artist_id) DO NOTHING
""")

insert_time = ("""
  INSERT INTO time
  (start_time, hour, day, week, month, year, weekday)
  VALUES (%s, %s, %s, %s, %s, %s, %s)
  ON CONFLICT (start_time) DO NOTHING
""")

insert_user = ("""
  INSERT INTO users
  (user_id, first_name, last_name, gender, level)
  VALUES (%s, %s, %s, %s, %s)
  ON CONFLICT (user_id) DO NOTHING
""")

insert_songplays = ("""
  INSERT INTO songplays
  (songplay_id, start_time, user_id, level, song_id, artist_id, session_id, location, user_agent)
  VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
  ON CONFLICT (songplay_id) DO NOTHING
""")

# SELECIONA REGISTROS QUE COINCIDAM COM O NOME DA MUSICA E DO ARTISTA

song_select = ("""
  SELECT song_id, artists.artist_id, duration
  FROM songs 
  JOIN artists 
  ON songs.artist_id = artists.artist_id
  WHERE songs.title = %s
  AND artists.name = %s
  AND songs.duration = %s
""")<jupyter_output><empty_output><jupyter_text>Neste momento temos as queries criadas, vamos disponibilizá-las num array para iterar posteriormente!<jupyter_code>create_table_queries = [create_table_songplays, create_table_users, create_table_songs, create_table_time, create_table_artists]

drop_table_queries = [drop_table_songplays, drop_table_users, drop_table_songs, drop_table_artists, drop_table_time]
# Agora vamos começar a criar o database e as tabelas

import psycopg2

def main():
    """Inicia conexão e cursor para criar database, dropar e criar as tabelas"""
    
    cur, conn = create_database()
    
    drop_tables(cur, conn)
    create_tables(cur, conn)
    
    cur.close()
    conn.close()<jupyter_output><empty_output><jupyter_text>Perceba que a primeira função que será chamada é a função main, a partir dela serão chamadas outras funções.A segunda função chamada será a create_database, que retornará o cursor e a conexão do banco de dados que vamos utilizar nas outras funções para executar os comdandos das queries.<jupyter_code>def create_database():
    """Cria o banco de dados, mas antes faz o drop do banco de dados 
    se ele já existe, e em seguida, cria-o novamente se não existe"""
    
    conn = psycopg2.connect("host=localhost dbname=postgres user=ebraim password=<PASSWORD>")
    conn.set_session(autocommit=True)
    cur = conn.cursor()
    
    cur.execute("DROP DATABASE sparkifydb")
    cur.execute("CREATE DATABASE sparkifydb WITH ENCODING 'utf-8'")
    cur.close()
    conn.close()
    
    conn = psycopg2.connect("host=localhost dbname=sparkifydb user=ebraim password=<PASSWORD>")
    cur = conn.cursor()
    
    return cur, conn<jupyter_output><empty_output><jupyter_text>A função create_database inicia fazendo uma conexão ao database postgres com usuario e senha para acesso.O comando `conn.set_session(autocommit=True)` serve para comitar de forma automática todas vez que executarmos algum comando. Por que precisamos comitar todas execuções? Bem vamos entrar no conceito de ACID.A sigla ACID significa Atomicity, Consistency, Isolation, Durability, ou em português Atomicidade, Consistência, Isolamento e Durabilidade, e diz respeito a um conjunto de propriedades em transações de bancos de dados que são importantes para garantir a validade dos dados mesmo que ocorram erros durante o armazenamento ou problemas mais graves no sistema, como crashes ou problemas físicos em um servidor. As propriedades ACID são fundamentais para o processamento de transações em bancos de dados. Uma transação é, basicamente, uma sequência de operações que satisfazem a essas quatro propriedades. Vamos nos atentar ao primeiro item, Atomicidade: As transações são, geralmente, compostas de várias declarações (comandos / operações). A atomicidade é uma propriedade que garante que cada transação seja tratada como uma entidade única, a qual deve ser executada por completo ou falhar completamente. Desta forma, todas as operações da transação devem ser executadas com sucesso para que a transação tenha sucesso.Se uma única operação que seja do bloco da transação falhar, toda a transação deverá ser cancelada – as transações são aplicadas de uma forma “tudo ou nada”. Caso haja falha em qualquer operação da transação, o banco de dados será retornado ao estado anterior ao início da transação. Chamamos a esse retorno de estado de Rollback (“transação desfeita”). Caso a transação tenha sucesso, o BD é alterado de forma permanente, em um processo denominado Commit (“efetivação”)<jupyter_code>def drop_tables(cur, conn):
    """Drops all tables created on the database"""
    for query in drop_table_queries:
        cur.execute(query)
        conn.commit()

def create_tables(cur, conn):
    """Create all tables on the database"""
    for query in create_table_queries:
        cur.execute(query)
        conn.commit()<jupyter_output><empty_output><jupyter_text>As funções drop_tables e create_tables recebem por parâmetros, o cursor e a conexão criada anteriormente pela função create_database, em seguida elas percorrem o array/lista criado anteriormente (drop_table_queries e create_table_queries) e executa cada comando que criamos, para dropar as tabelas e em seguida para criá-las.<jupyter_code>main()<jupyter_output><empty_output><jupyter_text>Chamamos a função main e temos nosso banco de dados e tabelas criadas, os últimos comandos da função fecham o cursor e a conexão.Agora podemos começar a trabalhar no Pipeline para Extrair, transformar e inserir os dados nessas tabelas criadas!<jupyter_code># ETL PIPELINE

import os
import glob
import psycopg2
import pandas as pd

def main_etl():
    """ Function used to extract, transform all data from song_data and log_data and load it into a PostgreSQL DB
    """

    conn = psycopg2.connect("host=localhost dbname=sparkifydb user=ebraim password=<PASSWORD>")
    cur = conn.cursor()

    process_data(cur, conn, filepath="data/song_data", func=process_song_data)
    process_data(cur, conn, filepath="data/log_data", func=process_log_data)

    cur.close()
    conn.close()

def process_data(cur, conn, filepath, func):
    """Walks through all files nested under filepath, and processes all logs found.

    Parameters:
      cur (psycopg2.cursor()): Cursor of the sparkifydb database
      conn (psycopg2.connect()): Connection to the sparkifycdb database
      filepath (str): Filepath parent of the logs to be analyzed
      func (python function): Function to be used to process each log

    Returns:
      Name of files processed
    """

    # 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))
  
    # iterate over files and process
    for i, datafile in enumerate(all_files, 1):
        func(cur, datafile)
        conn.commit()

    return all_files<jupyter_output><empty_output><jupyter_text>Início do ETL pipelineNossa Primeira função chamada será a main_etl, ela será responsável por fazer a conexão com o banco de dados, chamar a função process_data para os dados na pasta song_data (que servirão para inserir os dados nas tabelas songs e artists) e posteriormente, chamar a mesma função process_data para os dados da pasta log_data (que servirão para inserir os dados nas tabelas time, users e songplays).A função process_data recebe como parâmetros o cursor, a conexão, o caminho da pasta onde estão os arquivos para percorrer e a função específica para fazer o ETL para esses dois conjuntos de dados.A função process_data começa declarando um array/lista que receberá os caminhos dos arquivos para serem percorridos futuramente, usamos um for loop para percorrer primeiramente todos os arquivos da pasta raiz passada como parâmetro, em seguida, unimos o caminho relativo com o formato .json e dentro do primeiro loop criado usamos outro for loop para adicionar cada um ao final do nosso array com a função append.Por último, percorremos esse array `all_files` chamando a função passada como parâmetro e passamos para frente o cursor para executar os comandos nas tabelas e caminho de cada arquivo da pasta que será lido com auxílio da pandas para fazer esse meio de campo. Após cada iteração, comitamos, e no final a função retornará o array com os caminhos de todos os arquivos!<jupyter_code>def process_song_data (cur, datafile):
    """ Reads songs log file row by row, selects needed fields and inserts them into song and artist tables.

    Parameters:
      cur (psycopg2.cursor()): Cursor of the sparkifydb database
      filepath (str): Filepath of the file to be analyzed
    """

    # open song file
    df = pd.read_json(datafile, lines=True)

    for value in df.values:
        num_songs, artist_id, artist_latitude, artist_longitude, artist_location, artist_name, song_id, title, duration, year = value

        # insert artist record
        artist_data = [artist_id, artist_name, artist_location, artist_latitude, artist_longitude]
        cur.execute(insert_artist, artist_data)

        song_data = [song_id, title, artist_id, year, duration]
        cur.execute(insert_song, song_data)<jupyter_output><empty_output><jupyter_text>Processando os dados da pasta song_dataChegamos na parte de processar os dados, transformar e inserir nas nossas tabelas do banco de dados.Lembrando que a função será chamada para cada arquivo na pasta (a função esta sendo chamada dentro de um for loop percorrendo a lista com todos os arquivos!Recebemos como parâmetros da função, o cursor e o arquivo do dado (o caminho do arquivo em formato json)Nosso primeiro trabalho é ler esse arquivo com o auxílio do pandas, e para isso, usamos a função read_json para ler esse arquivo e atribuimos à variavel df.Em seguida, vamos usar um for loop para percorrer os valores desse dataframe, para cada linha do dataframe, extrairemos cada registro e alocaremos em uma variavel, essas variaveis serao usadas para serem inseridas nas tabelas. Inserindo os dados na tabela dimensão artistsSeguindo a linha do raciocínio, definimos uma lista com os dados para a tabela artists, em seguida, executamos o cursor para inserir na tabela artists, a lista dos dados dos artistas criados agora pouco. Inserindo os dados na tabela dimensão songsEm seguida, separamos os dados das musicas numa outra lista e executamos o cursor novamente, dessa vez para inserir na tabela songs os dados dessa lista criada.<jupyter_code>def process_log_data (cur, datafile):
    """ Reads user activity log file row by row, filters by NextSong, selects needed fields, transforms them and inserts them into time, user and songplay tables.

    Parameters: 
      cur (psycopg2.cursor()): Cursor of the sparkifydb database
      filepath (str): Filepath of the file to be analyzed
    """

    # open log file
    df = pd.read_json(datafile, lines=True)

    # filter by NextSong action
    df = df[df['page'] == 'NextSong']

    # convert timestamp column to datetime
    t = pd.to_datetime(df['ts'], unit='ms')

    # insert time records
    time_data = []
    for line in t:
        time_data.append([line, line.hour, line.day, line.week, line.month, line.year, line.day_name()])
    column_labels = ('start_time', 'hour', 'day', 'week', 'month', 'year', 'weekday')
    time_df = pd.DataFrame(time_data, columns=column_labels)

    for i, row in time_df.iterrows():
        cur.execute(insert_time, 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(insert_user, list(row))

    # insert songplay records
    for i, 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, duration = results
        else:
            songid, artistid, duration = None, None, None

        # insert songplay record
        # songplay_id, start_time, user_id, level, song_id, artist_id, session_id, location, user_agent
        songplay_data = (i, pd.to_datetime(row.ts, unit='ms'), int(row.userId), row.level, songid, artistid, row.sessionId, row.location, row.userAgent)
        cur.execute(insert_songplays, songplay_data)<jupyter_output><empty_output><jupyter_text>Explicando a função process_log_dataA função `process_log_data` recebe dois parametros, o cursor e o caminho do arquivo json.Logo no início, lemos esse json com a função do pandas `read_json` e alocamos esse dataframe na variável `df`.Como o time de análise quer os dados apenas da página `NextSong`, filtramos nosso dataframe para conter apenas os registros dessa página. Inserindo os dados na tabela dimensão timePara nossa tabela dimensão `time` só precisaremos de uma coluna do dataframe inicial, selecionamos essa coluna com o comando `df['ts']` mas precisamos converter cada registro desse para datetime, por isso, chamamos a função do pandas `to_datetime` e definimos a unidade para milisegundos, atribuindo à variável `t`.Em seguida, iniciamos nossa lista que receberá os dados para cada campo da query para inserir os dados na tabela.Usamos um for loop para percorrer todos os registros da variável `t` e dentro desse loop, separamos as informações que nossa query `insert_time` precisa e adicionamos à nossa lista `time_data`.Com essa lista, faremos um dataframe, e para isso atribuimos os nomes das colunas em `column_labels` e logo depois criamos o dataframe `time_df`.Agora só precisamos percorrer todas as linhas desse dataframe com um for loop e executar a query `insert_time`, lembrando que precisamos usar afunção nativa `list()` para que cada linha do dataframe seja um array/list.Os dados da tabela dimensão tempo estão inseridos! Inserindo os dados na tabela dimensão usersEsse é simples, vamos selecionar as colunas do dataframe inicial `df` e atribuimos à variável `user_df`.Então usamos um for loop para percorrer cada linha e executar a query `insert_user`, passando cada linha do dataframe como um array/list. Inserindo os dados na tabela fato song_plays Primeiro precisamos encontrar o id da música e do artistaNo primeiro momento, vamos usar um for loop para percorrer cada linha do dataframe inicial.Em cada etapa desse for, vamos executar aquela query `song_select` passando o nome da música, o nome do artista e o tempo de duração para dentro da query, assim poderemos selecionar as mśsicas que tenha o nome da música, o nome do artista e o tempo de duração igual, para termos acesso ao id da música e ao id do artista, pois essas informações estão nas tabela dimensão `artists` e `songs`, e não nosso dataframe inicial, que não dispõe desses dados.Atribuimos cada resultado à variável results, em seguida usamos um `if`, se results for `True`, separamos os resultados do id da música, id do artista e duração da música. Se for `False` essas variáveis receberão `None`. Continuando para inserir os dados na tabela fato song_playsIniciamos uma variável `songplay_data` para receber os dados necessários para inserir os dados na tabela `song_plays`. Essa variável receberá, na ordem exata, os dados de cada linha do dataframe inicial para passar os valores para dentro da query `insert_songplays`, entre essas variáveis, as duas `songid` e `artistid` que selecionameos agora pouco.Em seguida, executamos o cursor para inserir esses dados separados na query `insert_songplays` e nossa tabela fato está alimentada.<jupyter_code>main_etl()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.0643615483124716, 'fraction_numerical': 0.0050811748667740735, 'mean_word_length': 4.437556154537287, 'pattern_counts': {'":': 28, '<': 40, '<?xml version=': 0, '>': 40, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '9', 'dir': 'jupyter-structured-clean-dedup', 'id': '183607', 'license': 'MIT', 'n_tokens_mistral': 7986, 'n_tokens_neox': 7695, 'n_words': 3128, 'path': '1_dend_data_modeling/P1_Postgres_Data_Modeling_and_ETL/sparkify-project.ipynb', 'repo_name': 'ebraimcarvalho/udacity-nanodegree-data-engineer', 'size': '24195'}