You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user-zh@flink.apache.org by "gygzljy@163.com" <gy...@163.com> on 2021/12/07 10:38:33 UTC

flink sql支持细粒度的状态配置

Hi all

在我们生产中发现,如果在sql中配置状态的TTL会导致这个 ttl时间全局生效

如果我存在一个如下sql

select count(1),region from (select * from A join B on a.uid = b.uid)  group by region

如果我配置一个全局的TTL会导致count这个GroupAggFunction的状态被淘汰掉,比如说一天以后累计就被清零

如果不配置,又会导致Regular join的状态增大

这是其中一个场景,这里只是举一个例子

主要是想询问针对 Sql中需要配置局部State的ttl时间,或者同一个任务每个sql配置不同的TTL时间,这种场景应该如何去做 ?



gygzljy@163.com

回复:flink1.10.0-RestClusterClient cancel job, 报错

Posted by nicygan <re...@163.com>.
时间设长没有用,不限时都会报错。



---- 回复的原邮件 ----
| 发件人 | 何凯飞<18...@163.com> |
| 日期 | 2021年12月09日 15:07 |
| 收件人 | user-zh@flink.apache.org<us...@flink.apache.org> |
| 抄送至 | |
| 主题 | Re: flink1.10.0-RestClusterClient cancel job, 报错 |
有尝试过将 timeout 时间设置长一点嘛? 比如3 min



Re: flink1.10.0-RestClusterClient cancel job, 报错

Posted by 何凯飞 <18...@163.com>.
有尝试过将 timeout 时间设置长一点嘛? 比如3 min 




flink1.10.0-RestClusterClient cancel job, 报错

Posted by nicygan <re...@163.com>.
dear all:
      如题,我在调用RestClusterClient#cancel(JobID jobId)方法取消作业时,get不到结果,但作业能正常停止。

用future.get()会报错如下:
Number of retries has been exhausted.

用future.get(10, TimeUnit.SECONDS)会报错timeout.

调用#cancelWithSavepoint(...)和#stopWithSavepoint(...)就没问题,可以正常获取到结果,不报错。


作业运行于cdh yarn集群,版本2.6.0
作业部署,per-job

代码如下:
try (ClusterClient<?> clusterClient = new RestClusterClient<>(configuration, clusterId)) {
     clusterClient
          .cancel(jobId)
          .get(20, TimeUnit.SECONDS)
} catch (Exception e) {
 //
}

有谁知道如何解决此问题吗?
非常感谢!




---- 回复的原邮件 ----
| 发件人 | Yun Tang<my...@live.com> |
| 日期 | 2021年12月09日 10:57 |
| 收件人 | user-zh<us...@flink.apache.org> |
| 抄送至 | |
| 主题 | Re: flink sql支持细粒度的状态配置 |
Hi 你好,

我认为这是一个很好的需求,对于data stream以及python API来说,state TTL都是通过API逐个配置的,你的需求就可以直接满足。但是对于SQL来说,由于相同的SQL语句,不同优化器其生成的执行plan可能会差异很大,很难对某个operator内的state进行TTL进行配置,可能一种方式是增加一些SQL的优化hint,对于你示例中的join语句和groupBy 的count语句配以不同的TTL,但是目前Flink SQL尚未支持该功能。


祝好
唐云

________________________________
From: gygzljy@163.com <gy...@163.com>
Sent: Tuesday, December 7, 2021 18:38
To: user-zh <us...@flink.apache.org>
Subject: flink sql支持细粒度的状态配置

Hi all

在我们生产中发现,如果在sql中配置状态的TTL会导致这个 ttl时间全局生效

如果我存在一个如下sql

select count(1),region from (select * from A join B on a.uid = b.uid)  group by region

如果我配置一个全局的TTL会导致count这个GroupAggFunction的状态被淘汰掉,比如说一天以后累计就被清零

如果不配置,又会导致Regular join的状态增大

这是其中一个场景,这里只是举一个例子

主要是想询问针对 Sql中需要配置局部State的ttl时间,或者同一个任务每个sql配置不同的TTL时间,这种场景应该如何去做 ?



gygzljy@163.com

Re: Re: flink sql支持细粒度的状态配置

Posted by Yun Tang <my...@live.com>.
Hi,

如果你们可以自己实现一套SQL语句到jobgraph的预编译转换IDE,然后在IDE中可以手动配置jobgraph每个算子的配置,应该是可以达到你们的目的 (可能还需要结合细粒度调度模式)。

祝好
唐云
________________________________
From: gygzljy@163.com <gy...@163.com>
Sent: Thursday, December 9, 2021 16:14
To: user-zh <us...@flink.apache.org>
Subject: 回复: Re: flink sql支持细粒度的状态配置

Hi Yun Tang

感谢你的回复,我们在调研的过程中也发现,正如你所说的生成的plan可能差异很大

但是每个operator的TTL生效时间是在execNode转换成对应的Transformation时,通过传入的StreamPlanner带进去的,TableConfig属性中包含了全局的TTL时间

在每个ExecNode转换的过程translateToPlanInternal((PlannerBase) planner)中使用这个TTL时间生成对应的operator

所以我们在考虑是否可以在,每个Transformation生成阶段,先去修改一下TableConfig中TTL的配置再调用每个execNode转换成operator的方法,来做到Transformation级别的TTL控制,这个配置开放给平台的用户,通过Transformation的id做识别,是否能给一些建议




gygzljy@163.com

发件人: Yun Tang
发送时间: 2021-12-09 10:57
收件人: user-zh
主题: Re: flink sql支持细粒度的状态配置
Hi 你好,

我认为这是一个很好的需求,对于data stream以及python API来说,state TTL都是通过API逐个配置的,你的需求就可以直接满足。但是对于SQL来说,由于相同的SQL语句,不同优化器其生成的执行plan可能会差异很大,很难对某个operator内的state进行TTL进行配置,可能一种方式是增加一些SQL的优化hint,对于你示例中的join语句和groupBy 的count语句配以不同的TTL,但是目前Flink SQL尚未支持该功能。


祝好
唐云

________________________________
From: gygzljy@163.com <gy...@163.com>
Sent: Tuesday, December 7, 2021 18:38
To: user-zh <us...@flink.apache.org>
Subject: flink sql支持细粒度的状态配置

Hi all

在我们生产中发现,如果在sql中配置状态的TTL会导致这个 ttl时间全局生效

如果我存在一个如下sql

select count(1),region from (select * from A join B on a.uid = b.uid)  group by region

如果我配置一个全局的TTL会导致count这个GroupAggFunction的状态被淘汰掉,比如说一天以后累计就被清零

如果不配置,又会导致Regular join的状态增大

这是其中一个场景,这里只是举一个例子

主要是想询问针对 Sql中需要配置局部State的ttl时间,或者同一个任务每个sql配置不同的TTL时间,这种场景应该如何去做 ?



gygzljy@163.com

回复: Re: flink sql支持细粒度的状态配置

Posted by "gygzljy@163.com" <gy...@163.com>.
Hi Yun Tang

感谢你的回复,我们在调研的过程中也发现,正如你所说的生成的plan可能差异很大

但是每个operator的TTL生效时间是在execNode转换成对应的Transformation时,通过传入的StreamPlanner带进去的,TableConfig属性中包含了全局的TTL时间

在每个ExecNode转换的过程translateToPlanInternal((PlannerBase) planner)中使用这个TTL时间生成对应的operator

所以我们在考虑是否可以在,每个Transformation生成阶段,先去修改一下TableConfig中TTL的配置再调用每个execNode转换成operator的方法,来做到Transformation级别的TTL控制,这个配置开放给平台的用户,通过Transformation的id做识别,是否能给一些建议




gygzljy@163.com
 
发件人: Yun Tang
发送时间: 2021-12-09 10:57
收件人: user-zh
主题: Re: flink sql支持细粒度的状态配置
Hi 你好,
 
我认为这是一个很好的需求,对于data stream以及python API来说,state TTL都是通过API逐个配置的,你的需求就可以直接满足。但是对于SQL来说,由于相同的SQL语句,不同优化器其生成的执行plan可能会差异很大,很难对某个operator内的state进行TTL进行配置,可能一种方式是增加一些SQL的优化hint,对于你示例中的join语句和groupBy 的count语句配以不同的TTL,但是目前Flink SQL尚未支持该功能。
 
 
祝好
唐云
 
________________________________
From: gygzljy@163.com <gy...@163.com>
Sent: Tuesday, December 7, 2021 18:38
To: user-zh <us...@flink.apache.org>
Subject: flink sql支持细粒度的状态配置
 
Hi all
 
在我们生产中发现,如果在sql中配置状态的TTL会导致这个 ttl时间全局生效
 
如果我存在一个如下sql
 
select count(1),region from (select * from A join B on a.uid = b.uid)  group by region
 
如果我配置一个全局的TTL会导致count这个GroupAggFunction的状态被淘汰掉,比如说一天以后累计就被清零
 
如果不配置,又会导致Regular join的状态增大
 
这是其中一个场景,这里只是举一个例子
 
主要是想询问针对 Sql中需要配置局部State的ttl时间,或者同一个任务每个sql配置不同的TTL时间,这种场景应该如何去做 ?
 
 
 
gygzljy@163.com

flink1.10.0-RestClusterClient-cancel job报错

Posted by nicygan <re...@163.com>.
dear all:
      如题,我在调用RestClusterClient#cancel(JobID jobId)方法取消作业时,get不到结果,但作业能正常停止。

用future.get()会报错如下:
Number of retries has been exhausted.

用future.get(10, TimeUnit.SECONDS)会报错timeout.

调用#cancelWithSavepoint(...)和#stopWithSavepoint(...)就没问题,可以正常获取到结果,不报错。

目前发现,
1.10.0有问题,
1.14.0没有上述问题。


作业运行于cdh yarn集群,版本2.6.0
作业部署,per-job

代码如下:
try (ClusterClient<?> clusterClient = new RestClusterClient<>(configuration, clusterId)) {
     clusterClient
          .cancel(jobId)
          .get(20, TimeUnit.SECONDS)
} catch (Exception e) {
 //
}

有谁知道如何解决此问题吗?
非常感谢!

Re: flink sql支持细粒度的状态配置

Posted by Yun Tang <my...@live.com>.
Hi 你好,

我认为这是一个很好的需求,对于data stream以及python API来说,state TTL都是通过API逐个配置的,你的需求就可以直接满足。但是对于SQL来说,由于相同的SQL语句,不同优化器其生成的执行plan可能会差异很大,很难对某个operator内的state进行TTL进行配置,可能一种方式是增加一些SQL的优化hint,对于你示例中的join语句和groupBy 的count语句配以不同的TTL,但是目前Flink SQL尚未支持该功能。


祝好
唐云

________________________________
From: gygzljy@163.com <gy...@163.com>
Sent: Tuesday, December 7, 2021 18:38
To: user-zh <us...@flink.apache.org>
Subject: flink sql支持细粒度的状态配置

Hi all

在我们生产中发现,如果在sql中配置状态的TTL会导致这个 ttl时间全局生效

如果我存在一个如下sql

select count(1),region from (select * from A join B on a.uid = b.uid)  group by region

如果我配置一个全局的TTL会导致count这个GroupAggFunction的状态被淘汰掉,比如说一天以后累计就被清零

如果不配置,又会导致Regular join的状态增大

这是其中一个场景,这里只是举一个例子

主要是想询问针对 Sql中需要配置局部State的ttl时间,或者同一个任务每个sql配置不同的TTL时间,这种场景应该如何去做 ?



gygzljy@163.com