sql语句如下:
select count(clicknum) as num
from table(
cumulate(table testTable, desctiptor(crtTime),interval '1'minutes, interval
'10' minutes))
where clicknum <>'-99'
group by window_start,window_end
报错 信息:
Flink doesn't support individual window table-valued function
cumulate(time_col=[app_
at
org.apache.flink.client.deployment.application.DetachedApplicationRunner.tryExecuteJobs(DetachedApplicationRunner.java:84)
~
在 2021-08-11 12:44:38,"Caizhi Weng" 写道:
>Hi!
>
>descriptor 拼错了吧。我在本地没有复现这个问题,Flink 版本是多少呢?
>
>李航飞 于2021年8月11日周三 上午11:41写道:
>
>> sql语句如下:
>> s
抱歉,sql语句是我手打的,没注意到,我确定和这个descriptor没关系。我去掉where条件就能正常运行,同时,我测试in,not in
函数的时候,会报同样的错误。
At 2021-08-11 13:51:16, "李航飞" wrote:
>org.apache.flink.client.deployment.application.DetachedApplicationRunner [] -
>Could not execute application:
>org.apache.flink.client.program.Program
通过flinksql建立数据处理通道
SELECT window_start,window_end,SUM(price)
FROM TABLE(
CUMULATE(TABLE Bid,DESCRIPTOR(bidtime),INTERVAL '2' MINUTES,INTERVAL '10'
MINUTES))
GROUP BY window_start,window_end;
大致语句如上,该语句通过 StreamTableEnvironment对象 env.sqlQuery(sql)执行成功,没有问题
关键一步是 StatementSet对象 sta.execute() 执行报
哦哦,好的,谢谢你。比较函数不可以连用,简单的一些的条件(where a = '2')可以用,这个功能后续还会调整吗
在 2021-08-18 16:21:20,"Caizhi Weng" 写道:
>Hi!
>
>目前 window tvf 只能应用于 window agg 和 window top-n 两种场景。如果 where 条件是用来对 window
>agg 的结果进行过滤的,可以使用 having 而不是 where;若是对进入 window 之前的数据进行过滤的,可以 create view。
>
&g
版本 flink1.13.2
具体场景
flink-connector-redis自定义连接器,在实现DynamicTableSink接口时,
通过重写getChangelogMode方法,设置ChangelogMode.upsert()模式,以更新的方式进行数据写入,
使用的窗口时cumulate函数 开1天的窗口,小窗口1分钟计算一次
问题:
测试发现,每1分钟都会输出一次,落地的数据一样,
根据upsert()模式,正常来说一样的数据应该不会触发redis存储,这个是什么情况?
置了 state ttl 的缘故。为了防止数据过期,group agg 就算每次收到同样的 key
>value,仍然会往下游发数据(详见 GroupAggFunction#processElement),所以这是一个预期行为。
>
>李航飞 于2021年8月19日周四 下午5:03写道:
>
>> 版本 flink1.13.2
>> 具体场景
>> flink-connector-redis自定义连接器,在实现DynamicTableSink接口时,
>> 通过重写getChangelogMode方法,设置Changelo
你好:
我查到 Sink在Streaming场景 只能接收insert-only ,是这样吗?
在 2021-08-20 09:10:44,"李航飞" 写道:
>你好:
>我确实设置了过期时间,我将过期时间的代码注释了,但还是每分钟落地一次。
>
>
>我在RichMapFunction接口里面实现open方法
>设置过StateTtlConfig;
>之后在RedisConmmand.SETEX设置过期时间
>都注释了,但upsert()方法还是没效
>
>
>
>
>
>
你好:
具体场景是对agg结果之前进行过滤,现在通过create view进行提取过滤了
现在我想通过DynameicTable的方式,以upsert写入redis里面
在 2021-08-20 10:31:18,"Caizhi Weng" 写道:
>Hi!
>
>具体的条件是什么样的呢?理论上如果是和原本的 group by ... having ... 等价的语句应该是支持的。
>
>李航飞 于2021年8月18日周三 下午4:34写道:
>
>> 哦哦,好的,谢谢你。比较函数不可以连用,简单的一些的条件(
arg0){
return ChangelogMode.upsert();
}
实现RedisMapper 方法 落地redis 有输出语句,每1分钟都会落地一次,我确定数据每次都一样
这upsert 不合理啊
在 2021-08-20 11:15:17,"Caizhi Weng" 写道:
>Hi!
>
>之前没注意到是 cumulate 窗口,具体的 group by 语句是什么样的呢?描述里写着“小窗口 1 分钟计算一次”,那 sink
>确实应该每分钟收到一条消息。
>
>sink 在 streaming 场景可以接收 upsert
能不能通过设置,每分钟只输出 当前小窗口中计算数据的结果,之前计算的结果不再输出
目前测试发现,输入的数据越多,到下次输出的数据也会越来越多,
不同窗口的计算结果,都会再下次窗口中输出,
你好:
我使用的场景是要实时统计一天的数据,在小窗口进行即时输出,延迟不能太高,cumulate window符合要求,tumble window 延迟太高了。
在 2021-08-20 16:01:57,"Caizhi Weng" 写道:
>Hi!
>
>你可能想要的是 tumble window 而不是 cumulate window。
>
>李航飞 于2021年8月20日周五 下午3:26写道:
>
>> 能不能通过设置,每分钟只输出 当前小窗口中计算数据的结果,之前计算的结果不再输出
>>
&
通过Configuration 设置
table.exec.mini-batch.enabled= true;
table.exec.mini-batch.allow-latency = 15s;
table.exec.mini-batch.size = 50;
上面的配置测试了,没效果,下面的测试
table.exec.mini-batch.allow-latency = 15000
也没效果什么原因?
设置的延迟15秒,实际测试发现数据来了之后就处理了,没有延迟15秒
在 2021-08-25 11:12:19,"Caizhi Weng" 写道:
>Hi!
>
>所谓的没效果指的是什么现象呢?建议详细描述一下场景与问题。
>
>李航飞 于2021年8月25日周三 上午11:04写道:
>
>> 通过Configuration 设置
>> table.exec.mini-batch.enabled= true;
>> table.exec.mini-batch.allow-lat
Configuration conf = new Configuration();
conf.setString("table.exec.mini-batch.enabled","true");
conf.setString("table.exec.mini-batch.allow-latency","15s");
conf.setString("table.exec.mini-batch.size","50");
StreamExecutionEnvironment execEnv =
StreamExecutionEnvironment.getExecutionEnvironm
Configuration conf = new Configuration();
conf.setString("table.exec.mini-batch.enabled","true");
conf.setString("table.exec.mini-batch.allow-latency","15s");
conf.setString("table.exec.mini-batch.size","50");
conf.setString("table.exec.state.ttl","15 s");StreamExecutionEnvironment
execEnv = Strea
environment 的 table config 里的,加在 execution environment
>里无效。
>
>李航飞 于2021年8月26日周四 下午7:02写道:
>
>> Configuration conf = new Configuration();
>> conf.setString("table.exec.mini-batch.enabled","true");
>> conf.setString("table.exec.mini-batch.al
滚动窗口最小可开多大,100ms?
对性能有什么影响吗?
退订
现在有一个需求是,创建一个任务,消费kafka,仅消费一个片段 即设定起始消费点和结束消费位置
我看到DataStream Connectors kafka 中有一个setBounded (setUnbounded)属性 ,可以满足需求。
问题;
我想使用 Table API 完成上面的需求,该怎么办?
Table API 是否有相关属性?
有其他办法满足这个需求吗?
流处理批处理都行。
我想通过Table API 的方式消费kafka
需要设定消费kafka 开始位置(时间)和结束位置(时间)
请问有没有相关属性 例如DataStream中的setUnbounded属性,
有没有其他办法可以实现类似功能?
我想通过Table API 的方式消费kafka
需要设定消费kafka 开始位置(时间)和结束位置(时间)
请问有没有相关属性 例如DataStream中的setUnbounded属性,
有没有其他办法可以实现类似功能?
22 matches
Mail list logo