Doc Datalake Format Explorer

Descrição

O projeto consiste em coletar dados de uma API do Spotify, transformar os dados em diferentes formatos e armazenar em um bucket na AWS.

O objetivo final é fazer uma análise sobre os 3 tipos de arquivo gerado, CSV, Parquet e JSON, e verificar qual o melhor formato para armazenar os dados.

A análise final é salva em um arquivo do tipo LOG localmente

Fluxo

flowchart LR
    subgraph pipeline[Pipeline]
        A[Manager Spotify: Coleta de Dados] -->|Configura API Spotify| B(Handler HTTPX: Realizada Chamada async)
        B[Handler HTTPX: Realizada Chamada async] -->|Extrai dados| C(Manager Pandas: Converte Formatos)
        F[Pipeline: Analise] --> |Gera um analise.LOG| D2[Local]
        C -->|Converte p/ CSV| D[Load: Salva no Bucket]
        C -->|Converte p/ Parquet| D[Manager AWS: Salva no Bucket]
        C -->|Converte p/ JSON| D[Manager AWS: Salva no Bucket]
        D -->|Salva como CSV| E1[Bucket AWS]
        D -->|Salva como Parquet| E1[Bucket AWS]
        D -->|Salva como JSON| E1[Bucket AWS]
    end

Módulo Handler HTTP

Esse módulo foi criado depois para "desempacotar" o httpx/requests da aplicação, assim, caso seja necessário trocar a biblioteca de requisições, basta alterar o módulo.

Classe HTTPRequest

Bases: BaseModel

É uma classe que representa uma solicitação HTTP genérica.

Parameters:

Name	Type	Description	Default
url	str	A URL da solicitação.	required
method	str	O método HTTP a ser usado (padrão é "GET").	required
headers	dict	Um dicionário de cabeçalhos HTTP personalizados.	required
params	dict	Um dicionário de parâmetros de consulta.	required
data	dict	Um dicionário de dados a serem enviados na solicitação.	required

Source code in app/ETL/handler_http.py

class HTTPRequest(BaseModel):
    """
    É uma classe que representa uma solicitação HTTP genérica.

    Args:
        url (str): A URL da solicitação.
        method (str): O método HTTP a ser usado (padrão é "GET").
        headers (dict): Um dicionário de cabeçalhos HTTP personalizados.
        params (dict): Um dicionário de parâmetros de consulta.
        data (dict): Um dicionário de dados a serem enviados na solicitação.
    """

    url: str
    method: HTTPMETHODS = "GET"
    headers: Dict = None
    params: Dict = None
    data: Dict = None

Faz uma solicitação HTTP genérica.

Parameters:

Name	Type	Description	Default
http_request	HTTPRequest	Um objeto HTTPRequest contendo os detalhes da solicitação.	required

Returns:

Name	Type	Description
dict	dict	A resposta JSON da solicitação HTTP.

Source code in app/ETL/handler_http.py

async def make_http_request(HTTP_request: HTTPRequest) -> dict:
    """
    Faz uma solicitação HTTP genérica.

    Args:
        http_request (HTTPRequest): Um objeto HTTPRequest contendo os detalhes da solicitação.

    Returns:
        dict: A resposta JSON da solicitação HTTP.
    """
    async with httpx.AsyncClient() as client:
        response = await client.request(
            HTTP_request.method,
            HTTP_request.url,
            headers=HTTP_request.headers,
            params=HTTP_request.params,
            data=HTTP_request.data,
        )
        response.raise_for_status()
        return response.json()

Módulos do manager_spotify

Esse módulo foi criado para encapsular as chamadas a API do Spotify, assim, caso seja necessário trocar o dominio, basta alterar o módulo.

Obtém o access token do Spotify.

Returns:

Name	Type	Description
str	str	Access token.

Source code in app/ETL/manager_spotify.py

async def get_spotify_access_token() -> str:
    """
    Obtém o access token do Spotify.

    Returns:
        str: Access token.
    """
    AUTH_URL = "https://accounts.spotify.com/api/token"
    auth_header = base64.b64encode(
        f"{SPOTIFY_CLIENT_ID}:{SPOTIFY_CLIENT_SECRET}".encode("utf-8")
    ).decode("utf-8")
    headers = {"Authorization": f"Basic {auth_header}"}
    data = {"grant_type": "client_credentials"}

    # Faz a solicitação usando o http_handler
    response = await make_http_request(
        HTTPRequest(url=AUTH_URL, method=HTTPMETHODS.POST, headers=headers, data=data)
    )
    token_response_data = response
    print(f"[INFO] Access token: {token_response_data['access_token']}")
    return token_response_data["access_token"]

Procura por um artista no Spotify e retorna o ID do primeiro resultado.

Parameters:

Name	Type	Description	Default
name	str	Nome do artista.	required
token	str	Access token.	required

Returns:

Name	Type	Description
str	str	ID do artista.

Source code in app/ETL/manager_spotify.py

async def search_artist(name: str, token: str) -> str:
    """
    Procura por um artista no Spotify e retorna o ID do primeiro resultado.

    Args:
        name (str): Nome do artista.
        token (str): Access token.

    Returns:
        str: ID do artista.
    """
    BASE_URL = "https://api.spotify.com/v1/search"
    headers = {"Authorization": f"Bearer {token}"}
    params = {"q": name, "type": "artist", "limit": 1}

    # Faz a solicitação usando o http_handler
    response = await make_http_request(
        HTTPRequest(
            url=BASE_URL, method=HTTPMETHODS.GET, headers=headers, params=params
        )
    )
    data = response
    return data["artists"]["items"][0]["id"]

Retorna um objeto do tipo ArtistDetails

Bases: BaseModel

Modelo para os detalhes de um artista.

Source code in app/models/spotify_models.py

class ArtistDetails(BaseModel):
    """
    Modelo para os detalhes de um artista.
    """

    id: str
    name: str
    albums: List[AlbumDetails]

Obtém os álbuns de um artista.

Parameters:

Name	Type	Description	Default
artist_id	str	ID do artista.	required
token	str	Access token.	required

Returns:

Type	Description
List[AlbumDetails]	List[AlbumDetails]: Lista de objetos AlbumDetails.

Source code in app/ETL/manager_spotify.py

async def get_artist_albums(artist_id: str, token: str) -> List[AlbumDetails]:
    """
    Obtém os álbuns de um artista.

    Args:
        artist_id (str): ID do artista.
        token (str): Access token.

    Returns:
        List[AlbumDetails]: Lista de objetos AlbumDetails.
    """
    BASE_URL = f"https://api.spotify.com/v1/artists/{artist_id}/albums"
    headers = {"Authorization": f"Bearer {token}"}

    # Faz a solicitação usando o http_handler
    response = await make_http_request(
        HTTPRequest(url=BASE_URL, method=HTTPMETHODS.GET, headers=headers)
    )
    albums_data = response["items"]
    return [
        AlbumDetails(
            id=album["id"],
            name=album["name"],
            release_date=album.get("release_date", None),
            total_tracks=album["total_tracks"],
        )
        for album in albums_data
    ]

Retorna um objeto do tipo TrackDetails

Bases: BaseModel

Modelo para os detalhes de uma faixa.

Source code in app/models/spotify_models.py

class TrackDetails(BaseModel):
    """
    Modelo para os detalhes de uma faixa.
    """

    name: str
    duration_ms: int
    release_date: str
    popularity: int
    track_ids: list  # Lista de IDs das faixas

Obtém as músicas de um álbum.

Parameters:

Name	Type	Description	Default
album_id	str	ID do álbum.	required
token	str	Access token.	required

Returns:

Type	Description
List[TrackDetails]	List[TrackDetails]: Lista de objetos TrackDetails.

Source code in app/ETL/manager_spotify.py

async def get_album_tracks(album_id: str, token: str) -> List[TrackDetails]:
    """
    Obtém as músicas de um álbum.

    Args:
        album_id (str): ID do álbum.
        token (str): Access token.

    Returns:
        List[TrackDetails]: Lista de objetos TrackDetails.
    """
    BASE_URL = f"https://api.spotify.com/v1/albums/{album_id}/tracks"
    headers = {"Authorization": f"Bearer {token}"}

    # Faz a solicitação usando o http_handler
    response = await make_http_request(
        HTTPRequest(url=BASE_URL, method=HTTPMETHODS.GET, headers=headers)
    )
    tracks_data = response["items"]
    return [
        TrackDetails(
            id=track["id"],
            name=track["name"],
            duration_ms=track["duration_ms"],
            release_date=track.get("release_date", "N/A"),
            popularity=track.get("popularity", 0),
            track_ids=[],
        )
        for track in tracks_data
    ]

Retorna um objeto do tipo AlbumDetails

Bases: BaseModel

Modelo para os detalhes de um álbum.

Source code in app/models/spotify_models.py

class AlbumDetails(BaseModel):
    """
    Modelo para os detalhes de um álbum.
    """

    id: str
    name: str
    release_date: Optional[str]
    total_tracks: int
    tracks: Optional[List[TrackDetails]] = []

Obtém todos os detalhes de um artista.

Parameters:

Name	Type	Description	Default
artist_name	str	Nome do artista.	required

Returns:

Type	Description
List[CompleteArtistDetails]	List[CompleteArtistDetails]: Lista de objetos CompleteArtistDetails.

Source code in app/ETL/manager_spotify.py

async def get_all_artist_details(artist_name: str) -> List[CompleteArtistDetails]:
    """
    Obtém todos os detalhes de um artista.

    Args:
        artist_name (str): Nome do artista.

    Returns:
        List[CompleteArtistDetails]: Lista de objetos CompleteArtistDetails.
    """
    token = await get_spotify_access_token()
    artist_id = await search_artist(artist_name, token)
    albums = await get_artist_albums(artist_id, token)
    all_tracks = []

    for album in albums:
        tracks = await get_album_tracks(album.id, token)
        for track in tracks:
            all_tracks.append(
                CompleteArtistDetails(
                    artist=artist_name,
                    artist_id=artist_id,
                    album=album.name,
                    track=track.name,
                    minutes=track.duration_ms / 60000,  # Convert ms to minutes
                    release_date=track.release_date,
                    popularity=track.popularity,
                )
            )
    return all_tracks

Retorna um objeto do tipo CompleteArtistDetails

Bases: BaseModel

Modelo para os detalhes completos de um artista.

Source code in app/models/spotify_models.py

class CompleteArtistDetails(BaseModel):
    """
    Modelo para os detalhes completos de um artista.
    """

    artist: str
    artist_id: str
    album: str
    track: str
    minutes: float
    release_date: str
    popularity: int

Módulos do manager_polars

Esse módulo foi criado para encapsular as chamadas a API do Pandas e Polars, assim, caso seja necessário trocar a biblioteca de manipulação de dados, basta alterar o módulo.

Salva os dados em diferentes formatos.

Parameters:

Name	Type	Description	Default
data	List[CompleteArtistDetails]	Lista de objetos CompleteArtistDetails.	required
artist_name	str	Nome do artista.	required

Source code in app/ETL/manager_polars.py

def save_data_to_formats(data: List[CompleteArtistDetails], artist_name: str):
    """
    Salva os dados em diferentes formatos.

    Args:
        data (List[CompleteArtistDetails]): Lista de objetos CompleteArtistDetails.
        artist_name (str): Nome do artista.
    """
    # Converte os dados (que esperamos ser uma lista de objetos CompleteArtistDetails) em um DataFrame do Polars
    df = pl.DataFrame([item.model_dump() for item in data])

    # Substitui espaços em branco por hífens e converte para minúsculas
    artist_name = artist_name.replace(" ", "-").lower()

    # Salva em CSV
    df.write_csv(f"data/{artist_name}-data.csv")

    # Salva em JSON
    df.write_json(f"data/{artist_name}-data.json")

    # Salva em Parquet
    df.write_parquet(f"data/{artist_name}-data.parquet")

    print(f"[INFO] Data for {artist_name} saved in multiple formats!")

Módulos do manager_aws

Esse módulo foi criado para encapsular as chamadas a API do AWS, assim, caso seja necessário trocar o provedor de armazenamento, basta alterar o módulo.

Cria um bucket no S3.

Parameters:

Name	Type	Description	Default
bucket_name	str	Nome do bucket.	required

Source code in app/ETL/manager_aws.py

def create_bucket(bucket_name: str):
    """
    Cria um bucket no S3.

    Args:
        bucket_name (str): Nome do bucket.
    """
    try:
        s3.create_bucket(Bucket=bucket_name)
        print(f"[INFO] Bucket {bucket_name} created successfully!")
    except Exception as e:
        print(f"[ERROR] Failed to create bucket. Reason: {e}")

Faz upload de um arquivo para o S3.

Parameters:

Name	Type	Description	Default
file_name	str	Nome do arquivo.	required
bucket_name	str	Nome do bucket.	required

Source code in app/ETL/manager_aws.py

def upload_to_s3(file_name: str, bucket_name: str):
    """
    Faz upload de um arquivo para o S3.

    Args:
        file_name (str): Nome do arquivo.
        bucket_name (str): Nome do bucket.
    """
    try:
        s3.upload_file(file_name, bucket_name, file_name)
        print(f"[INFO] {file_name} uploaded successfully to {bucket_name}!")
    except Exception as e:
        print(f"[ERROR] Failed to upload {file_name}. Reason: {e}")

Módulos de análise

Esse módulo foi criado para fazer a análise final dos tipos de arquivo

Analisa e compara o tamanho dos arquivos gerados em diferentes formatos. Recomenda o formato com o menor volume total.

Source code in app/pipeline.py

async def analyze_file_sizes(artist_name: str):
    """
    Analisa e compara o tamanho dos arquivos gerados em diferentes formatos.
    Recomenda o formato com o menor volume total.
    """
    # Diretório onde os arquivos estão
    data_dir = "data"

    # Nomes dos arquivos
    csv_file = f"{artist_name.lower().replace(' ', '-')}-data.csv"
    json_file = f"{artist_name.lower().replace(' ', '-')}-data.json"
    parquet_file = f"{artist_name.lower().replace(' ', '-')}-data.parquet"

    # Caminhos completos dos arquivos
    csv_path = os.path.join(data_dir, csv_file)
    json_path = os.path.join(data_dir, json_file)
    parquet_path = os.path.join(data_dir, parquet_file)

    # Obtém os tamanhos dos arquivos
    csv_size = os.path.getsize(csv_path)
    json_size = os.path.getsize(json_path)
    parquet_size = os.path.getsize(parquet_path)

    # Calcula as somas dos tamanhos
    total_csv_size = csv_size
    total_json_size = json_size
    total_parquet_size = parquet_size

    # Determina o formato com o menor volume total
    smallest_format = "CSV"
    smallest_size = total_csv_size

    if total_json_size < smallest_size:
        smallest_format = "JSON"
        smallest_size = total_json_size

    if total_parquet_size < smallest_size:
        smallest_format = "Parquet"
        smallest_size = total_parquet_size

    # Imprime os resultados
    logger.info(f"Análise de tamanho dos arquivos para {artist_name}:")
    logger.info(f"Tamanho total do arquivo CSV: {total_csv_size} bytes")
    logger.info(f"Tamanho total do arquivo JSON: {total_json_size} bytes")
    logger.info(f"Tamanho total do arquivo Parquet: {total_parquet_size} bytes")
    logger.info(
        f"Recomendação: Use o formato {smallest_format} com o menor volume total."
    )

    # Retorna o formato recomendado
    return smallest_format