服务好 售后好的网站建设,中铁建设门户网入口,成都设计公司工资多少,企业邮箱登录入口163心得的体会
刚过了年刚开工#xff0c;闲暇之余调研了分布式SQL流处理数据库–RisingWave#xff0c;本人是Flink#xff08;包括FlinkSQL和Flink DataStream API#xff09;的资深用户#xff0c;但接触到RisingWave令我眼前一亮#xff0c;并且拿我们生产上的监控告警…心得的体会
刚过了年刚开工闲暇之余调研了分布式SQL流处理数据库–RisingWave本人是Flink包括FlinkSQL和Flink DataStream API的资深用户但接触到RisingWave令我眼前一亮并且拿我们生产上的监控告警场景在RisingWave上做了验证以下是自己的心得体会
RisingWave架构简单运维成本底基于云原生可以分别基于计算和存储动态伸缩同时在开发上屏蔽了Flink等实时处理框架底层需要处理的一些技术细节状态存储数据一致性分布式集群扩展等。提供了与PostgreSQL兼容的标准SQL接口用户可以像使用 PostgreSQL 一样处理数据流。并且RisingWave不单单可以处理流式数据还提供了数据其他流式处理框架如Flink、storm所不具备的数据存储能力基本可以完全取代FlinkSQL。相对于其他OLAP系统如apache dorisstarrocksRisingWave采用同步实时可以保证实时的新鲜度强一致性而不是最终一致性。用户需要做的仅仅是通过开发SQL就可以处理流数据当然首先需要具备流式数据处理思维相对于离线。
RisingWave当然也有自身的不足相对于Flink可以通过DataStream API自定义灵活的处理流式数据RisingWave只能解决一些特定的流式场景无法做太多定制开发相对于Apache Doris等OLAP实时分析性数据库RisingWave不适合做分析型随机查询。另外RisingWave是个新事物正在发展阶段周边生态和相关文档还不健全作为尝鲜者可能会踩很多坑。然而令人欣慰的是RisingWave的社区回复还是很及时的RisingWave官方投入了很多精力在做RisingWave的布道和答疑。
至于争论比较厉害的RisingWav VS FLink的性能和吞吐量上孰优孰劣针对不同应用场景可能有不同表现因此没有亲自调研就没有发言权。但我认为在不同的场景下他们应该有各自的优势。无论如何RisingWave部署简单上手容易试错成本低是一个不争的事实。RisingWave可以应用在一些数据看版监控实时指标等场景。
利用动态和时间过滤器实现监控告警
FlinkSQL解决不了定时触发的问题FlinkSQL的流处理逻辑只是按event触发不能按时间条件触发也就是没有触发器机制。FlinkSQL窗口的定时触发归根结底也是基于event触发event驱动的机制。因此需要触发器的场景就需要用到Flink DataStream API的KeyedProcessFunction等算子。但RisingWave利用Dynamic filters 和 Temporal filters 可以间接实现类似场景的触发器机制。
场景描述
现有如下群消息实时指标监控场景 数据有初始化init、查询query、回调callback:succfail三种先后顺序状态。 数据是按预设时间批次分组的例如2024-01-01 08:00:00、2024-01-01 08:30:00实时统计每一个批次内三种不同状态的数据count。
监控指标一在某一个批次延迟指定的时间query_timeout之内例如2024-01-01 08:00:00延迟1小时触发时间为系统时间2024-01-01 09:00:00该批次的query状态数据count没有达到init状态的数量count阀值即query_countinit_count*query_threshold就触发告警。 同时结束该批次数据统计下发该批次数据的指标包括批次时间、init_count、query_count等
监控指标二如果指标一告警没有被触发该批次在满足query状态数据count达到init状态数量count的阀值即query_countinit_countquery_threshold以后,在指定的延迟时间内callback_timeout,该批次的callback状态数据count没有达到query状态的数量count阀值即callback_countquery_countcallback_threshold就触发告警。 同时结束该批次数据统计下发该批次数据的指标包括批次时间、init_count、query_count、callback_count等
群消息实时指标监控流程图如下
实例demo
RisingWave部署可以参考RisingWave分布式SQL流处理数据库调研
假设 query_threshold1, callback_count1 query_timeout ‘5 minute’, callback_timeout ‘1 minute’ 0-init,1-query,2-callback
RisingWave SQL DROP TABLE t_msg;
CREATE TABLE t_msg(msg_id int,status smallint ,public_time timestamp,process_time timestamp as proctime()
) APPEND ONLY;set timezone PRC;--PRCPeople’s Republic of China
show timezone;select * from t_msg;--统计不同状态的count
DROP MATERIALIZED VIEW mv_t_msg_groupby;
CREATE MATERIALIZED VIEW mv_t_msg_groupby AS
SELECT public_time
,sum(case when status 0 then 1 else 0 end) AS init_count
,sum(case when status 1 then 1 else 0 end) AS query_count
,sum(case when status 2 then 1 else 0 end) AS callback_count
,max(process_time) as process_time
FROM t_msg
group by public_time;select * from mv_t_msg_groupby;--sink_query_alarm
DROP SINK sink_query_alarm;
CREATE SINK sink_query_alarm AS
SELECT public_time
,init_count
,query_count
,process_time
FROM mv_t_msg_groupby
where public_time INTERVAL 5 minute now() --query_timeout1 minute, public_time相当于当前时间延迟1分钟触发【Delay table changes】
and init_count*1 query_count --query_threshold1
WITH (connectorkafka,properties.bootstrap.server192.168.1.100:8092,topict_sink_query_alarm
)
FORMAT PLAIN ENCODE JSON(force_append_onlytrue
);--由于RisingWave不支持在【MATERIALIZED VIEW】和【SINK】等【可伸缩流】中指定处理时间字段因此需要借助外部存储kafka周转
--RisngWave官方给的解释support a proctime on an append only stream might be easier but on retractable stream could take extra cost. We must think it carefully to introduce such a feature.--sink_query_succ
DROP SINK sink_query_succ;
CREATE SINK sink_query_succ AS
SELECT public_time
,init_count
,query_count
,callback_count
,process_time as query_succ_process_time
FROM mv_t_msg_groupby
where public_time INTERVAL 5 minute now() --query_timeout1 minute, 在指定的时间内【Delete and clean expired data】
and init_count*1 query_count --query_threshold1,query_count达到了指定值
WITH (connectorkafka,properties.bootstrap.server192.168.1.100:8092,topict_sink_query_succ
)
FORMAT PLAIN ENCODE JSON(force_append_onlytrue
);--source连接器
DROP SOURCE source_query_succ;
CREATE SOURCE IF NOT EXISTS source_query_succ (init_count int,query_count int ,callback_count int ,public_time timestamp,query_succ_process_time timestamp
)
WITH (connectorkafka,topict_sink_query_succ,properties.bootstrap.server192.168.1.100:8092,scan.startup.modeearliest, -- earliest ,latest,default:earliest
) FORMAT PLAIN ENCODE JSON;select * from source_query_succ;--sink_callback_alarm用到动态过滤器和时间过滤器
DROP SINK sink_callback_alarm;
CREATE SINK sink_callback_alarm AS
WITH tmp AS (
select public_time, min(query_succ_process_time) as query_succ_process_time -- 动态过滤器
FROM source_query_succ
group by public_time
)
SELECT b.public_time
,b.init_count
,b.query_count
,b.callback_count
,b.process_time
,a.query_succ_process_time
FROM tmp a
JOIN mv_t_msg_groupby b ON a.public_timeb.public_time
where a.query_succ_process_time INTERVAL 1 minute now() --query_timeout1 minute, public_time相当于当前时间延迟1分钟触发【Delay table changes】
and b.query_count*1 b.callback_count --callback_threshold1
WITH (connectorkafka,properties.bootstrap.server192.168.1.100:8092,topict_sink_callback_alarm
)
FORMAT PLAIN ENCODE JSON(force_append_onlytrue
);-- 模拟数据
--init
INSERT INTO t_msg values(1,0,2024-02-23 15:55:00::TIMESTAMP); --比当前系统时间早
INSERT INTO t_msg values(2,0,2024-02-23 15:55:00::TIMESTAMP);
INSERT INTO t_msg values(3,0,2024-02-23 15:55:00::TIMESTAMP);
--query
INSERT INTO t_msg values(1,1,2024-02-23 15:55:00::TIMESTAMP);
INSERT INTO t_msg values(2,1,2024-02-23 15:55:00::TIMESTAMP);
INSERT INTO t_msg values(3,1,2024-02-23 15:55:00::TIMESTAMP);
--callback
INSERT INTO t_msg values(1,2,2024-02-23 15:55:00::TIMESTAMP);
INSERT INTO t_msg values(2,2,2024-02-23 15:55:00::TIMESTAMP);
INSERT INTO t_msg values(3,2,2024-02-23 15:55:00::TIMESTAMP);查看监控结果
docker run -it --rm edenhill/kcat:1.7.1 kcat -b 192.168.1.100:8092 -t t_sink_query_alarm -C
docker run -it --rm edenhill/kcat:1.7.1 kcat -b 192.168.1.100:8092 -t t_sink_callback_alarm -C
