Update sql files

Author: ProgrammingOperative

airflow/dags/postgres_dags/load_data.py

@@ -1,5 +1,7 @@
 import json
 import pathlib
+from datetime import datetime, timedelta
+
 
 import airflow
 import requests
@@ -12,23 +14,33 @@
 import sys
 import os
 
-dag = DAG(
-   dag_id="Load_traffic_data",
-   start_date=airflow.utils.dates.days_ago(1),
+default_args = {
+   'depends_on_past': False,
+   'email': ['airflow@example.com'],
+   'email_on_failure': False,
+   'email_on_retry': False,
+   'retries': 1,
+   'retry_delay': timedelta(minutes=5)
+}
+
+with DAG(
+   'Load_traffic_data',
+   default_args = default_args,
+   start_date=datetime(2022, 8, 20),
    schedule_interval=None,
    catchup=False,
-)
+) as dag:
 
-create_traffic_table = PostgresOperator(
-   task_id = 'Create_table'
-   postgres_conn_id="postgres_connect",
-   sql="sql/create_table.sql",
-)
+   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",
-)
+   load_traffic_table = PostgresOperator(
+      task_id = 'Load_table',
+      postgres_conn_id='postgres_connect',
+      sql="sql/create_table.sql",
+   )
 
 create_traffic_table >> load_traffic_table

airflow/dags/postgres_dags/sql/load_data.sql

@@ -1,4 +1,5 @@
-COPY vehicles(track_id, vehicle_type, traveled_d, avg_speed, lat, lon, speed, lon_acc, lat_acc, record_time)
-FROM '../../../../data/clean_data.csv'
+COPY traffic_table(id, track_id, vehicle_type, traveled_d, avg_speed, lat, lon, speed, lon_acc, lat_acc, record_time)
+FROM '/home/wacira/10Academy/Traffic Data ETL week11/repository/traffic_data_etl/data/clean_data.csv'
 DELIMITER ','
-CSV HEADER;
+CSV HEADER;
+