Re: flink sql group by后收集数据问题

Hi！ UDF 支持 ROW 类型，详见 [1] 中关于 ROW 的示例。 [1] https://nightlies.apache.org/flink/flink-docs-master/zh/docs/dev/table/functions/udfs/#%e8%87%aa%e5%8a%a8%e7%b1%bb%e5%9e%8b%e6%8e%a8%e5%af%bc casel.chen 于2021年12月1日周三 上午7:56写道： > 业务中使用flink sql group by操作后想收集所有的数据，如下示例： > > > kafka源表：

Re: flink sql ROW()语句中是不是不能再使用case when？

Hi！ 目前 ROW 的构造不支持内部调用函数，建议先在前面 create view 把需要的值计算好。 casel.chen 于2021年12月1日周三 上午7:59写道： > > > select ROW(field1, field2, case when field3 = 'xxx' then 'T' else 'F' as > field3) from source_table > > > 这样的语句语法检查会通不过。

Re: flink sql太多insert into语句问题

Hi！ 感谢提出问题。方案一应该是最合适的，“算子名称长度超过限制而失败”不是期望行为，具体是什么样的错误栈？ casel.chen 于2021年12月1日周三 上午8:10写道： > 我们有一个场景需要从一张kafka源表写很多不同告警规则到同一张告警目标表。规则数量大概有300多个，采用纯flink sql实现。 > > > 方案一是采用创建视图，将不同规则union all起来，再将视图插入目标表，发现算子都chain成了一个，因为flink > sql算子的名称是flink sql内容，所以算子名称长度超过限制而失败。因而转向方案二 >

flink sql支持state TTL的细粒度局部配置

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的状态增大 这是其中一个场景，这里只是举一个例子 主要是想表达针对

Re: Flink sql jdbc Partitioned Scan timestamp不生效

Hi！ scan.partition.lower-bound 和 scan.partition.upper-bound 都是一个 long 值（而不是一个 timestamp 字符串的形式）。它们将会转换成 where between and 的 SQL 语句通过 jdbc 获取数据。可以检查一下配置项的格式和值的范围是否符合期望。 天下五帝东 于2021年12月1日周三 上午9:23写道： > Hi： > 我在使用flink sql jdbc connector测试partitioned scan功能，发现指定 > scan.partition

Flink sql jdbc Partitioned Scan timestamp不生效

Hi： 我在使用flink sql jdbc connector测试partitioned scan功能，发现指定 scan.partition.column 为timestamp类型时，scan.partition.lower-bound 和scan.partition.upper-bound指定具体的值后，没有读取到相关数据，哪位大佬帮忙解答下 谢谢

flink sql太多insert into语句问题

我们有一个场景需要从一张kafka源表写很多不同告警规则到同一张告警目标表。规则数量大概有300多个，采用纯flink sql实现。 方案一是采用创建视图，将不同规则union all起来，再将视图插入目标表，发现算子都chain成了一个，因为flink sql算子的名称是flink sql内容，所以算子名称长度超过限制而失败。因而转向方案二 方案二是一条规则对应一条insert into语句，生成graph图会发现fan out特别多。这次没有算子名称超长问题，但作业起动会特别慢。考虑到后续规则还会进行修改，添加或删除。这样慢启动无法接受。 想问一下，这种场景最适合的做法是什

flink sql ROW()语句中是不是不能再使用case when？

select ROW(field1, field2, case when field3 = 'xxx' then 'T' else 'F' as field3) from source_table 这样的语句语法检查会通不过。

flink sql group by后收集数据问题

业务中使用flink sql group by操作后想收集所有的数据，如下示例： kafka源表： 班级 学号 姓名 年龄 1 20001张三 15 2 20011李四 16 1 20002王五 16 2 20012吴六 15 create table source_table ( class_no: INT, student_no: INT, name: STRING, age: INT

Re: flink sql建表异常问题

gt; > flink-table-planner-blink_${scala.binary.version} > ${flink.version} > > > org.apache.flink > > flink-table-api-java-bridge_${scala.binary.version} > ${flink.version} > > > org.apache.flink > > flink-table-api-scala-bridge_${scala.binary.version} > ${flink.version} > provided > > > > > 您好 > 我将flink > sql建表程序提交到k8s执行时抛出如下异常，网上提示需要引入flink-table-planner-blink等依赖，但实际pom文件中已经引入，请问这个该如何解决，附部分pom文件，flink版本为1.13。谢谢。 >

flink??sql??hdfs????????????????

flinksql??HDFS:part-c4a19762-bde3-4f37-8b3c-b92d182b450c-task-0-file-0 ??,flink??sql??hdfs ??

回复: flink sql建表异常问题

我将flink sql建表程序提交到k8s执行时抛出如下异常，网上提示需要引入flink-table-planner-blink等依赖，但实际pom文件中已经引入，请问这个该如何解决，附部分pom文件，flink版本为1.13。谢谢。

flink sql建表异常问题

您好 我将flink sql建表程序提交到k8s执行时抛出如下异常，网上提示需要引入flink-table-planner-blink等依赖，但实际pom文件中已经引入，请问这个该如何解决，附部分pom文件，flink版本为1.13。谢谢。

Re:Re: flink sql检查点是否可恢复问题

c）可能会导致报错或非期望行为，比如 offset 越界。 >修改作业配置（在不影响最终 plan 的前提下）也是兼容的（例如你提到的这三个配置都是兼容的），更换 statebackend 和调资源正是 >savepoint 原本需要解决的问题之一。 > >[1] >https://nightlies.apache.org/flink/flink-docs-master/docs/ops/upgrading/#compatibility-table > >casel.chen 于2021年11月13日周六 上午9:01写道： > >>

Re: flink sql检查点是否可恢复问题

://nightlies.apache.org/flink/flink-docs-master/docs/ops/upgrading/#compatibility-table casel.chen 于2021年11月13日周六 上午9:01写道： > 下面这些情况下哪些可以让flink sql作业从上一个检查点恢复? > > 1. 升级了flink引擎版本 > a) 小版本 1.13.1 -> 1.13.2 > b) 大版本 1.13.x -> 1.14.x > 2. 修改了sql，分为2种 >

flink sql检查点是否可恢复问题

下面这些情况下哪些可以让flink sql作业从上一个检查点恢复? 1. 升级了flink引擎版本 a) 小版本 1.13.1 -> 1.13.2 b) 大版本 1.13.x -> 1.14.x 2. 修改了sql，分为2种 a) 表schema发生了变化 b) 表schema没变，只是某些字段取值内容发生变化，例如 从 count(*) 变成 count(*) + 1 3. 修改了with options参数，例如 cache大小和时长 4. 修改了作业配置，例如并行度，资源大小，换了statebackend等

Re: 回复：Re:回复: flink sql消费kafka各分区消息不均衡问题

> > > > > > > > > 在 2021-10-31 10:13:11，"悟空" 写道： > >hi > >那你就需要排查下上有写入方是什么策略写入 或者通过Kafka 查看下 每个分区数据量是否明显不均衡。感觉可能和写入方的原因很大 > > > > > > > >发自我的iPhone > > > > > >------ 原始邮件 -- > >

Flink SQL Join ????????minBatch ????????

Hi : ??Flink 1.12 SQL ??Join ??Kafka DB ?? FOR SYSTEM_TIME AS Temporal Joins ?? ?? ?? In ?? QPS ??

Flink SQL Join 如何使用minBatch 方式查询

Hi : 我现在有一个场景，基于Flink 1.12 SQL 来实现， 想查询下游数据， 大概逻辑就是多张表Join 其中一张是Kafka 表，其他的是DB 表，我基于处理时间 FOR SYSTEM_TIME AS 做了Temporal Joins ， 但是下游数据库查询压力比较大， 现在想通过延迟 批量 In 的方式 减小QPS ，请问如何配置 可以调这个时间长度，我理解默认就是来一条就查询一次 --- Best, WuKong

Flink SQL Join 如何使用minBatch 方式查询

Hi : 我现在有一个场景，基于Flink 1.12 SQL 来实现， 想查询下游数据， 大概逻辑就是多张表Join 其中一张是Kafka 表，其他的是DB 表，我基于处理时间 FOR SYSTEM_TIME AS 做了Temporal Joins ， 但是下游数据库查询压力比较大， 现在想通过延迟 批量 In 的方式 减小QPS ，请问如何配置 可以调这个时间长度，我理解默认就是来一条就查询一次 --- Best, WuKong

回复：Flink Sql读取Hbase表

SQL 中注册 HBase 表 "mytable" CREATETABLEhTable(rowkeyINT,family1ROW,family2ROW,family3ROW,PRIMARYKEY(rowkey)NOTENFORCED)WITH('connector'='hbase-1.4','table-name'='mytable','zookeeper.quorum'='localhost:2181'); Flink sql在读取hbase表时，是一次将数据加载到内存还是每次加载一批数据呀？ 其实就是想知道，如果hbase表数据量特别大的时候，Flink sql是如何处理的？

Flink Sql读取Hbase表

请大佬指导下： -- 在 Flink SQL 中注册 HBase 表 "mytable" CREATETABLEhTable(rowkeyINT,family1ROW,family2ROW,family3ROW,PRIMARYKEY(rowkey)NOTENFORCED)WITH('connector'='hbase-1.4','table-name'='mytable','zookeeper.quorum'='localhost:2181'); Flink sql在读取hbase表时，是一次将数据加载到内存还是每次加载一批数据呀？ 其实就是想知道，如果hbase

Re:回复：Re:回复: flink sql消费kafka各分区消息不均衡问题

casel.chen 发送时间: 2021年10月30日 18:19 >收件人: user-zh 主题: 回复：Re:回复: flink sql消费kafka各分区消息不均衡问题 > > > >kafka是作为flink作业source来消费的，作业跑在k8s上以sessionmode运行，发现有的TM作业特别多，消耗资源也多，而有的TM作业少，占用资源也少。 >会不会是这个原因造成kafka不同分区消费能力差异？那些消费慢的task恰好落在比较忙的TM。有什么办法可以让各个TM负载均衡吗？ > > > > > >

Re: Flink Sql表概念

年10月31日周日 上午10:09写道： > 我的理解是时态表和版本表是一个类似概念 并且都是动态表的一种。最好还是看下flink 官网的解释更权威下 > > > > 发自我的iPhone > > > -- 原始邮件 -- > 发件人: guanyq 发送时间: 2021年10月31日 09:24 > 收件人: user-zh 主题: 回复：Flink Sql表概念 > > > > 请大佬指导下： > 动态表 > 时态表 > 版本表 > 这三个是什么关系？ > flinksql里面还有其他一些表的概念么？

回复：Re:回复: flink sql消费kafka各分区消息不均衡问题

hi 那你就需要排查下上有写入方是什么策略写入 或者通过Kafka 查看下 每个分区数据量是否明显不均衡。感觉可能和写入方的原因很大 发自我的iPhone -- 原始邮件 -- 发件人: casel.chen

回复：Flink Sql表概念

我的理解是时态表和版本表是一个类似概念 并且都是动态表的一种。最好还是看下flink 官网的解释更权威下 发自我的iPhone -- 原始邮件 -- 发件人: guanyq

Flink Sql表概念

请大佬指导下： 动态表 时态表 版本表 这三个是什么关系？ flinksql里面还有其他一些表的概念么？

Re:回复: flink sql消费kafka各分区消息不均衡问题

略应该不会造成数据不均衡，但是无法保证 分区有序性。同时也可以关注下 下游消费者 > 是否会有消费不同分区 处理性能不同问题。 > > > >--- >Best, >WuKong > >发件人： casel.chen >发送时间： 2021-10-29 09:30 >收件人： user-zh@flink.apache.org >主题： flink sql消费kafka各分区消息不均衡问题 >flink >sql消费kafka消息做数据同步，前期没有出现堆积不均的问题，这两天发现某些kafka分区积压特别多

回复: flink sql消费kafka各分区消息不均衡问题

Hi casel.chan： 请问你是sink端数据不均衡还是source端数据不均衡。 如果是写入端 ，看看你是否自定义了分区字段，flink 默认是策略应该不会造成数据不均衡，但是无法保证 分区有序性。同时也可以关注下 下游消费者 是否会有消费不同分区 处理性能不同问题。 --- Best, WuKong 发件人： casel.chen 发送时间： 2021-10-29 09:30 收件人： user-zh@flink.apache.org 主题： flink sql消费kafka各分区消息不均衡问题 flink sql消费kafka消息做数据同步，前期没有出现堆积不

Re: flink sql消费kafka各分区消息不均衡问题

Hi！ 如果没有具体的 SQL 很难分析这个问题。可以通过 Flink UI 观察各节点各并发的处理数据量看是否有部分并发处理量比较大；另外可以观察每个节点的反压情况，看是否有部分并发反压严重。另外可以特别注意 hash 边，看 hash key 是否有倾斜（这会表现在下游节点不同并发之间处理量差异比较大）。 casel.chen 于2021年10月29日周五 上午9:31写道： > flink > sql消费kafka消息做数据同步，前期没有出现堆积不均的问题，这两天发现某些kafka分区积压特别多，会是什么原因造成的？怎样解决呢？从统计结果上看，消息还算均匀地打到各

flink sql消费kafka各分区消息不均衡问题

flink sql消费kafka消息做数据同步，前期没有出现堆积不均的问题，这两天发现某些kafka分区积压特别多，会是什么原因造成的？怎样解决呢？从统计结果上看，消息还算均匀地打到各个kafka分区上。作业没有开窗和聚合，只是攒一批写一批这样子的。注：作业是跑在k8s上的 | 分区 ID | 客户端 | 最大位点 | 消费位点 | 堆积量 | | 0 | n/a | 155,397,108 | 155,396,747 | 361 | | 1 | n/a | 155,215,444 | 155,215,108 | 336 | | 2 | n/a | 155,369,596

Re: Re: Flink SQL 1.12 批量数据导入，如果加速性能

', 'url' = 'jdbc:mysql://x/?useUnicode=true=UTF-8', 'table-name' = 'table-name', 'username' = 'xxx', 'password' = 'xxx', 'driver' = 'com.mysql.jdbc.Driver' ); --- Best, WuKong 发件人： Caizhi Weng 发送时间： 2021-10-26 12:43 收件人： flink中文邮件组 主题： Re: Flink SQL 1.12 批量数据导入，如果加速性能 Hi！ 我通

Re: Flink SQL 1.12 批量数据导入，如果加速性能

Hi！ 我通过 Flink SQL 无论怎么加大并行度， 都是单并行度导入 你是如何加大并行度的？除 source 外其他节点也是单并行度吗，还是说可以成功加大？能否分享一下你的 SQL 便于解答？ WuKong 于2021年10月26日周二 上午11:36写道： > Hi： > 我目前有个场景， 需要通过基于Flink SQL 进行历史数据导入， 比如Source 端是一张MYSQL 表， Sink端 > 也是一张MSYQL 表， 我通过Flink SQL 无论怎么加大并行度， 都是单并行度导入，速率很慢， 请问有什么需要配置的吗？ 或者其他解决

Flink SQL 1.12 批量数据导入，如果加速性能

Hi： 我目前有个场景， 需要通过基于Flink SQL 进行历史数据导入， 比如Source 端是一张MYSQL 表， Sink端 也是一张MSYQL 表， 我通过Flink SQL 无论怎么加大并行度， 都是单并行度导入，速率很慢， 请问有什么需要配置的吗？ 或者其他解决方案 可以基于SQL进行大批量数据导入，数十亿量级。 --- Best, WuKong

Re: Flink SQL支持side output

举个例子 Kenyore Woo 于2021年10月14日周四 上午10:37写道： > > 你可以把使用反向条件把脏数据输出到另外一张表去。source会复用的。其实和side output效果是一致的 > On Oct 13, 2021 at 16:28:57, Ada Luna wrote: > > > 这个没有支持的打算是因为，目前我们假定Flink SQL处理的数据都是干净的经过清洗的是吧。 > > > > Ada Luna 于2021年9月19日周日 下午7:43写道： > > > >

Re: Flink SQL支持side output

你可以把使用反向条件把脏数据输出到另外一张表去。source会复用的。其实和side output效果是一致的 On Oct 13, 2021 at 16:28:57, Ada Luna wrote: > 这个没有支持的打算是因为，目前我们假定Flink SQL处理的数据都是干净的经过清洗的是吧。 > > Ada Luna 于2021年9月19日周日 下午7:43写道： > > > 主要是脏数据，Source、Sink或者其他算子产生的脏数据，向把这些数据侧向输出到外部数据库里存起来。 > > > Caizhi W

Re: Flink SQL支持side output

这个没有支持的打算是因为，目前我们假定Flink SQL处理的数据都是干净的经过清洗的是吧。 Ada Luna 于2021年9月19日周日 下午7:43写道： > > 主要是脏数据，Source、Sink或者其他算子产生的脏数据，向把这些数据侧向输出到外部数据库里存起来。 > > Caizhi Weng 于2021年9月16日周四 下午1:52写道： > > > > Hi！ > > > > 就我所知目前暂时没有支持 side output 的打算。可以描述一下需求和场景吗？ >

Re: Does the flink sql support checkpoints

the counting result? It might be that you're storing the results in a sink which only updates the result after a successful checkpoint. 王小宅的蜗居生活 于2021年10月12日周二 上午9:32写道： > The flink version is v1.13.2 > > 王小宅的蜗居生活 于2021年10月11日周一 下午9:01写道： > > > Use flink sql for rea

Re: Does the flink sql support checkpoints

The flink version is v1.13.2 王小宅的蜗居生活 于2021年10月11日周一 下午9:01写道： > Use flink sql for real-time calculation (deployment mode: on yarn). To use > the checkpoint, you need to configure the following in the flink-conf.yaml: > > > state.backend: filesystem > > state.checkpoin

Does the flink sql support checkpoints

Use flink sql for real-time calculation (deployment mode: on yarn). To use the checkpoint, you need to configure the following in the flink-conf.yaml: state.backend: filesystem state.checkpoints.dir: hdfs:///flink/flink-checkpoints state.savepoints.dir: hdfs:///flink/flink-savepoints

flink sql设置不同算子并行度问题

flink sql可以为不同算子设置不同并行度吗？可以的话，要如何实现呢？ 新发布的Flink 1.14.0支持细粒度资源管控这一特性可以应用到flink sql上吗？谢谢！

Re: flink sql计算新增用户

Hi！ 听起来 event time temporal join 符合你的需求。详见 https://ci.apache.org/projects/flink/flink-docs-release-1.13/zh/docs/dev/table/sql/queries/joins/#event-time-temporal-join z 于2021年9月28日周二 下午8:27写道： > hi各位，我想计算每日新增用户的数量，用户的登录日志在kafka中，在当日之前登录过的用户即为老用户，目前我的做法是将用户登录信息用flink > sql写到下游mysql表A中，表中存储

flink sql????????????

hi??kafkaflink sqlmysql??Aid??tsjoin

????: ?????? flink sql????????????????sink table?

hi : UDF ??SQL ?? ?? topic ?? ?? https://mp.weixin.qq.com/s/IKzCRTh8eamk6TX7_RHYdQ wukon...@foxmail.com JasonLee ?? 2021-09-23 21:56 user-zh@flink.apache.org ?? ?? flink sql

Re: Re:re:Re: 回复：Flink SQL官方何时能支持redis和mongodb连接器？

Hi 用github账号登陆之后，可以使用添加package的方式[1]自行上传共享。 [1] https://flink-packages.org/new-package 祝好 唐云 From: casel.chen Sent: Thursday, September 23, 2021 12:40 To: user-zh@flink.apache.org Cc: myasuka Subject: Re:re:Re: 回复：Flink SQL官方何时能支持redis和mongodb连接器？ >我司基于最新

?????? flink sql????????????????sink table?

ike casel.chennbsp; ??2021??9??18?? 8:27?? kafka topic??topic) ??????flink sqlsink table

?????? flink sql????????????????sink table?

pic??topic) ??????flink sqlsink table

Re: Flink SQL是否支持Count Window函数？

(summedStream); > tEnv.registerTable("S", summedTable); > tEnv.executeSql("SELECT f0, f1 FROM S").print(); > > casel.chen 于2021年9月17日周五 下午6:05写道： > > > 今天遇到一个业务场景用到count window，查了下Flink官网目前Flink SQL只支持time > > window，问一下官方是否打算sql支持count window呢？ > > 如果没有计划的话，自己要如何实现？是否可以像flink 1.13引入的cumulate window写一个自定义窗口函数呢？谢谢！ >

Re:re:Re: 回复：Flink SQL官方何时能支持redis和mongodb连接器？

5 >收件人: "user-zh@flink.apache.org"; >主题: Re: 回复：Flink SQL官方何时能支持redis和mongodb连接器？ > > > >Hi, > >其实目前Flink社区并不是那么欢迎新增官方支持的connector，主要原因就是社区的开发人员有限，没有精力维护太多的connector，尤其是一些connector的实现需要一定的相关背景，但很难保证review代码的开发人员具有相关背景，毕竟大家都需要为自己approve的代码负责。 >你可以在 flink-packages [1

re:Re: 回复：Flink SQL官方何时能支持redis和mongodb连接器？

我司基于最新提供流批一体的接口，实现了mongodb的连接器，支持source和sink，实现了控制批量插入频率、控制缓存批的数据量和mongo文档对象和java对象转换，同时还可选择批量更新，并且使用的mongo最新异步驱动，后期我还会不断优化性能，看大佬能否推动一下，把这个连接器贡献给社区 -- 原始邮件 -- 发件人: "Yun Tang"; 发件时间: 2021-09-22 10:55 收件人: "user-zh@flink.apache.org"; 主题: Re: 回复：F

??????re:?????? flink sql????????????????sink table?

ocs-release-1.13/zh/docs/dev/table/sql/create/#like casel.chen ??2021??9??18?? 8:27?? kafka topic??topic) ??????flink sqlsink table

??????re:?????? flink sql????????????????sink table?

sql??sql iPhone -- -- ??: 2572805166 <2572805...@qq.com.INVALID : 2021923 09:23 ??: user-zh

re:回复： flink sql是否支持动态创建sink table?

使用java的动态编译和类加载技术，实现类似于web项目的热加载 -- 原始邮件 -- 发件人: "JasonLee"<17610775...@163.com; 发件时间: 2021-09-22 22:33 收件人: "user-zh@flink.apache.org"; 主题: 回复： flink sql是否支持动态创建sink table? hi 事实上这个跟构建 graph 没有太大的关系 也不用在构建后调整 在构造 prod

回复： flink sql是否支持动态创建sink table?

elease-1.13/zh/docs/dev/table/sql/create/#like casel.chen 于2021年9月18日周六 上午8:27写道： 上游kafka topic消息带有一个用户类型字段，现在想根据不同用户类型将数据发到不同topic（为了扩展不想写死有哪些类型) ，请问flink sql支持动态创建sink table吗？

回复： flink sql是否支持动态创建sink table?

topic 都定义 DDL。 如果是因为各 topic 之间的 schema 重复度比较高，只有些许字段以及 topic 名称等不同，可以看一下 DDL LIKE 语法： https://ci.apache.org/projects/flink/flink-docs-release-1.13/zh/docs/dev/table/sql/create/#like casel.chen 于2021年9月18日周六 上午8:27写道： 上游kafka topic消息带有一个用户类型字段，现在想根据不同用户类型将数据发到不同topic（为了扩展不想写死有哪些类型) ，请问flink sql支持动态创建

回复： flink sql是否支持动态创建sink table?

://ci.apache.org/projects/flink/flink-docs-release-1.13/zh/docs/dev/table/sql/create/#like casel.chen 于2021年9月18日周六 上午8:27写道： 上游kafka topic消息带有一个用户类型字段，现在想根据不同用户类型将数据发到不同topic（为了扩展不想写死有哪些类型) ，请问flink sql支持动态创建sink table吗？

Re: flink sql是否支持动态创建sink table?

户类型将数据发到不同topic（为了扩展不想写死有哪些类型) ，请问flink > sql支持动态创建sink table吗？

Re: Flink SQL是否支持Count Window函数？

rTable("S", summedTable); tEnv.executeSql("SELECT f0, f1 FROM S").print(); casel.chen 于2021年9月17日周五 下午6:05写道： > 今天遇到一个业务场景用到count window，查了下Flink官网目前Flink SQL只支持time > window，问一下官方是否打算sql支持count window呢？ > 如果没有计划的话，自己要如何实现？是否可以像flink 1.13引入的cumulate window写一个自定义窗口函数呢？谢谢！

Re: 回复：Flink SQL官方何时能支持redis和mongodb连接器？

Sent: Saturday, September 18, 2021 17:17 To: user-zh@flink.apache.org Subject: 回复：Flink SQL官方何时能支持redis和mongodb连接器？ 这个可以自已定义一个，参考源码写一个，自己写出来的才是自己的，直接用别人的还是别人的 --原始邮件-- 发件人

Re: Flink SQL支持side output

主要是脏数据，Source、Sink或者其他算子产生的脏数据，向把这些数据侧向输出到外部数据库里存起来。 Caizhi Weng 于2021年9月16日周四 下午1:52写道： > > Hi！ > > 就我所知目前暂时没有支持 side output 的打算。可以描述一下需求和场景吗？ > > Ada Luna 于2021年9月15日周三 下午8:38写道： > > > Flink SQL 未来会支持side output，侧向输出一些脏数据吗？ > >

??????Flink SQL??????????????redis??mongodb????????

---- ??: "user-zh"

flink sql是否支持动态创建sink table?

上游kafka topic消息带有一个用户类型字段，现在想根据不同用户类型将数据发到不同topic（为了扩展不想写死有哪些类型) ，请问flink sql支持动态创建sink table吗？

Flink SQL是否支持Count Window函数？

今天遇到一个业务场景用到count window，查了下Flink官网目前Flink SQL只支持time window，问一下官方是否打算sql支持count window呢？ 如果没有计划的话，自己要如何实现？是否可以像flink 1.13引入的cumulate window写一个自定义窗口函数呢？谢谢！

Flink SQL官方何时能支持redis和mongodb连接器？

redis和mongodb经常在工作中用到，但Flink官方一直没有提供这两个标准连接器，想问一下什么时候能正式release方便大家使用呢？ ps: behair库已经很久没更新了，对应的flink版本太低。

Re: Flink SQL支持side output

Hi！ 就我所知目前暂时没有支持 side output 的打算。可以描述一下需求和场景吗？ Ada Luna 于2021年9月15日周三 下午8:38写道： > Flink SQL 未来会支持side output，侧向输出一些脏数据吗？ >

Flink SQL支持side output

Flink SQL 未来会支持side output，侧向输出一些脏数据吗？

Re:Re: flink sql streaming情况如何解决数据倾斜问题

按 key1 key2 分开统计，然后合并一下 在 2021-09-08 11:04:59，"yidan zhao" 写道： >我们流量大概4w的qps，如何根据key1+key2进行pv统计（SQL任务比较简单）。 > >但是key2的分布比较极端，有些可能90%集中的。 > >Shuo Cheng 于2021年9月7日周二 下午7:30写道： > >> 最好具体描述下什么场景的倾斜, sql 上也有一些解倾斜的手段 >> >> On 9/7/21, yidan zhao wrote: >> > 如题，目前非sql情况本身实现灵活，有很多方案。 >> >

Re: flink sql streaming情况如何解决数据倾斜问题

可以参考下针对 AGG 的调优指南 https://ci.apache.org/projects/flink/flink-docs-master/docs/dev/table/tuning/ On Wed, Sep 8, 2021 at 11:05 AM yidan zhao wrote: > 我们流量大概4w的qps，如何根据key1+key2进行pv统计（SQL任务比较简单）。 > > 但是key2的分布比较极端，有些可能90%集中的。 > > Shuo Cheng 于2021年9月7日周二 下午7:30写道： > > > 最好具体描述下什么场景的倾斜, sql

?????? flink sql streaming????????????????????????

?? ---- ??: "user-zh"

Re: flink sql streaming情况如何解决数据倾斜问题

我stream方式倒是简单，我会基于统计结果，即key => pv的情况，根据pv大小记录key，pv/1000 进入kafka topic（skew）。 然后任务会读取 skew 数据，根据这个数据将key进行拆分，然后做2层汇总统计。就是做了个动态倾斜key的发现和分拆。 yidan zhao 于2021年9月8日周三 上午11:04写道： > 我们流量大概4w的qps，如何根据key1+key2进行pv统计（SQL任务比较简单）。 > > 但是key2的分布比较极端，有些可能90%集中的。 > > Shuo Cheng 于2021年9月7日周二 下午7:30写道： > >>

Re: flink sql streaming情况如何解决数据倾斜问题

我们流量大概4w的qps，如何根据key1+key2进行pv统计（SQL任务比较简单）。 但是key2的分布比较极端，有些可能90%集中的。 Shuo Cheng 于2021年9月7日周二 下午7:30写道： > 最好具体描述下什么场景的倾斜, sql 上也有一些解倾斜的手段 > > On 9/7/21, yidan zhao wrote: > > 如题，目前非sql情况本身实现灵活，有很多方案。 > > 但是SQL情况下，倾斜严重，同时无解。有没有小伙伴解决过类似问题。 > > > > 注意：sql，流任务，数据倾斜。 > > >

Re: flink sql streaming情况如何解决数据倾斜问题

最好具体描述下什么场景的倾斜, sql 上也有一些解倾斜的手段 On 9/7/21, yidan zhao wrote: > 如题，目前非sql情况本身实现灵活，有很多方案。 > 但是SQL情况下，倾斜严重，同时无解。有没有小伙伴解决过类似问题。 > > 注意：sql，流任务，数据倾斜。 >

flink sql streaming情况如何解决数据倾斜问题

如题，目前非sql情况本身实现灵活，有很多方案。 但是SQL情况下，倾斜严重，同时无解。有没有小伙伴解决过类似问题。 注意：sql，流任务，数据倾斜。

回复: Re: 基于flink 1.12 如何通过Flink SQL 实现 多sink端 在同一个事务中

: Re: 基于flink 1.12 如何通过Flink SQL 实现 多sink端 在同一个事务中 对于Exactly-Once模式下的checkpoint，如果sink端支持两段式事务，应该就可以做到一个sink失败，整个checkpoint失败的。 不过JDBC sink支持Exactly-Once是1.13.0以后的事情了，建议检查一下你的版本和配置 在 2021-08-30 16:27:24，"wukon...@foxmail.com" 写道： >Hi: > 我理解这种方式， 目前我只是想让先

Re:Re: Re: 基于flink 1.12 如何通过Flink SQL 实现 多sink端 在同一个事务中

一逻辑 > > > >wukon...@foxmail.com > >发件人： Shuo Cheng >发送时间： 2021-08-30 10:19 >收件人： user-zh >主题： Re: 基于flink 1.12 如何通过Flink SQL 实现 多sink端 在同一个事务中 >你好, 你说的这种控制写入的方式在同一个 Flink SQL job 里是无法实现的. 控制数据是否写入某个 >Sink，可以看看是否从逻辑上能在 Sink 前加一个 Filter，从而达到过滤目的；如果 kafka sink 跟 MySQL >表是

Re: Re: 基于flink 1.12 如何通过Flink SQL 实现 多sink端 在同一个事务中

Hi: 我理解这种方式， 目前我只是想让先插入数据到Mysql 然后再通过通知到下游，来查询Mysql 进行数据etl 不知道大家如何通过SQL来实现这一逻辑 wukon...@foxmail.com 发件人： Shuo Cheng 发送时间： 2021-08-30 10:19 收件人： user-zh 主题： Re: 基于flink 1.12 如何通过Flink SQL 实现 多sink端 在同一个事务中 你好, 你说的这种控制写入的方式在同一个 Flink SQL job 里是无法实现的. 控制数据是否写入某个 Sink，可以看看是否从逻辑上能在 Sink 前加一个

Re: Re: 基于flink 1.12 如何通过Flink SQL 实现 多sink端 在同一个事务中

能具体说下如何实现吗？ 我用cdc 能实现什么，我现在想让两个Insert Sql 保持到一个事务里， 要么全成功，要么全失败，目前查看Flink 文档 并没有发现相关的解释 wukon...@foxmail.com 发件人： jie han 发送时间： 2021-08-26 21:36 收件人： user-zh 主题： Re: 基于flink 1.12 如何通过Flink SQL 实现 多sink端 在同一个事务中 HI： 可以尝试下使用flink cdc 的方式写入到第二个kafka里面呀 悟空 于2021年8月26日周四 下午1:54写道： > 我目前

Re: 基于flink 1.12 如何通过Flink SQL 实现 多sink端 在同一个事务中

你好, 你说的这种控制写入的方式在同一个 Flink SQL job 里是无法实现的. 控制数据是否写入某个 Sink，可以看看是否从逻辑上能在 Sink 前加一个 Filter，从而达到过滤目的；如果 kafka sink 跟 MySQL 表是一种类似级联的关系, 可以考虑先写入 MySQL, 然后另起一个 Job 用 CDC 方式读 MySQL changelog 再写入 Kafka sink. On 8/26/21, jie han wrote: > HI： > 可以尝试下使用flink cdc 的方式写入到第二个kafka里面呀 > > 悟空 于202

Flink sql CLI parsing avro format data error 解析avro数据失败

i have Apache Avro schema 我的avro schema 如下 { "type" : "record", "name" : "KafkaAvroMessage", "namespace" : "xxx", "fields" : [ { "name" : "aaa", "type" : "string" }, { "name" : "bbb", "type" : [ "null", "string" ], "default" : null },{ "name" : "ccc",

Re: 基于flink 1.12 如何通过Flink SQL 实现 多sink端 在同一个事务中

"user-zh" > < > tsreape...@gmail.com; > 发送时间:2021年8月26日(星期四) 中午1:25 > 收件人:"user-zh" > 主题:Re: 基于flink 1.12 如何通过Flink SQL 实现 多sink端 在同一个事务中 > > > > Hi！ > > 如果 statement set 已经包含了多个 insert 语句，那么写入 kafka 表和写入 db 应该是在同一个

?????? ????flink 1.12 ????????Flink SQL ???? ??sink?? ??????????????

flink-connector-kafka_2.11??flink-connector-jdbc_2.11?? ??mysql ?? ?? ?? kafka??java.sql.BatchUpdateException ??3 sink Kafka ??Kafka?? 'sink.semantic' = 'exactly-once',

Flink sql CLI parsing avro format data error 解析avro数据失败

i have Apache Avro schema 我的avro schema 如下 { "type" : "record", "name" : "KafkaAvroMessage", "namespace" : "xxx", "fields" : [ { "name" : "aaa", "type" : "string" }, { "name" : "bbb", "type" : [ "null", "string" ], "default" : null },{ "name" : "ccc",

Flink sql CLI parsing avro format data error 解析avro数据失败

i have Apache Avro schema 我的avro schema 如下 { "type" : "record", "name" : "KafkaAvroMessage", "namespace" : "xxx", "fields" : [ { "name" : "aaa", "type" : "string" }, { "name" : "bbb", "type" : [ "null", "string" ], "default" : null },{ "name" : "ccc",

Re: 基于flink 1.12 如何通过Flink SQL 实现 多sink端 在同一个事务中

"user-zh" > < > fskm...@gmail.com; > 发送时间:2021年8月26日(星期四) 中午12:36 > 收件人:"user-zh" > 主题:Re: 基于flink 1.12 如何通过Flink SQL 实现 多sink端 在同一个事务中 > > > > 说的是 statement set [1] 吗 ？ > > [1] > > https://ci.apache.org/projects

?????? ????flink 1.12 ????????Flink SQL ???? ??sink?? ??????????????

statement set[1] StatementSet.addInsertSql ??sql execute() ---- ??:

Re: 基于flink 1.12 如何通过Flink SQL 实现 多sink端 在同一个事务中

果写入db失败，则不写入kafka 。 > 语句类似这种： > insert into db_table_sinkselect * from > kafka_source_table; > insert into kafka_table_sink select * from kafka_source_table; > > > 请问flink SQL 有实现方式吗？ 目前经过测试，发现如果db写入失败，但是kafka依旧写入，同时flink 程序没有挂掉。

????flink 1.12 ????????Flink SQL ???? ??sink?? ??????????????

into kafka_table_sink select * from kafka_source_table; flink SQL ?? ??db??kafka??flink ??

Flink-SQL 消费Confluent Kafka 反序列化失败

同行们好， 最近遇到一个使用FlinkSQL无法连接（消费）Confluent Kafka的问题： Fink 版本：1.12-csadh1.3.0.0 集群平台：Cloudera（CDP） Kafka：Confluent Kafka 现象：使用如下Flink - SQL 连接（消费）kafka数据（反序列化）失败： SQL语句如下： [cid:a8372b16-f6a7-48c0-b19c-d85bf38c1163] 错误日志如下： java.io.IOException: Could not find schema with id 79 in registry

Re: Flink SQL Api不支持TIMESTAMP(p) WITH TIME ZONE 类型的列

你好 感谢解答，我仔细看了下Flink Table API & SQL Data Types页面： https://ci.apache.org/projects/flink/flink-docs-release-1.13/docs/dev/table/types/ ，

Re: Flink SQL Api不支持TIMESTAMP(p) WITH TIME ZONE 类型的列

Hello, Flink 还不支持 TIMESTAMP WITH TIME ZONE 类型， 目前支持的有： TIMESTAMP WITHOUT TIME ZONE, 缩写为 TIMESTAMP TIMESTAMP WITH LOCAL TIME ZONE，缩写为TIMESTAMP_LTZ 祝好， Leonard > 在 2021年8月19日，20:51，changfeng 写道： > > ` TIMESTAMP(6) WITH TIME ZONE

Re: Flink SQL Api不支持TIMESTAMP(p) WITH TIME ZONE 类型的列

; org.apache.flink.sql.parser.impl.FlinkSqlParserImpl.SqlStmtEof(FlinkSqlParserImpl.java:3918) > at > org.apache.flink.sql.parser.impl.FlinkSqlParserImpl.parseSqlStmtEof(FlinkSqlParserImpl.java:261) > at > org.apache.calcite.sql.parser.SqlParser.parseQuery(SqlParser.java:153) > ... 32 more > > 而使用SQL: CREATE TABLE source ( `ctimestamp` TIMESTAMP(6) WITH LOCAL TIME > ZONE) WITH ('connector' = 'print’) 创建表则不会有问题，请问Flink SQL > Api当前是否支持TIMESTAMP(p) WITH TIME ZONE 类型数据或者有相关Bug。 > >

Flink SQL Api不支持TIMESTAMP(p) WITH TIME ZONE 类型的列

seSqlStmtEof(FlinkSqlParserImpl.java:261) at org.apache.calcite.sql.parser.SqlParser.parseQuery(SqlParser.java:153) ... 32 more 而使用SQL: CREATE TABLE source ( `ctimestamp` TIMESTAMP(6) WITH LOCAL TIME ZONE) WITH ('connector' = 'print’) 创建表则不会有问题，请问Flink SQL Api当前是否支持TIMESTAMP(p) WITH TIME ZONE 类型数据或者有相关Bug。

Re:Re: Re: 关于flink sql自定义udf，eval方法被调用两次问题

感谢提供帮助 在 2021-08-16 11:31:11，"Qishang" 写道： >Hi >之前社区发过一个 JD 的解决方案，可以参考下[1]。 > >[1]: https://mp.weixin.qq.com/s/YluIj3vmebFmZjRbymKBZw > > > >黄志高 于2021年8月16日周一 上午11:04写道： > >> == Physical Execution Plan == >> >> Stage 1 : Data Source >> >> content : Source:

关于 Flink SQL 上的 State Schema Evolution

Hi 社区的各位： 来信之前做了一些测试，当前 Flink 目前不能保证一个修改过的 SQL 能够从之前的 checkpoint / savepoint 进行恢复。 目前我们遇到的场景是用户会要想在原有的 SQL 上加聚合指标。在 1.12.2 下如果 Flink SQL 新增一个聚合指标，org.apache.flink.table.runtime.typeutils.RowDataSerializer.RowDataSerializerSnapshot#resolveSchemaCompatibility 会返回 INCOMPATIBLE，然后抛出 目前了解下来字节做了一些相关工

Re: Re: 关于flink sql自定义udf，eval方法被调用两次问题

Hi 之前社区发过一个 JD 的解决方案，可以参考下[1]。 [1]: https://mp.weixin.qq.com/s/YluIj3vmebFmZjRbymKBZw 黄志高 于2021年8月16日周一 上午11:04写道： > == Physical Execution Plan == > > Stage 1 : Data Source > > content : Source: TableSourceScan(table=[[default_catalog, > default_database, test_kafka]], fields=[tz]) > > > > >

Re:Re: 关于flink sql自定义udf，eval方法被调用两次问题

== Physical Execution Plan == Stage 1 : Data Source content : Source: TableSourceScan(table=[[default_catalog, default_database, test_kafka]], fields=[tz]) Stage 2 : Operator content : Calc(select=[tt1(tz) AS tz], where=[tt1(tz) IS NOT NULL]) ship_strategy : FORWARD

Re: 关于flink sql自定义udf，eval方法被调用两次问题

Hi. 应该是这个问题，https://issues.apache.org/jira/browse/FLINK-21573 打印一下执行计划和code gen 黄志高 于2021年8月15日周日 下午10:06写道： > hi all, > 各位大佬，有没有遇到过eval方法被调用两次问题，在我使用select * from (select tt1(tz) from > test_kafka) as t where tz is not null > 时，在eval方法进行入参打印时，发现eval方法被调用了两次，当我使用select * from (select

关于flink sql自定义udf，eval方法被调用两次问题

hi all, 各位大佬，有没有遇到过eval方法被调用两次问题，在我使用select * from (select tt1(tz) from test_kafka) as t where tz is not null 时，在eval方法进行入参打印时，发现eval方法被调用了两次，当我使用select * from (select tt1(tz) from test_kafka) as t这个sql时，不进行where tz is not null 操作，eval方法此时只会调用一次，如果将where tz is not null 改成 where tz

Re: flink sql job 提交流程问题

补充： 这个问题在ha的情况下非常突出，因为和hdfs的交互式线性的，当文件达到几百的时候，特别慢 Peihui He 于2021年8月15日周日 上午11:18写道： > Hi all: > > 在使用zeppelin提交sql的过程中总是发现超时现象，通过定位发现有如下问题： > 1、blob client 和blob server 通信时采用单客户端通行，当有比较多的文件时，比如100个，这个耗时挺大的 > > 2、blob server 虽然有blob.fetch.num-concurrent 进行并发控制，但是blob >

flink sql job 提交流程问题

Hi all: 在使用zeppelin提交sql的过程中总是发现超时现象，通过定位发现有如下问题： 1、blob client 和blob server 通信时采用单客户端通行，当有比较多的文件时，比如100个，这个耗时挺大的 2、blob server 虽然有blob.fetch.num-concurrent 进行并发控制，但是blob server在moveTempFileToStore方法中采用了写锁的方式，当时并发失效。 通过本地测试，简单的调整了代码，示例如下： BlobServer: [image: image.png] ClientUtils [image:

<    1   2   3   4   5   6   7   8   9   10   >