Re: blinkSQL架构会自动清理过期的state吗

[LakeShen]

Hi 守护,
*可*以先确保你的watermark是否一直在更新。
然后可以把窗口时间间隔设置小一点，然后在运行程序，看看checkpoint 的状态文件是否会自动清理。



守护 <346531...@qq.com> 于2019年9月12日周四 下午2:35写道：

> 下面是我要执行的代码，麻烦帮看一下：
>
>
>     tableEnv.registerDataStream("testCountTable",
> waterMarkStream, 'curuserid,'timelong,'rowtime.rowtime)
>     
>     val result = tableEnv.sqlQuery(s"SELECT COUNT(0) as
> pv,COUNT(distinct curuserid)" +
>       s" as uv,TUMBLE_END(rowtime, INTERVAL '10' MINUTE)
> FROM testCountTable GROUP BY TUMBLE(rowtime, INTERVAL '10' MINUTE)")
>
>
>     val dsRow: DataStream[Row] =
> tableEnv.toAppendStream[Row](result)
>
>
>     val data=dsRow.map(w => {
>       val StrArrary = w.toString.split(",")
>       val
> str="{\"pv\":"+"\""+StrArrary(0)+"\""+",\"uv\":"+"\""+StrArrary(1)+"\""+",\"rowtime\":"+"\""+StrArrary(2)+"\""+"}"
>       str
>     })
>     data.print()
>
>
>
>
>
> -- 原始邮件 --
> 发件人: "Jark Wu" 发送时间: 2019年9月12日(星期四) 中午12:51
> 收件人: "user-zh"
> 主题: Re: blinkSQL架构会自动清理过期的state吗
>
>
>
> Hi，
> 能提供下 SQL 么？
>
> blink sql 的 window 理论上是会自动清理的。
>
> > 在 2019年9月11日，18:56，守护 <346531...@qq.com> 写道：
> >
> > 社区各位大佬：
> > 请教一个问题，flink1.9中使用blink
> SQL语句，设置时间窗口，state存储方式选择FSStateBackend，现在发现State一直在增大，过了窗口后也没有删除过期state，是blink架构就不支持窗口自动清理state吗，还是我哪使用的不对，我测试1.9的flinkSQL是不会有这个问题的。

?????? blinkSQL????????????????????state??

[????]

 
    
 blinkSQL??watermarkcheckpointflinkSQLblinkSQL??






--  --
??: "LakeShen"

Re: 如何优化flink内存?

[cai yi]

你好, 没有明白"对每条数据使用slideWindow得出这条数据对应的90天的窗口结束时间"这一步是怎么做的, 
滑动窗口中一条数据最大可以包含在size/slide个window中,所以这个结束时间是怎么取值的呢?

在 2019/9/5 下午4:08，“陈赋赟” 写入:

HI
  我在项目中有遇到过类似的情况，我说下我的想法和思路。
  伊始是需要统计90天事件窗口中用户浏览事件总数，如果是在近30天内有浏览事件则累加1次，在30天内没有浏览事件但在 30天 ～ 
90天内有其他浏览事件则记0次（需求比较奇葩），我们使用了滑动窗口（长度90天 步长1天 
数据进来实时trigger触发计算）因为需要拿到窗口的结束时间所以一开始是用windowProcessFunction去做的聚合统计，这意味着90个窗口每个窗口里都需要缓存着全部的数据而不是一个聚合汇总数据，在线上跑了两天后发现checkpoint
 size已经陡增到20个G并且不久就OOM了。后面想了一下，Flink 
提供的SlideWindow的算法不是闭包可以直接复用，用flatmap对每条数据使用slideWindow得出这条数据对应的90天的窗口结束时间，然后在keyby后使用ProcessFunction，在里面自定义valueState对数据进行聚合汇总，并且在processFunction内部还可以访问TimeService，可以注册清理过期state数据的Timer，并在onTimer回调方法中清理状态。
 以上是我的思路，希望能帮助到你～




祝好





在 2019-09-05 13:43:00，"Yifei Qi"  写道：
>你的意思是自己去实现滑动窗口的功能么?
>
>戴嘉诚  于2019年9月4日周三 下午10:51写道：
>
>> 这里我建议你是用key process然后在里面用state来管理和聚集数据，这样会节省很大一部分内存
>>
>> Yifei Qi 于2019年9月4日 周三20:07写道：
>>
>> > 大家好:
>> >
>> >
>> >
>> > 不知道大家在使用flink时遇到过内存消耗过大的问题么?
>> >
>> >
>> >
>> > 我们最近在用flink算一些实时的统计数据, 但是内存消耗很大, 不知道有没有人知道如何优化?
>> >
>> >
>> >
>> > 具体情况是这样的:
>> >
>> > 准备的测试数据模拟一天时间内3万个用户的5万条数据. 原始数据一共是100M.
>> >
>> > 按照用户进行分组.
>> >
>> > 计算两个滑动窗口任务:一个是近1小时, 每5秒滑动一次的窗口. 一个是近24小时, 每1分钟滑动一次的窗口.
>> >
>> >
>> >
>> >
>> >
>> > flink运行在3个节点后, 内存合计就用了5G.
>> >
>> >
>> >
>> >
>> >
>> > flink如此消耗内存, 不知道是它本来就这么吃内存, 还是我使用的有问题.
>> >
>> >
>> >
>> >
>> >
>> > 顺祝商祺
>> >
>> >
>> > --
>> >
>> >
>> > Qi Yifei
>> > [image: https://]about.me/qyf404
>> > <
>> >
>> 
https://about.me/qyf404?promo=email_sig&utm_source=product&utm_medium=email_sig&utm_campaign=gmail_api
>> > >
>> >
>>
>
>
>-- 
>
>
>Qi Yifei
>[image: https://]about.me/qyf404

>