Loading repository data…
nkoda / repository
This data pipeline project focuses on ingesting and processing raw stock market data, performing feature engineering on top of the processed data, training a predictive model, and building an API service to serve the trained model.
A transparent discovery signal based on current public GitHub metadata.
This score does not audit code, security, maintainers, documentation quality, or suitability. Verify the repository and its current documentation before adoption.
This data pipeline project focuses on ingesting and processing raw stock market data, performing feature engineering on top of the processed data, training a predictive model, and building an API service to serve the trained model. The API is currently hosted on Render. The pipeline is designed using a DAG (Directed Acyclic Graph) oriented tool, Apache Airflow, to streamline data processing and management with tracking data lineage, ensuring data integrity, and minimizing errors during processing.
The model was trained using LightGBM, an open-source gradient boosting framework that uses tree-based learning algorithms. LightGBM is designed to be memory-efficient and fast, making it a good choice for large datasets.
To improve the runtime for data ingestion and feature engineering, the ThreadPoolExecutor module was used to enable parallel processing of data. This allowed for faster loading of data into memory during the training process. Additionally, a subset of the dataframe was used during the feature engineering to improve runtime by reducing the amount of calculations made.
The pipeline consists of four stages, each building on top of the previous one:
The pipeline is defined in Airflow using DAG (Directed Acyclic Graph), where each stage is represented by a task and dependencies between tasks are represented by edges. The pipeline specification can be found in dags/data_pipeline.py, and a visual diagram representation can be found in the docs/data_pipeline.drawio.svg.
Docker: The pipeline is designed to be run using Docker, so you will need to have Docker installed on your system. You can download and install Docker from the official Docker website:
Selected from shared topics, language and repository description—not editorial ratings.
chayansraj /
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.
59/100 healthultranet1 /
Project Description: A music streaming company wants to introduce more automation and monitoring to their data warehouse ETL pipelines and they have come to the conclusion that the best tool to achieve this is Apache Airflow. As their Data Engineer, I was tasked to create a reusable production-grade data pipeline that incorporates data quality checks and allows for easy backfills. Several analysts and Data Scientists rely on the output generated by this pipeline and it is expected that the pipeline runs daily on a schedule by pulling new data from the source and store the results to the destination. Data Description: The source data resides in S3 and needs to be processed in a data warehouse in Amazon Redshift. The source datasets consist of JSON logs that tell about user activity in the application and JSON metadata about the songs the users listen to. Data Pipeline design: At a high-level the pipeline does the following tasks. Extract data from multiple S3 locations. Load the data into Redshift cluster. Transform the data into a star schema. Perform data validation and data quality checks. Calculate the most played songs for the specified time interval. Load the result back into S3. dag Structure of the Airflow DAG Design Goals: Based on the requirements of our data consumers, our pipeline is required to adhere to the following guidelines: The DAG should not have any dependencies on past runs. On failure, the task is retried for 3 times. Retries happen every 5 minutes. Catchup is turned off. Do not email on retry. Pipeline Implementation: Apache Airflow is a Python framework for programmatically creating workflows in DAGs, e.g. ETL processes, generating reports, and retraining models on a daily basis. The Airflow UI automatically parses our DAG and creates a natural representation for the movement and transformation of data. A DAG simply is a collection of all the tasks you want to run, organized in a way that reflects their relationships and dependencies. A DAG describes how you want to carry out your workflow, and Operators determine what actually gets done. By default, airflow comes with some simple built-in operators like PythonOperator, BashOperator, DummyOperator etc., however, airflow lets you extend the features of a BaseOperator and create custom operators. For this project, I developed several custom operators. operators The description of each of these operators follows: StageToRedshiftOperator: Stages data to a specific redshift cluster from a specified S3 location. Operator uses templated fields to handle partitioned S3 locations. LoadFactOperator: Loads data to the given fact table by running the provided sql statement. Supports delete-insert and append style loads. LoadDimensionOperator: Loads data to the given dimension table by running the provided sql statement. Supports delete-insert and append style loads. SubDagOperator: Two or more operators can be grouped into one task using the SubDagOperator. Here, I am grouping the tasks of checking if the given table has rows and then run a series of data quality sql commands. HasRowsOperator: Data quality check to ensure that the specified table has rows. DataQualityOperator: Performs data quality checks by running sql statements to validate the data. SongPopularityOperator: Calculates the top ten most popular songs for a given interval. The interval is dictated by the DAG schedule. UnloadToS3Operator: Stores the analysis result back to the given S3 location. Code for each of these operators is located in the plugins/operators directory. Pipeline Schedule and Data Partitioning: The events data residing on S3 is partitioned by year (2018) and month (11). Our task is to incrementally load the event json files, and run it through the entire pipeline to calculate song popularity and store the result back into S3. In this manner, we can obtain the top songs per day in an automated fashion using the pipeline. Please note, this is a trivial analyis, but you can imagine other complex queries that follow similar structure. S3 Input events data: s3://<bucket>/log_data/2018/11/ 2018-11-01-events.json 2018-11-02-events.json 2018-11-03-events.json .. 2018-11-28-events.json 2018-11-29-events.json 2018-11-30-events.json S3 Output song popularity data: s3://skuchkula-topsongs/ songpopularity_2018-11-01 songpopularity_2018-11-02 songpopularity_2018-11-03 ... songpopularity_2018-11-28 songpopularity_2018-11-29 songpopularity_2018-11-30 The DAG can be configured by giving it some default_args which specify the start_date, end_date and other design choices which I have mentioned above. default_args = { 'owner': 'shravan', 'start_date': datetime(2018, 11, 1), 'end_date': datetime(2018, 11, 30), 'depends_on_past': False, 'email_on_retry': False, 'retries': 3, 'retry_delay': timedelta(minutes=5), 'catchup_by_default': False, 'provide_context': True, } How to run this project? Step 1: Create AWS Redshift Cluster using either the console or through the notebook provided in create-redshift-cluster Run the notebook to create AWS Redshift Cluster. Make a note of: DWN_ENDPOINT :: dwhcluster.c4m4dhrmsdov.us-west-2.redshift.amazonaws.com DWH_ROLE_ARN :: arn:aws:iam::506140549518:role/dwhRole Step 2: Start Apache Airflow Run docker-compose up from the directory containing docker-compose.yml. Ensure that you have mapped the volume to point to the location where you have your DAGs. NOTE: You can find details of how to manage Apache Airflow on mac here: https://gist.github.com/shravan-kuchkula/a3f357ff34cf5e3b862f3132fb599cf3 start_airflow Step 3: Configure Apache Airflow Hooks On the left is the S3 connection. The Login and password are the IAM user's access key and secret key that you created. Basically, by using these credentials, we are able to read data from S3. On the right is the redshift connection. These values can be easily gathered from your Redshift cluster connections Step 4: Execute the create-tables-dag This dag will create the staging, fact and dimension tables. The reason we need to trigger this manually is because, we want to keep this out of main dag. Normally, creation of tables can be handled by just triggering a script. But for the sake of illustration, I created a DAG for this and had Airflow trigger the DAG. You can turn off the DAG once it is completed. After running this DAG, you should see all the tables created in the AWS Redshift. Step 5: Turn on the load_and_transform_data_in_redshift dag As the execution start date is 2018-11-1 with a schedule interval @daily and the execution end date is 2018-11-30, Airflow will automatically trigger and schedule the dag runs once per day for 30 times. Shown below are the 30 DAG runs ranging from start_date till end_date, that are trigged by airflow once per day. schedule
Docker requirements: Upon testing, the container needs sufficient resources to complete the DAG pipeline. If your container instance is failing to complete the DAG, please allocate more RAM / CPU cores for your Docker container. You will know if you need more resources when the Airflow task fails with "Negsignal.SIGKILL" appearing in the task logs. Refer to this Stackoverflow post
Git LFS: The pipeline uses Git LFS (Large File Storage) to manage large files. You will need to have Git LFS installed on your system. You can download and install Git LFS from the official Git LFS website: https://git-lfs.github.com/
Local Instance: If you plan to run Airflow locally, you can find all the necessary python packages to run the scripts under requirements_data-pipeline.txt. Additionally be sure to set the dags_folder in the AIRFLOW_HOME directory to point to this repositories /dags directory. This can be done by modifying the airflow.cfg file located in cd ~/ariflow.
Note: If you are using a Linux-based operating system, you may need to install Git LFS using your package manager. For example, on Ubuntu, you can install Git LFS by running the following command:
sudo apt-get install git-lfs
To run the pipeline, follow the instructions below:
git clone https://github.com/nkoda/Work-Sample_Data-Engineer.git
cd Work-Sample_Data-Engineer
docker-compose up --build
Access the Airflow web interface at http://localhost:8080 and enable the data_pipeline DAG. The login credentials is the string "admin" for both the username and password.
Run the pipeline by pressing the trigger button highlighted in red in the image below:
After running the pipeline, the following artifacts will be produced:
data/processed/preprocessed_data.parquet.data/training/augmented_data.parquet.web_api/ml-model/lightgbm_predictor.joblib.logs/ directory. Training metrics are specifically saved in logs/training.log.The API is seperate from the data pipeline therefore not intended to be ran directly in the Docker container. As such there are seperate instructions to run a local API service. A cloud hosted API of this ML model can be found on Render.
cd ml-model/
pip install requirements_web-api.txt
flask run --port=5000
The service will be available at http://localhost:5000.
The following API endpoint is available:
GET /predict?vol_moving_avg={vol_moving_avg}&adj_close_rolling_med={adj_close_rolling_med}: Returns the predicted trading volume for the given values of vol_moving_avg and adj_close_rolling_med.The following improvement suggestions can be made for future iterations of the pipeline:
AnthonyByansi /
Automate your data pipelines using Apache Airflow with this ready-to-use DAG for data integration, ETL and workflow automation.
66/100 healthgetfelipe /
🔹This project builds a real-time ELT pipeline where a Python script sends data to Kafka. An Airflow DAG ingests the data and stores it in S3. A second DAG moves the data from S3 to Postgres, and a dbt DAG handles data transformation. Finally, Metabase is used to create dashboards based on the transformed data.
37/100 healthomar-elmaria /
This repo contains a data pipeline composed of Python and Big Query scripts that extract, clean, and aggregate data, as well as perform statistical significance tests. The code is fully orchestrated on Airflow and feeds a Tableau dashboard that displays the success metrics of surge pricing switchback experiments.
41/100 healthpriyanshubiswas-tech /
This project was created as part of an assessment for DigitalXC AI. It demonstrates a cloud-based ELT pipeline using AWS MWAA, Airflow, dbt, PostgreSQL, and Superset. The pipeline automates data ingestion from S3, transformation with dbt, and visualization through Superset, following modern data engineering practices on a scalable AWS architecture.
47/100 health