Merge pull request #15 from ProgrammingOperative/orchestration

Orchestration

Author: ProgrammingOperative

,

@@ -0,0 +1,78 @@
+import json
+import pathlib
+from datetime import datetime, timedelta
+import os, sys
+import pandas as pd
+
+import airflow
+import requests
+import requests.exceptions as requests_exceptions
+from airflow import DAG
+
+from airflow.operators.bash import BashOperator
+from airflow.operators.python import PythonOperator
+from airflow.providers.postgres.operators.postgres import PostgresOperator
+
+import sys
+import os
+
+SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
+sys.path.append(os.path.dirname(SCRIPT_DIR))
+
+DBT_PROJECT_DIR = "~/dbt"
+
+from postgres_dags.extract_data import ExtractCSV
+
+extract_it = ExtractCSV()
+
+default_args = {
+   "owner": "ProgrammingOperative",
+   'depends_on_past': False,
+   'email': ['wachura11t@gmail.com'],
+   'email_on_failure': False,
+   'email_on_retry': False,
+   'retries': 1,
+   'retry_delay': timedelta(minutes=5)
+}
+
+# /home/wacira/10Academy/ETL_week11/repository/traffic_data_etl/dbt
+def extract():
+   data = extract_it.load_csv("~/data/warehousedata.csv")
+   restructured_df = extract_it.restructure(data)
+   path = '~/data/cleaned.csv'
+   restructured_df.to_csv(path)
+   
+
+with DAG(
+   'Load_traffic_data',
+   default_args = default_args,
+   start_date=datetime(2022, 8, 22),
+   schedule_interval=None,
+   catchup=False,
+) as dag:
+
+   extract_data = PythonOperator(
+      task_id = 'extract_data',
+      python_callable=extract
+   )
+
+   create_traffic_table = PostgresOperator(
+      task_id = 'Create_table',
+      postgres_conn_id='postgres_connect',
+      sql="sql/create_table.sql",
+   )
+
+   load_traffic_table = PostgresOperator(
+      task_id = 'Load_table',
+      postgres_conn_id='postgres_connect',
+      sql="sql/create_table.sql",
+   )
+
+   transform = BashOperator(
+      task_id = 'dbt_transformation',
+      bash_command='cd ~/10Academy/ETL_week11/repository/traffic_data_etl/dbt && dbt run',
+      )
+   
+
+
+extract_data >> create_traffic_table >> load_traffic_table >> transform 

airflow/dags/postgres_dags/elt_workflow.py

@@ -0,0 +1,56 @@
+import pandas as pd
+
+import os, sys
+
+class ExtractCSV:
+    def __init__(self) -> None:
+        pass
+
+    def load_csv(self, path):
+        data_frame = pd.read_csv(path)
+        return data_frame
+
+    def restructure(self, data):
+        # data = pd.read_csv("~/10 Academy/week 11/repository/traffic_data_etl/data/warehousedata.csv")
+        cols = data.columns[0].split(";")
+
+    # Strip the data of empty spaces
+        for i in range(len(cols)):
+            cols[i] = cols[i].strip()
+        
+    # Start restructuring by including records with only the columns stipulated
+        track_ids = []
+        types = []
+        traveled_d = []
+        avg_speeds = []
+        lat = []
+        lon = []
+        speed = []
+        lon_acc = []
+        lat_acc = []
+        time = []
+
+        for r in range(len(data)): 
+            row = data.iloc[r,:][0].split(";")
+            row_p1 = row[:]    
+            track_ids.append(row_p1[0])
+            types.append(row_p1[1])
+            traveled_d.append(row_p1[2])
+            avg_speeds.append(row_p1[3])
+            lat.append(row_p1[4])
+            lon.append(row_p1[5])
+            speed.append(row_p1[6])
+            lon_acc.append(row_p1[7])
+            lat_acc.append(row_p1[8])
+            time.append(row_p1[9])
+
+        #Create a dictionary first before creation of pandas dataframe
+        data_dict = {cols[0]:track_ids, cols[1]:types, cols[2]:traveled_d, cols[3]:avg_speeds, cols[4]:lat, cols[5]:lon, cols[6]:speed, cols[7]:lon_acc, cols[8]:lat_acc, cols[9]:time}
+        
+        #Create dataframe
+        extract_df = pd.DataFrame(data_dict)
+
+
+        #Final output
+        print("-----------------Successfully Extracted data---------------------")
+        return extract_df

airflow/dags/postgres_dags/extract_data.py

@@ -0,0 +1,14 @@
+create table if not exists traffic_table
+(
+id BIGSERIAL PRIMARY KEY,
+track_id INT NOT NULL,
+vehicle_type VARCHAR(500) NOT NULL,
+traveled_d VARCHAR(500) NOT NULL,
+avg_speed FLOAT NOT NULL,
+lat FLOAT NOT NULL,
+lon FLOAT NOT NULL,
+speed FLOAT NOT NULL,
+lon_acc FLOAT NOT NULL,
+lat_acc FLOAT NOT NULL,
+record_time FLOAT NOT NULL
+);

scripts/load_data.py renamed to airflow/dags/postgres_dags/load_data.py

@@ -0,0 +1,5 @@
+COPY traffic_table(id, track_id, vehicle_type, traveled_d, avg_speed, lat, lon, speed, lon_acc, lat_acc, record_time)
+FROM '../../../../data/clean_data.csv'
+DELIMITER ','
+CSV HEADER;
+

airflow/dags/postgres_dags/sql/create_table.sql

@@ -538,6 +538,25 @@
     "\n",
     "\n"
    ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 1,
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "['/home/wacira/10Academy/Traffic Data ETL week11/repository/traffic_data_etl/notebooks', '/home/wacira/.vscode/extensions/ms-toolsai.jupyter-2022.7.1102252217/pythonFiles', '/home/wacira/.vscode/extensions/ms-toolsai.jupyter-2022.7.1102252217/pythonFiles/lib/python', '/home/wacira/anaconda3/envs/data_engineering/lib/python310.zip', '/home/wacira/anaconda3/envs/data_engineering/lib/python3.10', '/home/wacira/anaconda3/envs/data_engineering/lib/python3.10/lib-dynload', '', '/home/wacira/.local/lib/python3.10/site-packages', '/home/wacira/anaconda3/envs/data_engineering/lib/python3.10/site-packages', '/home/wacira/anaconda3/envs/data_engineering/lib/python3.10/site-packages/GDAL-3.5.0-py3.10-linux-x86_64.egg']\n"
+     ]
+    }
+   ],
+   "source": [
+    "import sys\n",
+    "\n",
+    "print(sys.path)"
+   ]
   }
  ],
  "metadata": {