數據流程介紹

1.創建源表kafka接入消息隊列數據，定義字段映射規則；
2.創建目標表es_sink配置Elasticsearch輸出；
3.通過多級視圖（tmp→tmp_dedup→tmp1/tmp2→tmp3→tmp_groupby）實現數據清洗、去重、狀態計算；
4.使用ROLLUP進行多維聚合統計；
5.最終計算結果寫入ES，包含成功率等衍生指標。

Flink SQL 邏輯

SET table.exec.state.ttl=2592000s; --30 days,默認: 0 ms
--MiniBatch 聚合
SET table.exec.mini-batch.enabled = true;
SET table.exec.mini-batch.allow-latency = 1s;
SET table.exec.mini-batch.size = 10000;
--Local-Global 聚合
SET table.optimizer.agg-phase-strategy = TWO_PHASE;
-- 單位：ms, 10天
--SET table.exec.state.ttl = 864000000CREATE TABLE kafkaTable (mid bigint,db string,sch string,tab string,opt string,ts bigint,ddl string,err string,src map<string,string>,cur map<string,string>,cus map<string,string>,id AS IF(cur['id'] IS NOT NULL , cur['id'], src ['id']),task_id AS IF(cur['task_id'] IS NOT NULL , cur['task_id'], src ['task_id']),account_id AS IF(cur['account_id'] IS NOT NULL , cur['account_id'], src ['account_id']),publish_time AS IF(cur['publish_time'] IS NOT NULL , cur['publish_time'], src ['publish_time']),msg_status AS IF(cur['msg_status'] IS NOT NULL , cur['msg_status'], src ['msg_status']),send_type AS IF(cur['send_type'] IS NOT NULL , cur['send_type'], src ['send_type']),retry_status AS IF(cur['retry_status'] IS NOT NULL , cur['retry_status'], src ['retry_status']),update_time as IF(cur['update_time'] IS NOT NULL , cur['update_time'], src ['update_time']),event_time as cast(IF(cur['update_time'] IS NOT NULL , cur['update_time'], src ['update_time']) AS TIMESTAMP(3)), -- TIMESTAMP(3)/TIMESTAMP_LTZ(3)proctime AS PROCTIME()
--                           WATERMARK FOR event_time AS event_time - INTERVAL '1' MINUTE     --SECOND
) WITH ('connector' = 'kafka','topic' = 'xxx','jdq.client.id' = 'xxx','jdq.password' = 'xxx','jdq.domain' = 'xxx','scan.startup.mode' = 'group-offsets', --  default: group-offsets,other: latest-offset,earliest-offset--  'properties.enable.auto.commit',= 'true' -- default:false, 如果為false，則在發生checkpoint時觸發offset提交'format' = 'binlog');CREATE TABLE es_sink(send_type      STRING,task_id        STRING,month_dim      STRING,day_dim        STRING,grouping_id    INTEGER,init           INTEGER,cancel         INTEGER,succ           INTEGER,fail           INTEGER,cancel_rate    float,succ_rate      float,fail_rate      float,update_date    STRING,PRIMARY KEY (grouping_id,send_type,month_dim,day_dim,task_id) NOT ENFORCED
)with ('connector' = 'elasticsearch-6','index' = 'index01','document-type' = 'type01','hosts' = 'xx','format' = 'json','filter.null-value'='true','sink.bulk-flush.max-actions' = '1000','sink.bulk-flush.max-size' = '10mb');
-- 維度:
--   - send_type, 發送類型
--   - month_dim,月份維度
--   - day_dim,天維度
--   - task_id,任務IDCREATE view  tmp as
selectsend_type,task_id,publish_time,msg_status,case when UPPER(opt) = 'INSERT' and msg_status='0'  then 1 else 0 end AS init,case when UPPER(opt) = 'UPDATE' and msg_status='4' then 1 else 0 end AS cancel,case when UPPER(opt) = 'UPDATE' and msg_status='1' then 1 else 0 end AS succ,case when UPPER(opt) = 'UPDATE' and msg_status='2' then 1 else 0 end AS fail,update_time,opt,ts,id,proctime,SUBSTRING(publish_time,1,7) as month_dim,SUBSTRING(publish_time,1,10) as day_dim
FROM kafkaTable
where trim(retry_status) = '0'and publish_time >= '2025-01-01 00:00:00'and(    (UPPER(opt) = 'INSERT' and msg_status='0' and position( '_R' in task_id) = 0)or   (UPPER(opt) = 'UPDATE' and msg_status in ('1','2','3','4') and position( '_R' in task_id) = 0)or   (UPPER(opt) = 'UPDATE' and msg_status='1' and position( '_R' in task_id) > 0));--去重模式，去重是指對在列的集合內重復的行進行刪除，只保留第一行或最后一行數據。在聚合sum或count時，Flink回撤流會對數據進行回撤處理
create view tmp_dedup as
select * from(select *,row_number() over(partition by id,msg_status order by proctime desc) as rnfrom tmp) t
where rn=1;CREATE view tmp1 as
selectsend_type,task_id,month_dim,day_dim,init,case when cancel = 1 and update_time <= publish_time then 1 else 0 end AS cancel,succ,case when cancel = 1 and update_time > publish_time then 1 else fail end AS fail,update_time
from tmp_dedup
where position( '_R' in task_id) = 0;CREATE view tmp2 as
selectsend_type,SPLIT_INDEX(task_id,'_R',0) AS task_id,month_dim,day_dim,init,cancel,succ,-1 AS fail,update_time
from tmp_dedup
where position( '_R' in task_id) > 0
and   succ = 1 ;CREATE view tmp3 as
selectsend_type,task_id,month_dim,day_dim,init,cancel,succ,fail
from tmp1
UNION ALL
selectsend_type,task_id,month_dim,day_dim,init,cancel,succ,fail
from tmp2;CREATE view  tmp_groupby as
select
--/*+ STATE_TTL('tmp' = '10d') */COALESCE(send_type,'N') AS send_type,COALESCE(month_dim,'N') AS month_dim,COALESCE(day_dim,'N') AS day_dim,COALESCE(task_id,'N') AS task_id,case when send_type is null and month_dim is null and day_dim is null and task_id is null then 1when send_type is not null and month_dim is null and day_dim is null and task_id is null then 2when send_type is not null and month_dim is not null and day_dim is null and task_id is null then 3when send_type is not null and month_dim is not null and day_dim is not null and task_id is null then 4when send_type is not null and month_dim is not null and day_dim is not null and task_id is not null then 5end grouping_id,sum(init) as init,sum(cancel) as cancel,sum(succ) as succ,sum(fail) as fail
from tmp3
--GROUP BY GROUPING SETS ((send_type,account_id,publish_time), (send_type,account_id),(send_type), ())
GROUP BY ROLLUP (send_type,month_dim,day_dim,task_id); --等同于以上INSERT INTO es_sink
selectcase when trim(send_type) = '1'  then '發送類型1'when trim(send_type) = '2'  then '發送類型2'else send_type end AS send_type,task_id,month_dim,day_dim,grouping_id,init,cancel,succ,fail,ROUND(cancel*100.0/init,2) AS cancel_rate,ROUND(succ*100.0/(init - cancel),2) AS succ_rate,ROUND(fail*100.0/(init - cancel),2) AS fail_rate,CAST(LOCALTIMESTAMP AS STRING) as update_date
from tmp_groupby
where init > 0
and (init - cancel) > 0;

es mapping

#POST index01/type01/_mapping
{"type01": {"properties": {"grouping_id": {"type": "byte"},"send_type": {"type": "keyword","ignore_above": 256},"month_dim": {"type": "keyword","fields": {"text": {"type": "keyword"},"date": {"type": "date","format": "yyyy-MM","ignore_malformed":"true" --忽略錯誤的各式}}},"day_dim": {"type": "keyword","fields": {"text": {"type": "keyword"},"date": {"type": "date","format": "yyyy-MM-dd","ignore_malformed":"true"}}},"task_id": {"type": "keyword"},"init": {"type": "integer"},"cancel": {"type": "integer"},"succ": {"type": "integer"},"fail": {"type": "integer"},"cancel_rate": {"type": "float"},"succ_rate": {"type": "float"},"fail_rate": {"type": "float"},"update_date": {"type": "date","format": "yyyy-MM-dd HH:mm:ss.SSS||yyyy-MM-dd HH:mm:ss||yyyy-MM-dd||epoch_millis"}}}
}