Spotify data pipeline ingestion

git clone https://gitlab.com/lelouvincx/fde01_project_fde220103_dinhminhchinh.git
mv fde01_project_fde220103_dinhminhchinh project
cd project

# Create env file
touch env
cp env.template env

# Setup python packages
make install

# Build docker images
make build

# Run locally
make up

# Check running containers
docker ps

# Check code quality
make check
make lint

# Use black to reformat if any tests failed, then try again
black ./elt_pipeline

# Test coverage
make test

# Enter mysql cli
make to_mysql

SET GLOBAL local_infile=true;
-- Check if local_infile is turned on
SHOW VARIABLES LIKE "local_infile";
exit

# Create tables with schema
make mysql_create
# Load csv into created tables
make mysql_load
# Set their foreign keys
make mysql_set_foreign_key

# Enter psql cli
make psql_create

# Test utils
python3 -m pytest -vv --cov=utils elt_pipeline/tests/utils
# Test ops
python3 -m pytest -vv --cov=ops elt_pipeline/tests/ops

cd terraform

# Initialize infra
make tf-init

# Checkout planned infra
make tf-plan

# Build up infra
make tf-up

# Connect to EC2 from local terminal
make ssh-ec2
# Generate new ssh key for gitlab
ssh-keygen
# Then press Enter until done
cat ~/.ssh/id_rsa.pub

def extract_data(self, sql: str) -> pd.DataFrame:
    pd_data = None
    with self.get_db_connection() as db_conn:
        pd_data = pd.read_sql(sql, db_conn)
    return pd_data

def get_recently(self, number: int, token: str) -> (int, dict):
    headers = {
        "Accept": "application/json",
        "Content-Type": "application/json",
        "Authorization": f"Bearer " + token,
    }
    params = [("limit", number),]
    try:
        response = requests.get(
            "https://api.spotify.com/v1/me/player/recently-played",
            headers=headers,
            params=params,
            timeout=10,
        )
        return (response.status_code, response.json())
    except:
        return None

def extract_data(self, token: str) -> pd.DataFrame:
    (code, content) = self.get_recently(50, token)
    my_tracks = {
        "album_id": [], "artists_id": [], "track_id": [], "track_unique_id": [], "name": [], "popularity": [], "type": [], "duration_ms": [], "played_at": [], "danceability": [], "energy": [], "track_key": [], "loudness": [], "mode": [], "speechiness": [], "acousticness": [], "instrumentalness": [], "liveness": [], "valence": [], "tempo": [],
    }

    items = content.get("items", [])
    for item in items:
        # Take album_id, artists_id, track_id, name, popularity, type, duration_ms
        played_at = item.get("played_at", [])
        track = item.get("track", [])
        album = track.get("album", [])
        album_id = album.get("id", [])
        artists = track.get("artists", [])
        artists_id = []
        for artist in artists:
            artists_id.append(artist.get("id", []))
        track_id = track.get("id", [])
        name = track.get("name", [])
        popularity = track.get("popularity", [])
        type = track.get("type", [])
        duration_ms = track.get("duration_ms", [])

        # Take features
        features = self.get_features(track_id, token)
        danceability = features.get("danceability", [])
        energy = features.get("energy", [])
        track_key = features.get("key", [])
        loudness = features.get("loudness", [])
        mode = features.get("mode", [])
        speechiness = features.get("speechiness", [])
        acousticness = features.get("acousticness", [])
        instrumentalness = features.get("instrumentalness", [])
        liveness = features.get("liveness", [])
        valence = features.get("valence", [])
        tempo = features.get("tempo", [])

        # Extract row into dict
        my_tracks["album_id"].append(album_id)
        my_tracks["artists_id"].append(artists_id)
        my_tracks["track_id"].append(track_id)
        my_tracks["track_unique_id"].append(track_id + played_at)
        my_tracks["name"].append(name)
        my_tracks["popularity"].append(popularity)
        my_tracks["type"].append(type)
        my_tracks["duration_ms"].append(duration_ms)
        my_tracks["played_at"].append(played_at[:10])
        my_tracks["danceability"].append(danceability)
        my_tracks["energy"].append(energy)
        my_tracks["track_key"].append(track_key)
        my_tracks["loudness"].append(loudness)
        my_tracks["mode"].append(mode)
        my_tracks["speechiness"].append(speechiness)
        my_tracks["acousticness"].append(acousticness)
        my_tracks["instrumentalness"].append(instrumentalness)
        my_tracks["liveness"].append(liveness)
        my_tracks["valence"].append(valence)
        my_tracks["tempo"].append(tempo)

    pd_data = pd.DataFrame(my_tracks)
    return pd_data

def extract_data_from_mysql(context, run_config):
    updated_at = context.op_config.get("updated_at")
    context.log.info(f"Updated at: {updated_at}")
    if updated_at is None or updated_at == "":
        context.log.info("Nothing to do!")
        return None
    context.log.info(f"Op extracts data from MySQL at {updated_at}")

    # Choose extract strategy (default: full load)
    sql_stm = f"""
        SELECT *
        FROM {run_config.get('src_tbl')}
        WHERE 1=1
    """
    if run_config.get("strategy") == "incremental_by_partition":
        if updated_at != "init_dump":
            sql_stm += f""" AND CAST({run_config.get('partition')} AS DATE) = '{updated_at}' """

    if run_config.get("strategy") == "incremental_by_watermark":
        data_loader = get_data_loader(
            run_config.get("db_provider"), run_config.get("target_db_params")
        )
        watermark = data_loader.get_watermark(
            f"{run_config.get('target_schema')}.{run_config.get('target_tbl')}",
            run_config.get("watermark"),
        )
        watermark = (
            updated_at if watermark is None or watermark > updated_at else watermark
        )
        if updated_at != "init_dump":
            sql_stm += f""" AND {run_config.get('watermark')} >= '{watermark}' """

    context.log.info(f"Extracting with SQL: {sql_stm}")
    db_loader = MysqlLoader(run_config.get("src_db_params"))
    pd_data = db_loader.extract_data(sql_stm)
    context.log.info(f"Data extracted successfully with shape: {pd_data.shape}")

    # Update params
    run_config.update(
        {
            "updated_at": updated_at,
            "data": pd_data,
            "s3_path": f"bronze/{run_config.get('data_source')}/{run_config.get('ls_target').get('target_tbl')}",
            "load_dtypes": run_config.get("load_dtypes"),
        }
    )

    return run_config

def extract_data_from_api(context, run_config):
    updated_at = context.op_config.get("updated_at")
    context.log.info(f"Updated at: {updated_at}")
    if updated_at is None or updated_at == "":
        context.log.info("Nothing to do!")
        return None
    context.log.info(f"Op extracts data from API at {updated_at}")

    # Extract strategy (only support incremental_by_partition)
    context.log.info(f"Extracting on date: {updated_at}")
    api_loader = ApiLoader(run_config.get("src_api_params"))
    token = api_loader.get_api_token()
    pd_data = api_loader.extract_data(token)
    index_played_at = pd_data[pd_data["played_at"] != updated_at].index  # Drop data
    pd_data.drop(index_played_at, inplace=True)
    context.log.info(
        f"Data loaded and filtered successfully with shape: {pd_data.shape}"
    )

    run_config.update(
        {
            "updated_at": updated_at,
            "data": pd_data,
            "s3_path": f"bronze/{run_config.get('data_source')}/{run_config.get('ls_target').get('target_tbl')}",
            "load_dtypes": run_config.get("load_dtypes"),
        }
    )

    return run_config

def load_data_to_s3(context, upstream):
    if upstream is None:
        return None

    updated_at = upstream.get("updated_at")
    s3_bucket = os.getenv("DATALAKE_BUCKET")
    if type(updated_at) == list:
        updated_at = max(updated_at)
    s3_file = f"s3://{s3_bucket}/{upstream.get('s3_path')}/updated_at={updated_at}"
    context.log.info(f"Loading data to S3: {s3_file}")

    # Load data to S3
    pd_data = upstream.get("data")

    # Preprocess data
    load_dtypes = upstream.get("load_dtypes")
    try:
        for col, data_type in load_dtypes.items():
            if data_type == "str":
                pd_data[col] = pd_data[col].fillna("")
                pd_data[col] = pd_data[col].astype(str)
                pd_data[col] = pd_data[col].str.strip()
                pd_data[col] = pd_data[col].str.rstrip()
                pd_data[col] = pd_data[col].str.replace("'", "")
                pd_data[col] = pd_data[col].str.replace('"', "")
                pd_data[col] = pd_data[col].str.replace(r"\n", "", regex=True)
            elif data_type == "int":
                cur_bit = np.log2(pd_data[col].max())
                if cur_bit > 32:
                    pd_data[col] = pd_data[col].astype({col: "int64"})
                elif cur_bit > 16:
                    pd_data[col] = pd_data[col].astype({col: "int32"})
                elif cur_bit > 8:
                    pd_data[col] = pd_data[col].astype({col: "int16"})
                else:
                    pd_data[col] = pd_data[col].astype({col: "int8"})
            elif data_type == "float":
                pd_data[col] = pd_data[col].astype({col: "float32"})
        context.log.info(f"Data preprocessed successfully")
    except Exception as e:
        context.log.info(f"Exception: {e}")

    # Write parquet object to S3
    pa_data = pa.Table.from_pandas(df=pd_data, preserve_index=False)
    pq.write_table(pa_data, s3_file)
    context.log.info("Data loaded successfully to S3")

    # Update stream
    upstream.update({"s3_bucket": s3_bucket, "s3_file": s3_file})

    return upstream

def load_data_to_psql(context, upstream):
    if upstream is None:
        return None

    # Load data to target
    context.log.info("Loading data to postgreSQL")
    context.log.info(f"Extracting data from {upstream.get('s3_file')}")
    pd_stag = pd.read_parquet(upstream.get("s3_file"))
    context.log.info(f"Extracted data shape: {pd_stag.shape}")

    if len(pd_stag) == 0:
        context.log.info("No data to upload!")
        return "No data to upload!"

    # Execute
    db_loader = PsqlLoader(upstream.get("target_db_params"))
    result = db_loader.load_data(pd_stag, upstream)
    context.log.info(f"Batch inserted status: {result}")
    return result

{{ config(materialized='table') }}

with my_unique_tracks AS (

  select *
  from spotify.my_tracks
  group by track_unique_id

)

select * from my_unique_tracks

{{ config(materialized='table') }}

with unique_tracks AS (

  select acousticness, album_id, artists_id, country, danceability, duration_ms, energy, track_id, instrumentalness, track_key, liveness, loudness, mode, name, popularity, speechiness, tempo, valence
  from spotify.spotify_tracks
  group by track_id

)

select * from unique_tracks

{{ config(materialized='table') }}

with unique_artists AS (

  select artist_popularity, followers, genres, artist_id, name, track_id
  from spotify.spotify_artists
  group by artist_id

)

select * from unique_artists

{{ config(materialized='table') }}

with unique_albums AS (

  select album_id, artist_id, album_type, name, release_date, release_date_precision, total_tracks, track_id
  from spotify.spotify_albums
  group by album_id

)

select * from unique_albums

# Tear down containers
make down

# Tear down AWS
cd terraform
make tf-down