当前位置： 首页 > wzjs >正文

网站建设模板一次收费全国疫情最新情况公布

wzjs 2025/7/25 21:38:20
网站建设模板一次收费,全国疫情最新情况公布,一个小型网站开发成本,网上赚钱的app数据流程介绍 1.创建源表kafka接入消息队列数据，定义字段映射规则； 2.创建目标表es_sink配置Elasticsearch输出； 3.通过多级视图（tmp→tmp_dedup→tmp1/tmp2→tmp3→tmp_groupby）实现数据清洗、去重、状态计算&#x…

数据流程介绍

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"}}}
}
http://www.dtcms.com/wzjs/72851.html

相关文章：

  • 建设网站交纳党费百度小说排行榜
  • 网站分屏布局设计竞价推广的企业
  • 网站建设百度推广商品推广
  • 南通做网站ntwsd廊坊seo排名公司
  • 广州建网站价格外贸网站建设
  • 杭州建设局网站首页品牌推广的方式
  • 最好用的素材网站网页设计需要学什么
  • html制作音乐网站营销型网站的分类不包含
  • 百度的网站名济南seo怎么优化
  • 外国人学做中国菜 网站济南网站seo优化
  • 网站建设名词免费b2b网站推广
  • 微信上的小说网站是怎么做的北京百度seo服务
  • 自己建个电影网站可以吗推客平台
  • 局网站建设工作总结软文关键词排名推广
  • 强的小企业网站建设百度热门搜索排行榜
  • 如何制作餐馆网站杭州网站优化公司哪家好
  • 网站开发怎么切换多种语言快速排名优化推广手机
  • 自己做的网站怎么给别人访问成人职业技能培训班
  • 乐东黎族自治县住房建设局网站竞价排名点击
  • 山西省建设厅网站官网如何建立自己的博客网站
  • 手机网站设计框架怎样有效的做网上宣传
  • 网站建设的id调用怎么操作核心关键词和长尾关键词举例
  • 网站建设需要个体营业执照吗电商培训学校
  • 手机制作网站免费大型集团网站建设公司
  • photoshop做网站广州各区最新动态
  • 做移动端网站软件下载搜狗官网
  • 用java做的网站有哪些内容友情链接交换网
  • 做ui设计工资一般多少深圳关键词优化平台
  • 校园互动网站建设浏览器下载安装2022最新版
  • 域名备案和网站备案是一回事吗软件测试培训机构哪家好