From 1645f32d51999d41d62166e296a89aeeccac0d29 Mon Sep 17 00:00:00 2001 From: shana-git Date: Wed, 11 Sep 2024 21:38:09 +0300 Subject: [PATCH 1/2] add airflow --- Storage/Orchestration/ELTWeeklyProcess.py | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/Storage/Orchestration/ELTWeeklyProcess.py b/Storage/Orchestration/ELTWeeklyProcess.py index bd35a5cd..915ed057 100644 --- a/Storage/Orchestration/ELTWeeklyProcess.py +++ b/Storage/Orchestration/ELTWeeklyProcess.py @@ -3,6 +3,8 @@ from datetime import datetime from ..ELTS.CustomerPurchaseFrequencyTotalSpendWeeklyELT import load_elt from ..ELTS.topSellingArtistsWeeklyELT import load_and_transform_data +from ..ELTS.employeeSalePerformanceCustomerInteractionsWeeklyELT import load_employees_sales_customer_interactions_elt +from ..ELTS.CustomerAverageSpendWeeklyELT import load_average_purchase_value_elt # from ELTS import X # Define your Python functions here @@ -24,6 +26,12 @@ def run_customer_purchase_frequency_total_spend(): def run_top_sell_artists(): load_and_transform_data() +def run_employees_sales_customer_interactions(): + load_employees_sales_customer_interactions_elt() + +def run_customer_invoices_count(): + load_average_purchase_value_elt() + # Define default arguments for the DAG default_args = { 'owner': 'airflow', @@ -75,6 +83,14 @@ def run_top_sell_artists(): task_id='run_top_sell_artists', python_callable=run_top_sell_artists, ) + task_employees_sales_customer_interactions( + task_id='run_employees_sales_customer_interactions', + python_callable=run_employees_sales_customer_interactions, + ) + task_average_purchase_value_elt( + task_id='run_customer_invoices_count_etl', + python_callable=run_customer_invoices_count, + ) # Define dependencies task_1 >> task_2 task_3 >> task_4 From dcc45135b9b59187950301ddd22618ca62bcb2d0 Mon Sep 17 00:00:00 2001 From: shana-git Date: Thu, 12 Sep 2024 09:04:51 +0300 Subject: [PATCH 2/2] rename files --- ...verageSpendDailyELT.py => AveragePurchaseValueDailyELT.py} | 0 ...rageSpendWeeklyELT.py => AveragePurchaseValueWeeklyELT.py} | 0 Storage/ETLS/AveragePurchaseValueDailyETL.py | 4 ++-- Storage/ETLS/AveragePurchaseValueWeeklyETL.py | 4 ++-- 4 files changed, 4 insertions(+), 4 deletions(-) rename Storage/ELTS/{CustomerAverageSpendDailyELT.py => AveragePurchaseValueDailyELT.py} (100%) rename Storage/ELTS/{CustomerAverageSpendWeeklyELT.py => AveragePurchaseValueWeeklyELT.py} (100%) diff --git a/Storage/ELTS/CustomerAverageSpendDailyELT.py b/Storage/ELTS/AveragePurchaseValueDailyELT.py similarity index 100% rename from Storage/ELTS/CustomerAverageSpendDailyELT.py rename to Storage/ELTS/AveragePurchaseValueDailyELT.py diff --git a/Storage/ELTS/CustomerAverageSpendWeeklyELT.py b/Storage/ELTS/AveragePurchaseValueWeeklyELT.py similarity index 100% rename from Storage/ELTS/CustomerAverageSpendWeeklyELT.py rename to Storage/ELTS/AveragePurchaseValueWeeklyELT.py diff --git a/Storage/ETLS/AveragePurchaseValueDailyETL.py b/Storage/ETLS/AveragePurchaseValueDailyETL.py index 6c166ea8..2ba756ab 100644 --- a/Storage/ETLS/AveragePurchaseValueDailyETL.py +++ b/Storage/ETLS/AveragePurchaseValueDailyETL.py @@ -6,7 +6,7 @@ import sqlite3 import pandas as pd -def load_customer_invoices_count_etl_increment(): +def incremental_load_average_purchase_value_etl(): # Step 1: Initialize Spark session spark = SparkSession.builder \ .appName("ETL - Average Purchase Value Over Time by Customer Type") \ @@ -99,4 +99,4 @@ def load_customer_invoices_count_etl_increment(): spark.stop() if __name__ == "__main__": - load_customer_invoices_count_etl_increment() + incremental_load_average_purchase_value_etl() diff --git a/Storage/ETLS/AveragePurchaseValueWeeklyETL.py b/Storage/ETLS/AveragePurchaseValueWeeklyETL.py index fb81ef4c..d97e69c6 100644 --- a/Storage/ETLS/AveragePurchaseValueWeeklyETL.py +++ b/Storage/ETLS/AveragePurchaseValueWeeklyETL.py @@ -8,7 +8,7 @@ # import KT_DB # Assuming KT_DB is the library for SQLite operations -def load_customer_invoices_count_etl(): +def load_average_purchase_value_etl(): # Step 1: Initialize Spark session spark = SparkSession.builder \ .appName("ETL - Average Purchase Value Over Time by Customer Type") \ @@ -66,4 +66,4 @@ def load_customer_invoices_count_etl(): if __name__ == "__main__": - load_customer_invoices_count_etl() + load_average_purchase_value_etl()