add workflow 天润Smart-ccc客户数据,dev

dev/workflow/TK_Cust/smart_ccc_custom/tr_custom_details/S98_S_tr_custom_details.sql

@@ -85,9 +85,9 @@ insert into p10_sa.S98_S_tr_custom_details
 	, label_ids  
 	, etl_tx_dt  
   from p00_tal.S98_S_tr_custom_details
-  ;
+   ;
   delete from p12_sfull.S98_S_tr_custom_details
-;
+ where (id) in (select id from p10_sa.S98_S_tr_custom_details) ;
 ;
 insert into p12_sfull.S98_S_tr_custom_details
 (	 id  

dev/workflow/TK_Cust/smart_ccc_custom/天润Smart-ccc客户数据/wf_dag_smart_ccc_custom.py

@@ -37,7 +37,7 @@ task_failed = EmailOperator (
     cc=[""],
     subject="smart_ccc_custom_failed",
     html_content='<h3>您好，smart_ccc_custom作业失败，请及时处理" </h3>')
-
+
 customer_list_feign = SSHOperator(
 ssh_hook=sshHook,
 task_id='customer_list_feign',
@@ -46,7 +46,17 @@ depends_on_past=False,
 retries=3, 
 dag=dag)
 
+customer_list_load = SSHOperator(
+ssh_hook=sshHook,
+task_id='customer_list_load',
+command='/data/airflow/etl/API/run_psql.sh {{ ds_nodash }} {{params.my_param }}',
+params={'my_param':"customer_list_load"},
+depends_on_past=False, 
+retries=3, 
+dag=dag)
 
+customer_list_feign >> customer_list_load
+
 customer_labels_feign = SSHOperator(
 ssh_hook=sshHook,
 task_id='customer_labels_feign',
@@ -65,7 +75,7 @@ retries=3,
 dag=dag)
 
 customer_labels_feign >> customer_labels_load
-
+
 tr_custom_details_5516 = SSHOperator(
 ssh_hook=sshHook,
 task_id='tr_custom_details_5516',
@@ -74,7 +84,7 @@ params={'my_param':"S98_S_tr_custom_details"},
 depends_on_past=False, 
 retries=3, 
 dag=dag)
-
+
 tr_custom_labels_8280 = SSHOperator(
 ssh_hook=sshHook,
 task_id='tr_custom_labels_8280',
@@ -83,8 +93,27 @@ params={'my_param':"S98_S_tr_custom_labels"},
 depends_on_past=False, 
 retries=3, 
 dag=dag)
+
+customer_detail_feign = SSHOperator(
+ssh_hook=sshHook,
+task_id='customer_detail_feign',
+command='python3 /data/airflow/etl/API/customer_detail_feign.py',
+depends_on_past=False, 
+retries=3, 
+dag=dag)
 
-customer_labels_load >> tr_custom_labels_8280
-customer_list_feign >> tr_custom_details_5516
-tr_custom_details_5516 >> task_failed 
-tr_custom_labels_8280 >> task_failed 
+customer_detail_load = SSHOperator(
+ssh_hook=sshHook,
+task_id='customer_detail_load',
+command='/data/airflow/etl/API/run_psql.sh {{ ds_nodash }} {{params.my_param }}',
+params={'my_param':"customer_detail_load"},
+depends_on_past=False, 
+retries=3, 
+dag=dag)
+
+customer_detail_feign >> customer_detail_load
+
+customer_labels_load >> tr_custom_labels_8280
+customer_list_load >> tr_custom_details_5516
+tr_custom_details_5516 >> task_failed 
+customer_detail_load >> task_failed