This project implements an end-to-end data pipeline that extracts cryptocurrency data from a public API, processes it, and stores the data in Google BigQuery for analysis. The pipeline is orchestrated using Apache Airflow (via Google Cloud Composer) to run daily, ensuring that the most recent data is available for visualization and reporting.
The data pipeline involves multiple stages:
-
Fetching Crypto Data:
- The pipeline fetches real-time cryptocurrency data from external APIs, specifically focusing on 5 selected cryptocurrencies: 'bitcoin', 'ethereum', 'tether', 'binancecoin', 'solana'.
- This data is retrieved in CSV format from the CoinGecko API and uploaded to a Google Cloud Storage bucket for storage.
-
Data Transformation:
- The fetched data undergoes transformation and is prepared for loading into Google BigQuery.
- The transformation process includes cleaning, reformatting, and filtering unnecessary data to make it suitable for analysis.
-
Loading Data to BigQuery:
- Once the data is cleaned and transformed, it is loaded into a BigQuery table.
- The data is partitioned by date, and the
datecolumn in the data is used for partitioning. - Data is loaded into a table in BigQuery, where each row represents a cryptocurrency's market data at a given time.
-
Triggering BigQuery Stored Procedure:
- After loading data into BigQuery, a stored procedure is triggered in BigQuery for further data analysis. This stored procedure can be used for performing aggregations, generating insights, or running advanced analytics on the loaded data.
Airflow is used to orchestrate the entire pipeline, allowing you to define, schedule, and monitor each step of the data processing workflow. The pipeline is built with PythonOperators in Airflow, and the following steps are executed in sequence:
-
Start Task (
start):- This task logs the start timestamp for the pipeline execution.
-
Fetch Crypto Data Task (
fetch_and_upload_data_gcs_task):- This task calls the function to fetch cryptocurrency data and uploads it to Google Cloud Storage.
-
Load Data to BigQuery Task (
load_data_bq_task):- This task loads the cleaned data from Google Cloud Storage to Google BigQuery.
-
Trigger BigQuery Procedure Task (
trigger_procedure_task):- This task triggers the BigQuery procedure after data is loaded, allowing for further analysis.
-
End Task (
end):- This task logs the end timestamp, marking the completion of the pipeline.
Google Cloud Storage (GCS) is used as a staging area for storing the fetched cryptocurrency data before it is loaded into BigQuery. GCS allows for easy storage and access of large datasets, making it an ideal choice for this pipeline.
The pipeline fetches cryptocurrency data in real-time and stores it in a GCS bucket. From there, it is processed and loaded into BigQuery.
Here’s a snapshot of the data loaded into Google BigQuery after running the pipeline:
The database consists of the following tables:
crypto_data_temp: Stores raw cryptocurrency data for each coin in USD, INR, and EUR.final_data: Stores processed data with details for each cryptocurrency in different currencies.market_metrics: Stores total market cap data for each currency.volume_metrics: Stores 24-hour trading volume for each currency.
These tables are linked by the cryptocurrency coin and partitioned by date to enable efficient querying and data analysis.
Looker is used to visualize and analyze the processed cryptocurrency data. With its interactive dashboards and powerful query-building capabilities, Looker enables detailed insights into the performance and trends of selected cryptocurrencies.
-
Overall Market Trends:
- A dashboard showcasing real-time market cap, 24-hour volume, and price changes for the 5 cryptocurrencies.
-
Performance Comparison:
- A comparative analysis of price changes and market trends for 'bitcoin', 'ethereum', 'tether', 'binancecoin', and 'solana'.
With these visualizations, users can easily track cryptocurrency performance over time and make informed decisions based on up-to-date market data.
Before starting, ensure you have set up the following on your Google Cloud account:
- A GCS bucket for storing cryptocurrency data.
- A BigQuery dataset and table for loading and analyzing data.
- Ensure the necessary IAM roles and permissions are granted for interacting with GCS and BigQuery.