dtgitea
/
dataops
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

add workflow 泰克客户,dev

This commit is contained in:
root 2024-05-30 17:47:15 +08:00
parent 08d75aa4ef
commit 1d69f784ea
3 changed files with 394 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
dev/workflow/TK_Cust/tk_cust/数据源更新时间/data_source_update.sql Normal file
View File

@ -0,0 +1,34 @@
DROP TABLE IF EXISTS p60_mart.data_source_update;
CREATE TABLE IF NOT EXISTS p60_mart.data_source_update (
date_layer varchar(20)
, data_area varchar(20)
, data_table varchar(200)
, data_rows bigint
, column_qty int
, last_upt_tm timestamp(0)
, Etl_Batch_No varchar(50)
, Etl_First_Dt timestamp(0)
, Etl_Job varchar(200)
, Etl_Proc_Dt timestamp(0)
, Etl_Tx_Dt timestamp(0)
, Src_Sysname varchar(50)
, Src_Table varchar(50)
,primary key( date_layer,data_table )
);
COMMENT ON COLUMN p60_mart.data_source_update.date_layer IS '数据层级';
COMMENT ON COLUMN p60_mart.data_source_update.data_area IS '数据领域';
COMMENT ON COLUMN p60_mart.data_source_update.data_table IS '数据表';
COMMENT ON COLUMN p60_mart.data_source_update.data_rows IS '数据行数';
COMMENT ON COLUMN p60_mart.data_source_update.column_qty IS '字段数';
COMMENT ON COLUMN p60_mart.data_source_update.last_upt_tm IS '最后更新日期';
COMMENT ON COLUMN p60_mart.data_source_update.Etl_Batch_No IS '作业批次号';
COMMENT ON COLUMN p60_mart.data_source_update.Etl_First_Dt IS '最初入库时间';
COMMENT ON COLUMN p60_mart.data_source_update.Etl_Job IS '作业名称';
COMMENT ON COLUMN p60_mart.data_source_update.Etl_Proc_Dt IS '本次入库时间';
COMMENT ON COLUMN p60_mart.data_source_update.Etl_Tx_Dt IS '作业运行时间';
COMMENT ON COLUMN p60_mart.data_source_update.Src_Sysname IS '来源系统';
COMMENT ON COLUMN p60_mart.data_source_update.Src_Table IS '来源表';
COMMENT ON TABLE p60_mart.data_source_update IS '数据源更新时间';

350
dev/workflow/TK_Cust/tk_cust/数据源更新时间/data_source_update_agi.sql Normal file
View File

@ -0,0 +1,350 @@
/***************************************************************************************************/
/*script in Sql, generate by SdmCreateScript 2020(by Qihang Feng, QF255001@TERADATA.COM) */
/*VERSION 01.10 revised on 2020-08-25 */
/*Brilliance stems from wisdoms. */
/*************Head Section**************************************************************************/
/*Script Use: Periodically load data to :data_source_update(数据源更新时间) */
/*Create Date:2024-05-30 15:30:54 */
/*SDM Developed By: dev */
/*SDM Developed Date: 2024-05-28 */
/*SDM Checked By: dev */
/*SDM Checked Date: 2024-05-30 */
/*Script Developed By: dev */
/*Script Checked By: dev */
/*Source table 1: p10_sa.s98_s_customer_event_meta */
/*Job Type: Inbound transform (Tier 1 to Tier 2) */
/*Target Table:data_source_update */
/*ETL Job Name:data_source_update */
/*ETL Frequency:Daily */
/*ETL Policy:F2 */
/********************************************************************************************/
/*******Main Section**************************************************************************/
\set ON_ERROR_STOP on
\set AUTOCOMMIT on
\timing on
/*创建临时表加载当前数据 */
CREATE TEMPORARY TABLE data_source_update_agi_CUR_I
( LIKE :MARTDB.data_source_update)
ON COMMIT PRESERVE ROWS;
/*创建临时表加载不同数据 */
CREATE TEMPORARY TABLE data_source_update_agi_INS
( LIKE :MARTDB.data_source_update)
ON COMMIT PRESERVE ROWS;
/*****************************************************************************************************/
/* GROUP 1:Source Table:s98_s_customer_event_meta*****************************************************/
/*****************************************************************************************************/
INSERT INTO data_source_update_agi_CUR_I (
date_layer /*数据层级*/
,data_area /*数据领域*/
,data_table /*数据表*/
,data_rows /*数据行数*/
,column_qty /*字段数*/
,last_upt_tm /*最后更新日期*/
,Etl_Batch_No /*作业批次号*/
,Etl_First_Dt /*最初入库时间*/
,Etl_Job /*作业名称*/
,Etl_Proc_Dt /*本次入库时间*/
,Etl_Tx_Dt /*作业运行时间*/
,Src_Sysname /*来源系统*/
,Src_Table /*来源表*/
)
SELECT
'Data Source' /*date_layer*/
,case when p0 .data_table like '%scrm%' then 'SCRM'
when p0.data_table like '%customer%' then 'SCRM'
when p0.data_table like '%crm%' then 'CRM'
when p0.data_table like '%livechat%' then 'LiveChat'
when p0.data_table like '%udesk%'
then 'Udesk'
else 'All' end /*data_area*/
,coalesce(p0.data_table,'') /*data_table*/
,coalesce(p0.row_count,0) /*data_rows*/
,coalesce(p0.columns_qty,0) /*column_qty*/
,p0.last_upt_tm /*last_upt_tm*/
,0 /*Etl_Batch_No*/
,TO_DATE(:TXDATE,'YYYYMMDD') /*Etl_First_Dt*/
,:ETLJOB /*Etl_Job*/
,current_timestamp(0) /*Etl_Proc_Dt*/
,TO_DATE(:TXDATE,'YYYYMMDD') /*Etl_Tx_Dt*/
,Substr('s98_s_customer_event_meta',1,3) /*Src_Sysname*/
,'s98_s_customer_event_meta' /*Src_Table*/
FROM (select 's98_s_crm_account' data_table,count(*) row_count ,max(etl_tx_dt) last_upt_tm , 41 columns_qty from p12_sfull.s98_s_crm_account
union all
select 's98_s_crm_contact_ccp' data_table,count(*) row_count ,max(etl_tx_dt) last_upt_tm , 4 columns_qty from p12_sfull.s98_s_crm_contact_ccp
union all
select 's98_s_crm_contact_part' data_table,count(*) row_count ,max(etl_tx_dt) last_upt_tm , 57 columns_qty from p12_sfull.s98_s_crm_contact_part
union all
select 's98_s_crm_opp' data_table,count(*) row_count ,max(etl_tx_dt) last_upt_tm , 50 columns_qty from p12_sfull.s98_s_crm_opp
union all
select 's98_s_crm_raw_leads' data_table,count(*) row_count ,max(etl_tx_dt) last_upt_tm , 12 columns_qty from p12_sfull.s98_s_crm_raw_leads
union all
select 's98_s_custom_events_activity_submit' data_table,count(*) row_count ,max(etl_tx_dt) last_upt_tm , 15 columns_qty from p12_sfull.s98_s_custom_events_activity_submit
union all
select 's98_s_customer_event_meta' data_table,count(*) row_count ,max(etl_tx_dt) last_upt_tm , 4 columns_qty from p12_sfull.s98_s_customer_event_meta
union all
select 's98_s_customer_events' data_table,count(*) row_count ,max(etl_tx_dt) last_upt_tm , 12 columns_qty from p12_sfull.s98_s_customer_events
union all
select 's98_s_customer_events_add_user' data_table,count(*) row_count ,max(etl_tx_dt) last_upt_tm , 17 columns_qty from p12_sfull.s98_s_customer_events_add_user
union all
select 's98_s_customer_events_c_minipro_page_view' data_table,count(*) row_count ,max(etl_tx_dt) last_upt_tm , 25 columns_qty from p12_sfull.s98_s_customer_events_c_minipro_page_view
union all
select 's98_s_customer_events_click_link_in_page' data_table,count(*) row_count ,max(etl_tx_dt) last_upt_tm , 27 columns_qty from p12_sfull.s98_s_customer_events_click_link_in_page
union all
select 's98_s_customer_events_lastupdated' data_table,count(*) row_count ,max(etl_tx_dt) last_upt_tm , 28 columns_qty from p12_sfull.s98_s_customer_events_lastupdated
union all
select 's98_s_customer_events_open_app' data_table,count(*) row_count ,max(etl_tx_dt) last_upt_tm , 27 columns_qty from p12_sfull.s98_s_customer_events_open_app
union all
select 's98_s_customer_events_open_content_page' data_table,count(*) row_count ,max(etl_tx_dt) last_upt_tm , 27 columns_qty from p12_sfull.s98_s_customer_events_open_content_page
union all
select 's98_s_customer_events_open_page' data_table,count(*) row_count ,max(etl_tx_dt) last_upt_tm , 33 columns_qty from p12_sfull.s98_s_customer_events_open_page
union all
select 's98_s_customer_events_submit_form' data_table,count(*) row_count ,max(etl_tx_dt) last_upt_tm , 25 columns_qty from p12_sfull.s98_s_customer_events_submit_form
union all
select 's98_s_customer_events_subscribe' data_table,count(*) row_count ,max(etl_tx_dt) last_upt_tm , 13 columns_qty from p12_sfull.s98_s_customer_events_subscribe
union all
select 's98_s_customer_events_wechat_scan' data_table,count(*) row_count ,max(etl_tx_dt) last_upt_tm , 13 columns_qty from p12_sfull.s98_s_customer_events_wechat_scan
union all
select 's98_s_livechat' data_table,count(*) row_count ,max(etl_tx_dt) last_upt_tm , 98 columns_qty from p12_sfull.s98_s_livechat
union all
select 's98_s_scrm_contact' data_table,count(*) row_count ,max(etl_tx_dt) last_upt_tm , 133 columns_qty from p12_sfull.s98_s_scrm_contact
union all
select 's98_s_scrm_contact_af0912' data_table,count(*) row_count ,max(etl_tx_dt) last_upt_tm , 11 columns_qty from p12_sfull.s98_s_scrm_contact_af0912
union all
select 's98_s_scrm_contact_lastupdated' data_table,count(*) row_count ,max(etl_tx_dt) last_upt_tm , 133 columns_qty from p12_sfull.s98_s_scrm_contact_lastupdated
union all
select 's98_s_scrm_contact_merge' data_table,count(*) row_count ,max(etl_tx_dt) last_upt_tm , 20 columns_qty from p12_sfull.s98_s_scrm_contact_merge
union all
select 's98_s_scrm_contact_merge_af0912' data_table,count(*) row_count ,max(etl_tx_dt) last_upt_tm , 16 columns_qty from p12_sfull.s98_s_scrm_contact_merge_af0912
union all
select 's98_s_scrm_contact_update_af0912' data_table,count(*) row_count ,max(etl_tx_dt) last_upt_tm , 11 columns_qty from p12_sfull.s98_s_scrm_contact_update_af0912
union all
select 's98_s_scrm_leads_af0912' data_table,count(*) row_count ,max(etl_tx_dt) last_upt_tm , 13 columns_qty from p12_sfull.s98_s_scrm_leads_af0912
union all
select 's98_s_udesk_record' data_table,count(*) row_count ,max(etl_tx_dt) last_upt_tm , 21 columns_qty from p12_sfull.s98_s_udesk_record
) p0
;
/*****************************************************************************************************/
/* GROUP 2:Source Table:s98_s_customer_event_meta*****************************************************/
/*****************************************************************************************************/
INSERT INTO data_source_update_agi_CUR_I (
date_layer /*数据层级*/
,data_area /*数据领域*/
,data_table /*数据表*/
,data_rows /*数据行数*/
,column_qty /*字段数*/
,last_upt_tm /*最后更新日期*/
,Etl_Batch_No /*作业批次号*/
,Etl_First_Dt /*最初入库时间*/
,Etl_Job /*作业名称*/
,Etl_Proc_Dt /*本次入库时间*/
,Etl_Tx_Dt /*作业运行时间*/
,Src_Sysname /*来源系统*/
,Src_Table /*来源表*/
)
SELECT
'COM' /*date_layer*/
,case when p0 .data_table like '%scrm%' then 'SCRM'
when p0.data_table like '%customer%' then 'SCRM'
when p0.data_table like '%crm%' then 'CRM'
when p0.data_table like '%livechat%' then 'LiveChat'
when p0.data_table like '%udesk%'
then 'Udesk'
else 'All' end /*data_area*/
,coalesce(p0.data_table) /*data_table*/
,coalesce(p0.row_count) /*data_rows*/
,coalesce(p0.columns_qty) /*column_qty*/
,p0.last_upt_tm /*last_upt_tm*/
,0 /*Etl_Batch_No*/
,TO_DATE(:TXDATE,'YYYYMMDD') /*Etl_First_Dt*/
,:ETLJOB /*Etl_Job*/
,current_timestamp(0) /*Etl_Proc_Dt*/
,TO_DATE(:TXDATE,'YYYYMMDD') /*Etl_Tx_Dt*/
,Substr('s98_s_customer_event_meta',1,3) /*Src_Sysname*/
,'s98_s_customer_event_meta' /*Src_Table*/
FROM (select 'cust_contact_info' data_table,count(*) row_count ,max(etl_tx_dt) last_upt_tm , 51 columns_qty from p30_common.cust_contact_info
union all
select 'cust_contact_mapping' data_table,count(*) row_count ,max(etl_tx_dt) last_upt_tm , 13 columns_qty from p30_common.cust_contact_mapping
union all
select 'cust_leads' data_table,count(*) row_count ,max(etl_tx_dt) last_upt_tm , 10 columns_qty from p30_common.cust_leads
union all
select 'd_crm_contact' data_table,count(*) row_count ,max(etl_tx_dt) last_upt_tm , 65 columns_qty from p30_common.d_crm_contact
union all
select 'd_livechat_contact' data_table,count(*) row_count ,max(etl_tx_dt) last_upt_tm , 19 columns_qty from p30_common.d_livechat_contact
union all
select 'd_scrm_contact' data_table,count(*) row_count ,max(etl_tx_dt) last_upt_tm , 20 columns_qty from p30_common.d_scrm_contact
union all
select 'd_udesk_contact' data_table,count(*) row_count ,max(etl_tx_dt) last_upt_tm , 16 columns_qty from p30_common.d_udesk_contact
) p0
;
/*****************************************************************************************************/
/* GROUP 3:Source Table:s98_s_customer_event_meta*****************************************************/
/*****************************************************************************************************/
INSERT INTO data_source_update_agi_CUR_I (
date_layer /*数据层级*/
,data_area /*数据领域*/
,data_table /*数据表*/
,data_rows /*数据行数*/
,column_qty /*字段数*/
,last_upt_tm /*最后更新日期*/
,Etl_Batch_No /*作业批次号*/
,Etl_First_Dt /*最初入库时间*/
,Etl_Job /*作业名称*/
,Etl_Proc_Dt /*本次入库时间*/
,Etl_Tx_Dt /*作业运行时间*/
,Src_Sysname /*来源系统*/
,Src_Table /*来源表*/
)
SELECT
'MART' /*date_layer*/
,case when p0 .data_table like '%scrm%' then 'SCRM'
when p0.data_table like '%customer%' then 'SCRM'
when p0.data_table like '%crm%' then 'CRM'
when p0.data_table like '%livechat%' then 'LiveChat'
when p0.data_table like '%udesk%'
then 'Udesk'
else 'All' end /*data_area*/
,coalesce(p0.data_table) /*data_table*/
,coalesce(p0.row_count,0) /*data_rows*/
,coalesce(p0.columns_qty,0) /*column_qty*/
,p0.last_upt_tm /*last_upt_tm*/
,0 /*Etl_Batch_No*/
,TO_DATE(:TXDATE,'YYYYMMDD') /*Etl_First_Dt*/
,:ETLJOB /*Etl_Job*/
,current_timestamp(0) /*Etl_Proc_Dt*/
,TO_DATE(:TXDATE,'YYYYMMDD') /*Etl_Tx_Dt*/
,Substr('s98_s_customer_event_meta',1,3) /*Src_Sysname*/
,'s98_s_customer_event_meta' /*Src_Table*/
FROM (select 'cust_all_info' data_table,count(*) row_count ,max(etl_tx_dt) last_upt_tm , 33 columns_qty from p60_mart.cust_all_info ) p0
;
/*将不同数据插入到临时表 */
;INSERT INTO data_source_update_agi_INS (
data_area /*数据领域*/
,data_rows /*数据行数*/
,column_qty /*字段数*/
,last_upt_tm /*最后更新日期*/
,date_layer /*数据层级*/
,data_table /*数据表*/
,Etl_Batch_No /*作业批次号*/
,Etl_First_Dt /*最初入库时间*/
,Etl_Job /*作业名称*/
,Etl_Proc_Dt /*本次入库时间*/
,Etl_Tx_Dt /*作业运行时间*/
,Src_Sysname /*来源系统*/
,Src_Table /*来源表*/
)
SELECT
P1.data_area /*数据领域*/
,P1.data_rows /*数据行数*/
,P1.column_qty /*字段数*/
,P1.last_upt_tm /*最后更新日期*/
,P1.date_layer /*数据层级*/
,P1.data_table /*数据表*/
,P1.Etl_Batch_No /*作业批次号*/
,P1.Etl_First_Dt /*最初入库时间*/
,P1.Etl_Job /*作业名称*/
,P1.Etl_Proc_Dt /*本次入库时间*/
,P1.Etl_Tx_Dt /*作业运行时间*/
,P1.Src_Sysname /*来源系统*/
,P1.Src_Table /*来源表*/
FROM data_source_update_agi_CUR_I P1
LEFT JOIN :MARTDB.data_source_update P2
ON P1.data_area = P2.data_area
AND P1.data_rows = P2.data_rows
AND P1.column_qty = P2.column_qty
AND P1.last_upt_tm = P2.last_upt_tm
AND P1.date_layer = P2.date_layer
AND P1.data_table = P2.data_table
WHERE P2.data_area IS NULL
OR P2.data_rows IS NULL
OR P2.column_qty IS NULL
OR P2.last_upt_tm IS NULL
OR P2.date_layer IS NULL
OR P2.data_table IS NULL
;
/*将新增数据插入到目标表 */
;INSERT INTO :MARTDB.data_source_update (
data_area /*数据领域*/
,data_rows /*数据行数*/
,column_qty /*字段数*/
,last_upt_tm /*最后更新日期*/
,date_layer /*数据层级*/
,data_table /*数据表*/
,Etl_Batch_No /*作业批次号*/
,Etl_First_Dt /*最初入库时间*/
,Etl_Job /*作业名称*/
,Etl_Proc_Dt /*本次入库时间*/
,Etl_Tx_Dt /*作业运行时间*/
,Src_Sysname /*来源系统*/
,Src_Table /*来源表*/
)
SELECT
P1.data_area /*数据领域*/
,P1.data_rows /*数据行数*/
,P1.column_qty /*字段数*/
,P1.last_upt_tm /*最后更新日期*/
,P1.date_layer /*数据层级*/
,P1.data_table /*数据表*/
,P1.Etl_Batch_No /*作业批次号*/
,P1.Etl_First_Dt /*最初入库时间*/
,P1.Etl_Job /*作业名称*/
,P1.Etl_Proc_Dt /*本次入库时间*/
,P1.Etl_Tx_Dt /*作业运行时间*/
,P1.Src_Sysname /*来源系统*/
,P1.Src_Table /*来源表*/
FROM data_source_update_agi_INS P1
ON CONFLICT ( date_layer,data_table)
DO UPDATE SET
date_layer=excluded.date_layer
,data_table=excluded.data_table
,data_area=excluded.data_area
,data_rows=excluded.data_rows
,column_qty=excluded.column_qty
,last_upt_tm=excluded.last_upt_tm
,Etl_Batch_No=excluded.Etl_Batch_No
,Etl_First_Dt=excluded.Etl_First_Dt
,Etl_Job=excluded.Etl_Job
,Etl_Proc_Dt=excluded.Etl_Proc_Dt
,Etl_Tx_Dt=excluded.Etl_Tx_Dt
,Src_Sysname=excluded.Src_Sysname
,Src_Table=excluded.Src_Table
;
/*****程序结束退出 */
\q

11
dev/workflow/TK_Cust/tk_cust/泰克客户/wf_dag_tk_cust.py
View File

@ -220,6 +220,14 @@ params={'my_param':"cust_all_info_agi"},
depends_on_past=False,
retries=3,
dag=dag)
data_source_update = SSHOperator(
ssh_hook=sshHook,
task_id='data_source_update',
command='/data/airflow/etl/MART/run_psql.sh {{ ds_nodash }} {{params.my_param}} >>/data/airflow/logs/run_tpt_{{ds_nodash}}.log 2>&1 ',
params={'my_param':"data_source_update_agi"},
depends_on_past=False,
retries=3,
dag=dag)
file_Tk01 >> livechat_6381
file_Udesk_record >> udesk_record_3768
file_CRM_Raw_Leads >> crm_raw_leads_6024
@ -239,4 +247,5 @@ cust_contact_mapping >> cust_leads
t01_crm_raw_leads >> cust_leads
cust_leads >> cust_all_info
cust_contact_info >> cust_all_info
cust_all_info >> task_failed
cust_all_info >> data_source_update
data_source_update >> task_failed