Loading repository data…
chayansraj / repository
This project demonstrates how to build and automate an ETL pipeline using DAGs in Airflow and load the transformed data to Bigquery. There are different tools that have been used in this project such as Astro, DBT, GCP, Airflow, Metabase.
This project demonstrates how to build and automate data pipeline using DAGs in Airflow and load the transformed data to Bigquery. There are different tools that have been used in this project such as Astro (A docker wrapper around Airflow), DBT (Used for Data Modelling and creating reports using SQL), Soda (Used for Data Quality Checks), Metabase (Containarized Data visualization tool) and Google Cloud Platform for storing the tables.
There has been a deluge of new tools and technologies in the market accentuating the data engineering field and the best way to keep abreast is to pry them out and do hands-on. So, in this project there are a lot of new tools that I have used to create an effective and robust data pipeline. It goes without saying that when there are more tools involved, the integration of the tools (because of dependencies or much worse "dependency hell" :scream: ) becomes a big hassle but at the same time, it allows us to understand the nitty gritties of the architecture.
Data Ingestion - Create a data ingestion pipeline to extract new incoming data into the GCP BigQuery.
Data Quality - Use Soda to create highly customized data quality checks using yml files.
Data Transformation - Use DBT to perform data modeling and transform the data into star schema.
Data Loading - Use data pipeline to load the extracted and transformed data into GCP BigQuery.
Data Reporting/Analytics - Use Metabase to create dashboards for reporting or analytics purposes.
The architecture (Data flow) used in this project uses different different tools and languages.
The Online Retail II data set contains all the transactions occurring for a UK-based and registered, non-store online retail between 01/12/2009 and 09/12/2011.The company mainly sells unique all-occasion gift-ware. Many customers of the company are wholesalers.
Dataset link: Online Retail Dataset
astro dev init
This command will create a structure of airflow components in your directory, that we used to manually create files earlier. The project foler structure is as shown below:
After this step, we need to create an account in google cloud with a free tier. Then create a new project with a name of your choice. Make sure to note the project ID and keep it handy as we will use it many different places later on. To let google cloud interact with AIrflow, we need a way to establish a connection and that is done by something called "Service Accounts" in GCP under IAM settings. The connection is between google cloud storage and airflow, so create a storage admin account and a BigQuery admin account in service account settings. Download the key and put it in your 'gcp' folder under 'include' folder. Do not share it with anyone
A connection is a two-way street
Just like google needs to connect to Airflow, same goes to the other side. In Airflow, there is a big community of "providers", think of it as a package that we can install on top of Airflow core to create more functionalities and interact with different softwares. So, in our case, we will install Airflow google provider
apache-airflow-providers-google
After adding the above command in requirements.txt file, we have to start the docker container which can be done as shown below:
astro dev start
If everything is good, you can access the airflow UI at localhost:8080 where we will create a GCP connection using the service account credentials, in the airflow include directory as shown below:
Step 2 - Create the data pipeline using DAGs in Airflow A DAG (Directed Acyclic Graph) is a core concept and a fundamental building block. It represents a workflow or a sequence of tasks to be executed, where each task is a unit of work, and the edges between tasks define the order in which they should be executed.
from airflow.decorators import (
dag,
task,)
# DAG and task decorators for interfacing with the TaskFlow API
from datetime import datetime
from airflow.providers.google.cloud.transfers.local_to_gcs import LocalFilesystemToGCSOperator
from airflow.providers.google.cloud.operators.bigquery import BigQueryCreateEmptyDatasetOperator
upload_csv_to_gcs = LocalFilesystemToGCSOperator(
task_id='upload_csv_to_gcs',
src='/usr/local/airflow/include/dataset/online_retail.csv',
dst='raw/online_retail.csv',
bucket='online_retail_database',
gcp_conn_id='gcp',
mime_type='text/csv')
astro dev bash -> airflow tasks test <DAG name> <Task Name> <Start Date>
This should store a file in google cloud storage bucket and marke the task as success with corresponding dag id and task id as shown below:
create_retail_dataset = BigQueryCreateEmptyDatasetOperator(
task_id='create_retail_dataset',
dataset_id='retail',
gcp_conn_id='gcp',)
gcs_to_raw = aql.load_file(
task_id='gcs_to_raw',
input_file=File(
'gs://online_retail_database/raw/online_retail.csv',
conn_id='gcp',
filetype=FileType.CSV,
),
output_table=Table(
name='raw_online_retail',
Selected from shared topics, language and repository description—not editorial ratings.
7vaibhavjk /
This project demonstrates an end-to-end ELT pipeline leveraging Snowflake, dbt, Apache Airflow, and Astronomer. Using Snowflake's TPCH sample dataset, it showcases modern data engineering practices including data transformation, testing, modeling, and workflow orchestration.