diff --git a/airflow/dags/config_interface_dag.py b/airflow/dags/config_interface_dag.py index 640cdc9..9f83064 100644 --- a/airflow/dags/config_interface_dag.py +++ b/airflow/dags/config_interface_dag.py @@ -77,14 +77,14 @@ dag = DAG( begin_job = EmptyOperator(task_id='Begin') -load_training_results = PythonOperator( +load_request_submission = PythonOperator( task_id="load_request_submissions", python_callable=call_stored_proc, dag=dag, op_kwargs={'proc_name':'LOAD_REQUEST_SUBMISSION', 'file_type':'request_submission'} ) -load_attempt_logs_sp = PythonOperator( +load_contract_config = PythonOperator( task_id="load_contract_config", python_callable=call_stored_proc, dag=dag, @@ -97,4 +97,4 @@ load_attempt_logs_sp = PythonOperator( end_job = EmptyOperator(task_id='End') -begin_job >> load_training_results >> load_attempt_logs_sp >> end_job \ No newline at end of file +begin_job >> load_request_submission >> load_contract_config >> end_job \ No newline at end of file diff --git a/airflow/dags/raw_training_data_dag.py b/airflow/dags/raw_training_data_dag.py index c337f65..f8156d4 100644 --- a/airflow/dags/raw_training_data_dag.py +++ b/airflow/dags/raw_training_data_dag.py @@ -37,7 +37,7 @@ ONE_FAILED = 'one_failed' # Passing empty params for now, this will be overridden by the payload from the trigger # These params can also be set from the Airflow UI while manually triggering the DAG -default_params = {"column_config_file_name": "", "raw_training_data_file_name":""} +default_params = {"column_config_file_name": "", "raw_training_data_file_name":"", "business_config_file_name":""} # This will be replaced with the payload from the event after API connection is setup # training_results_file_name = "training_results_sample.csv" # attempt_logs_file_name = "attempt_logs_sample.csv" @@ -55,6 +55,8 @@ def call_stored_proc(proc_name,file_type, params): file_name = params['column_config_file_name'] elif file_type == 'raw_training_data': file_name = params['raw_training_data_file_name'] + elif file_type == 'business_config': + file_name = params['business_config_file_name'] cur.execute(f"CALL {DATABASE}.STG.{proc_name}('{file_name}');") result = cur.fetchone() @@ -91,9 +93,16 @@ load_attempt_logs_sp = PythonOperator( op_kwargs={'proc_name':'LOAD_TRAINING_DATA_RAW', 'file_type':'raw_training_data'} ) +load_business_config = PythonOperator( + task_id="load_business_config", + python_callable=call_stored_proc, + dag=dag, + op_kwargs={'proc_name':'LOAD_BUSINESS_CONFIG', 'file_type':'business_config'} +) + end_job = EmptyOperator(task_id='End') -begin_job >> load_training_results >> load_attempt_logs_sp >> end_job \ No newline at end of file +begin_job >> load_training_results >> load_attempt_logs_sp >> load_business_config >> end_job \ No newline at end of file diff --git a/snowflake/scripts/training_interface/R__005_LOAD_COLUMN_CONFIG_SP.sql b/snowflake/scripts/training_interface/R__005_LOAD_COLUMN_CONFIG_SP.sql index 950c02f..5248a83 100644 --- a/snowflake/scripts/training_interface/R__005_LOAD_COLUMN_CONFIG_SP.sql +++ b/snowflake/scripts/training_interface/R__005_LOAD_COLUMN_CONFIG_SP.sql @@ -16,7 +16,7 @@ BEGIN EXECUTE IMMEDIATE 'CREATE OR REPLACE STAGE STG.COLUMN_CONFIG_STAGE STORAGE_INTEGRATION = dev_bucket_integration URL = ''s3://doczy-dev-infra-raw-data-ingestion/training_data_raw/' || :file_name || ''' - FILE_FORMAT = (FORMAT_NAME = ''STG.RAW_TRAINING_DATA_FORMAT'');'; + FILE_FORMAT = (FORMAT_NAME = ''STG.CSV_HEADER'');'; call stg.log_audit(:procedure_name, 'Section 1', 99, 'END'); @@ -36,7 +36,7 @@ BEGIN NULLIF(TRIM($2), '') AS COLUMN_DATATYPE FROM @STG.COLUMN_CONFIG_STAGE ) - FILE_FORMAT = (FORMAT_NAME = 'STG.RAW_TRAINING_DATA_FORMAT') + FILE_FORMAT = (FORMAT_NAME = 'STG.CSV_HEADER') ON_ERROR = ABORT_STATEMENT; call stg.log_audit(:procedure_name, 'Section 3', 99, 'END');