All-Time Spotify Wrapped

1-1. Objective

The main goal of this project is to extract all of my song listening history data and continuously extract the Spotify usage activity through Spotify API. I will deep dive into my complete song listening history to analyze top artists, tracks, playlists, and listening patterns.

1-2. Key Features

• Real-time Tracking: Hourly API calls during active hours (0-6, 14-23 UTC)
• Historical Analysis: Integrates 2+ years of extended streaming history (89K+ plays)
• Usage Pattern Discovery: Discover my usage patterns and habits over time
• Data Lineage: Complete automation of column-level lineage tracking with OpenLineage + Marquez
• Cost Efficient: 100% open-source tools, hosted locally

2-2. Data Flow

Below is the holistic view of the data pipeline orchestrated by Apache Airflow:

The main components of the data pipeline are Extraction, Loading, Transformation, Lineage Tracking, Quality & Documentation, and Slack Notifications.

1. Extraction

The extraction process has two main parts: API Extraction and Extended History.

• API Extraction is responsible for fetching the recently played tracks and their audio features. The limitation of this API extraction is that it only returns 50 most recent tracks, and this is where the extended history comes into play.
• Extended History is responsible for fetching the historical streaming data from Spotify. Unlike the API extraction, this I need to manually request the extended history data from Spotify. Once I receive the data in JSON format, I need to parse the data and store it as a CSV file.

2. Loading

After ingesting the data from the extraction process, they are loaded into DuckDB for further analysis.

One issue I'm facing right now is that with multiple connections to the same DuckDB database, the database file is getting locked and I have to restart the docker container every time to unlock the database. My highest priority right now is to resolve this issue. The major approach I'm trying is to change the database configuration in Metabase to establish a read-only connection, but this doesn't seems to resolve the issue. Other approaches I'm considering is to set up a retry mechanism to set an exponential backoff delay for the connection to the database, but so far, the only temporary resolution is to restart the docker container.

3. Transformation (dbt)

dbt is used to transform the loaded data into a star schema for analysis. What I focused here was to have a proper dimensional modeling implemented and have a medallion structure. The tables are organized into three different layers: Staging Layer, Marts Layer, and Analytics Layer. Tables in staging layer is essentially silver tables where the raw data is loaded and cleaned up. The gold tables are in the mart layer, having fact tables and dimension tables from the silver tables. The analytics layer is the layer where different tables for analytics are created.

4. Lineage Tracking

I was able to automate end-to-end column level lineage tracking for all dbt models using OpenLineage. One challenge I faced here was integrating OpenLineage with dbt. The official documentation suggests that running dbt-ol run instead of dbt run automatically emits column level lineage. However, running the command gave me an error saying that dbt-ol does not support duckdb, so as a woraround, I had to make a separate script in python to manually emit the lineage events. Later, I found out that I wasn't downloading the correct version of the package openlineage-dbt, so configuring the exact version of the package resolved the issue, and I was able to automate column level lineage of all 16 dbt models.

5. Quality & Documentation

For the dbt models, I wanted to test out the dbt test and documentation features, so I added some light tests and documentation to the models for the data governance purposes. For the next step, I want to implement more comprehensive data quality checks that cross reference different columns and tables to ensure the data is accurate and consistent. I want to also be able to create a data quality dashboard if I can get the data quality testing data from dbt.

6. Slack Notifications

To ensure that I am notified for any pipeline failures or retries, I integrated Slack into the pipeline to send notifications to my Slack workspace. The message tells me the task that failed, the error message, and the retry count. I thought this was a good way to ensure even when I am out of home to check on the pipeline, I can still be notified and take action accordingly.