Skip to content

add airflow #253

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
shana9317 wants to merge 2 commits into
base: main
Choose a base branch
from
Open

add airflow #253

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
1645f32
add airflow
shana9317 Sep 11, 2024
dcc4513
rename files
shana9317 Sep 12, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
0 Storage/ELTS/CustomerAverageSpendDailyELT.py → Storage/ELTS/AveragePurchaseValueDailyELT.py
View file
Open in desktop
File renamed without changes.
0 ...age/ELTS/CustomerAverageSpendWeeklyELT.py → ...age/ELTS/AveragePurchaseValueWeeklyELT.py
View file
Open in desktop
File renamed without changes.
4 changes: 2 additions & 2 deletions Storage/ETLS/AveragePurchaseValueDailyETL.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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") \
Expand Down Expand Up @@ -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()
4 changes: 2 additions & 2 deletions Storage/ETLS/AveragePurchaseValueWeeklyETL.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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") \
Expand Down Expand Up @@ -66,4 +66,4 @@ def load_customer_invoices_count_etl():


if __name__ == "__main__":
load_customer_invoices_count_etl()
load_average_purchase_value_etl()
16 changes: 16 additions & 0 deletions Storage/Orchestration/ELTWeeklyProcess.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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',
Expand Down Expand Up @@ -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
Expand Down